Merge pull request #458 from aramase/load

feat: add filtered watch for reconcile

Author: k8s-ci-robot

cmd/secrets-store-csi-driver/main.go

@@ -25,18 +25,19 @@ import (
 	"strings"
 	"time"
 
+	"sigs.k8s.io/secrets-store-csi-driver/pkg/cache"
 	"sigs.k8s.io/secrets-store-csi-driver/pkg/metrics"
 	"sigs.k8s.io/secrets-store-csi-driver/pkg/rotation"
 	"sigs.k8s.io/secrets-store-csi-driver/pkg/version"
 
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	json "k8s.io/component-base/logs/json"
 	"k8s.io/klog/v2"
 
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/client/config"
 
 	"sigs.k8s.io/secrets-store-csi-driver/apis/v1alpha1"
 	"sigs.k8s.io/secrets-store-csi-driver/controllers"
@@ -52,7 +53,6 @@ var (
 	logFormatJSON      = flag.Bool("log-format-json", false, "set log formatter to json")
 	providerVolumePath = flag.String("provider-volume", "/etc/kubernetes/secrets-store-csi-providers", "Volume path for provider")
 	// this will be removed in a future release
-	_           = flag.String("min-provider-version", "", "[DEPRECATED] set minimum supported provider versions with current driver")
 	metricsAddr = flag.String("metrics-addr", ":8095", "The address the metric endpoint binds to")
 	// grpcSupportedProviders is a ; separated string that can contain a list of providers. The reason it's a string is to allow scenarios
 	// where the driver is being used with 2 providers, one which supports grpc and other using binary for provider.
@@ -62,6 +62,11 @@ var (
 	enableProfile          = flag.Bool("enable-pprof", false, "enable pprof profiling")
 	profilePort            = flag.Int("pprof-port", 6065, "port for pprof profiling")
 
+	// enable filtered watch for NodePublishSecretRef secrets. The filtering is done on the csi driver label: secrets-store.csi.k8s.io/used=true
+	// For Kubernetes secrets used to provide credentials for use with the CSI driver, set the label by running: kubectl label secret secrets-store-creds secrets-store.csi.k8s.io/used=true
+	// This feature flag will be enabled by default after n+2 releases giving time for users to label all their existing credential secrets.
+	filteredWatchSecret = flag.Bool("filtered-watch-secret", false, "enable filtered watch for NodePublishSecretRef secrets with label secrets-store.csi.k8s.io/used=true")
+
 	scheme = runtime.NewScheme()
 )
 
@@ -90,6 +95,9 @@ func main() {
 			klog.ErrorS(http.ListenAndServe(addr, nil), "unable to start profiling server")
 		}()
 	}
+	if *filteredWatchSecret {
+		klog.Infof("Filtered watch for nodePublishSecretRef secret based on secrets-store.csi.k8s.io/used=true label enabled")
+	}
 
 	// initialize metrics exporter before creating measurements
 	err := metrics.InitMetricsExporter()
@@ -100,10 +108,28 @@ func main() {
 	cfg := ctrl.GetConfigOrDie()
 	cfg.UserAgent = version.GetUserAgent("controller")
 
+	// this enables filtered watch of pods based on the node name
+	// only pods running on the same node as the csi driver will be cached
+	fieldSelectorByResource := map[schema.GroupResource]string{
+		{Group: "", Resource: "pods"}: fields.OneTermEqualSelector("spec.nodeName", *nodeID).String(),
+	}
+	labelSelectorByResource := map[schema.GroupResource]string{
+		// this enables filtered watch of secretproviderclasspodstatuses based on the internal node label
+		// internal.secrets-store.csi.k8s.io/node-name=<node name> added by csi driver
+		{Group: v1alpha1.GroupVersion.Group, Resource: "secretproviderclasspodstatuses"}: fmt.Sprintf("%s=%s", v1alpha1.InternalNodeLabel, *nodeID),
+		// this enables filtered watch of secrets based on the label (secrets-store.csi.k8s.io/managed=true)
+		// added to the secrets created by the CSI driver
+		{Group: "", Resource: "secrets"}: fmt.Sprintf("%s=true", controllers.SecretManagedLabel),
+	}
+
 	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
 		Scheme:             scheme,
 		MetricsBindAddress: *metricsAddr,
 		LeaderElection:     false,
+		NewCache: cache.Builder(cache.Options{
+			FieldSelectorByResource: fieldSelectorByResource,
+			LabelSelectorByResource: labelSelectorByResource,
+		}),
 	})
 	if err != nil {
 		klog.Fatalf("failed to start manager, error: %+v", err)
@@ -152,23 +178,15 @@ func main() {
 	}()
 
 	if *enableSecretRotation {
-		rec, err := rotation.NewReconciler(scheme, *providerVolumePath, *nodeID, *rotationPollInterval, providerClients)
+		rec, err := rotation.NewReconciler(scheme, *providerVolumePath, *nodeID, *rotationPollInterval, providerClients, *filteredWatchSecret)
 		if err != nil {
 			klog.Fatalf("failed to initialize rotation reconciler, error: %+v", err)
 		}
 		go rec.Run(ctx.Done())
 	}
 
-	ccfg, err := config.GetConfig()
-	if err != nil {
-		klog.Fatalf("failed to initialize driver, error getting config: %+v", err)
-	}
-	c, err := client.New(ccfg, client.Options{Scheme: scheme, Mapper: nil})
-	if err != nil {
-		klog.Fatalf("failed to initialize driver, error creating client: %+v", err)
-	}
 	driver := secretsstore.GetDriver()
-	driver.Run(ctx, *driverName, *nodeID, *endpoint, *providerVolumePath, providerClients, c)
+	driver.Run(ctx, *driverName, *nodeID, *endpoint, *providerVolumePath, providerClients, mgr.GetClient())
 }
 
 // withShutdownSignal returns a copy of the parent context that will close if

controllers/secretproviderclasspodstatus_controller.go

@@ -24,6 +24,9 @@ import (
 	"time"
 
 	"k8s.io/client-go/kubernetes"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
@@ -38,7 +41,6 @@ import (
 
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
 	corev1 "k8s.io/api/core/v1"
 	v1 "k8s.io/api/core/v1"
@@ -52,7 +54,8 @@ import (
 )
 
 const (
-	secretManagedLabel         = "secrets-store.csi.k8s.io/managed"
+	SecretManagedLabel         = "secrets-store.csi.k8s.io/managed"
+	SecretUsedLabel            = "secrets-store.csi.k8s.io/used"
 	secretCreationFailedReason = "FailedToCreateSecret"
 )
 
@@ -108,7 +111,7 @@ func (r *SecretProviderClassPodStatusReconciler) Patcher(ctx context.Context) er
 
 	spcPodStatusList := &v1alpha1.SecretProviderClassPodStatusList{}
 	spcMap := make(map[string]v1alpha1.SecretProviderClass)
-	secretOwnerMap := make(map[types.NamespacedName][]*v1alpha1.SecretProviderClassPodStatus)
+	secretOwnerMap := make(map[types.NamespacedName][]metav1.OwnerReference)
 	// get a list of all spc pod status that belong to the node
 	err := r.reader.List(ctx, spcPodStatusList, r.ListOptionsLabelSelector())
 	if err != nil {
@@ -119,21 +122,56 @@ func (r *SecretProviderClassPodStatusReconciler) Patcher(ctx context.Context) er
 	for i := range spcPodStatuses {
 		spcName := spcPodStatuses[i].Status.SecretProviderClassName
 		spc := &v1alpha1.SecretProviderClass{}
-		if val, exists := spcMap[spcPodStatuses[i].Namespace+"/"+spcName]; exists {
+		namespace := spcPodStatuses[i].Namespace
+
+		if val, exists := spcMap[namespace+"/"+spcName]; exists {
 			spc = &val
 		} else {
-			if err := r.reader.Get(ctx, client.ObjectKey{Namespace: spcPodStatuses[i].Namespace, Name: spcName}, spc); err != nil {
+			if err := r.reader.Get(ctx, client.ObjectKey{Namespace: namespace, Name: spcName}, spc); err != nil {
 				return fmt.Errorf("failed to get spc %s, err: %+v", spcName, err)
 			}
-			spcMap[spcPodStatuses[i].Namespace+"/"+spcName] = *spc
+			spcMap[namespace+"/"+spcName] = *spc
+		}
+		// get the pod and check if the pod has a owner reference
+		pod := &v1.Pod{}
+		err = r.reader.Get(ctx, client.ObjectKey{Namespace: namespace, Name: spcPodStatuses[i].Status.PodName}, pod)
+		if err != nil {
+			return fmt.Errorf("failed to fetch pod during patching, err: %+v", err)
 		}
+		var ownerRefs []metav1.OwnerReference
+		for _, ownerRef := range pod.GetOwnerReferences() {
+			ownerRefs = append(ownerRefs, metav1.OwnerReference{
+				APIVersion: ownerRef.APIVersion,
+				Kind:       ownerRef.Kind,
+				UID:        ownerRef.UID,
+				Name:       ownerRef.Name,
+			})
+		}
+		// If a pod has no owner references, then it's a static pod and
+		// doesn't belong to a replicaset. In this case, use the spcps as
+		// owner reference just like we do it today
+		if len(ownerRefs) == 0 {
+			// Create a new owner ref.
+			gvk, err := apiutil.GVKForObject(&spcPodStatuses[i], r.scheme)
+			if err != nil {
+				return err
+			}
+			ref := metav1.OwnerReference{
+				APIVersion: gvk.GroupVersion().String(),
+				Kind:       gvk.Kind,
+				UID:        spcPodStatuses[i].GetUID(),
+				Name:       spcPodStatuses[i].GetName(),
+			}
+			ownerRefs = append(ownerRefs, ref)
+		}
+
 		for _, secret := range spc.Spec.SecretObjects {
-			key := types.NamespacedName{Name: secret.SecretName, Namespace: spcPodStatuses[i].Namespace}
+			key := types.NamespacedName{Name: secret.SecretName, Namespace: namespace}
 			val, exists := secretOwnerMap[key]
 			if exists {
-				secretOwnerMap[key] = append(val, &spcPodStatuses[i])
+				secretOwnerMap[key] = append(val, ownerRefs...)
 			} else {
-				secretOwnerMap[key] = []*v1alpha1.SecretProviderClassPodStatus{&spcPodStatuses[i]}
+				secretOwnerMap[key] = ownerRefs
 			}
 		}
 	}
@@ -191,22 +229,6 @@ func (r *SecretProviderClassPodStatusReconciler) Reconcile(ctx context.Context,
 		return ctrl.Result{}, err
 	}
 
-	// reconcile delete
-	if !spcPodStatus.GetDeletionTimestamp().IsZero() {
-		klog.InfoS("reconcile complete", "spcps", req.NamespacedName.String())
-		return ctrl.Result{}, nil
-	}
-
-	node, ok := spcPodStatus.GetLabels()[v1alpha1.InternalNodeLabel]
-	if !ok {
-		klog.V(3).InfoS("node label not found, ignoring this spc pod status", "spcps", klog.KObj(spcPodStatus))
-		return ctrl.Result{}, nil
-	}
-	if !strings.EqualFold(node, r.nodeID) {
-		klog.V(3).InfoS("ignoring as spc pod status belongs diff node", "node", node, "spcps", klog.KObj(spcPodStatus))
-		return ctrl.Result{}, nil
-	}
-
 	// Obtain the full pod metadata. An object reference is needed for sending
 	// events and the UID is helpful for validating the SPCPS TargetPath.
 	pod := &v1.Pod{}
@@ -296,7 +318,7 @@ func (r *SecretProviderClassPodStatusReconciler) Reconcile(ctx context.Context,
 			// Set secrets-store.csi.k8s.io/managed=true label on the secret that's created and managed
 			// by the secrets-store-csi-driver. This label will be used to perform a filtered list watch
 			// only on secrets created and managed by the driver
-			labelsMap[secretManagedLabel] = "true"
+			labelsMap[SecretManagedLabel] = "true"
 
 			createFn := func() (bool, error) {
 				if err := r.createK8sSecret(ctx, secretName, req.Namespace, datamap, labelsMap, secretType); err != nil {
@@ -335,9 +357,41 @@ func (r *SecretProviderClassPodStatusReconciler) Reconcile(ctx context.Context,
 func (r *SecretProviderClassPodStatusReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1.SecretProviderClassPodStatus{}).
+		WithEventFilter(r.belongsToNodePredicate()).
 		Complete(r)
 }
 
+// belongsToNodePredicate defines predicates for handlers
+func (r *SecretProviderClassPodStatusReconciler) belongsToNodePredicate() predicate.Funcs {
+	return predicate.Funcs{
+		UpdateFunc: func(e event.UpdateEvent) bool {
+			return r.processIfBelongsToNode(e.ObjectNew)
+		},
+		CreateFunc: func(e event.CreateEvent) bool {
+			return r.processIfBelongsToNode(e.Object)
+		},
+		DeleteFunc: func(e event.DeleteEvent) bool {
+			return false
+		},
+		GenericFunc: func(e event.GenericEvent) bool {
+			return r.processIfBelongsToNode(e.Object)
+		},
+	}
+}
+
+// processIfBelongsToNode determines if the secretproviderclasspodstatus belongs to the node based on the
+// internal.secrets-store.csi.k8s.io/node-name: <node name> label. If belongs to node, then the spcps is processed.
+func (r *SecretProviderClassPodStatusReconciler) processIfBelongsToNode(objMeta metav1.Object) bool {
+	node, ok := objMeta.GetLabels()[v1alpha1.InternalNodeLabel]
+	if !ok {
+		return false
+	}
+	if !strings.EqualFold(node, r.nodeID) {
+		return false
+	}
+	return true
+}
+
 // createK8sSecret creates K8s secret with data from mounted files
 // If a secret with the same name already exists in the namespace of the pod, the error is nil.
 func (r *SecretProviderClassPodStatusReconciler) createK8sSecret(ctx context.Context, name, namespace string, datamap map[string][]byte, labelsmap map[string]string, secretType corev1.SecretType) error {
@@ -363,7 +417,7 @@ func (r *SecretProviderClassPodStatusReconciler) createK8sSecret(ctx context.Con
 }
 
 // patchSecretWithOwnerRef patches the secret owner reference with the spc pod status
-func (r *SecretProviderClassPodStatusReconciler) patchSecretWithOwnerRef(ctx context.Context, name, namespace string, spcPodStatus ...*v1alpha1.SecretProviderClassPodStatus) error {
+func (r *SecretProviderClassPodStatusReconciler) patchSecretWithOwnerRef(ctx context.Context, name, namespace string, ownerRefs ...metav1.OwnerReference) error {
 	secret := &corev1.Secret{}
 	secretKey := types.NamespacedName{
 		Namespace: namespace,
@@ -380,23 +434,23 @@ func (r *SecretProviderClassPodStatusReconciler) patchSecretWithOwnerRef(ctx con
 	patch := client.MergeFromWithOptions(secret.DeepCopy(), client.MergeFromWithOptimisticLock{})
 	needsPatch := false
 
+	secretOwnerRefs := secret.GetOwnerReferences()
 	secretOwnerMap := make(map[string]types.UID)
-	for _, or := range secret.GetOwnerReferences() {
+	for _, or := range secretOwnerRefs {
 		secretOwnerMap[or.Name] = or.UID
 	}
 
-	for i := range spcPodStatus {
-		if _, exists := secretOwnerMap[spcPodStatus[i].Name]; exists {
+	for i := range ownerRefs {
+		if _, exists := secretOwnerMap[ownerRefs[i].Name]; exists {
 			continue
 		}
 		needsPatch = true
-		err := controllerutil.SetOwnerReference(spcPodStatus[i], secret, r.scheme)
-		if err != nil {
-			return err
-		}
+		klog.Infof("Adding %s/%s as owner ref for %s/%s", ownerRefs[i].APIVersion, ownerRefs[i].Name, namespace, name)
+		secretOwnerRefs = append(secretOwnerRefs, ownerRefs[i])
 	}
 
 	if needsPatch {
+		secret.SetOwnerReferences(secretOwnerRefs)
 		return r.writer.Patch(ctx, secret, patch)
 	}
 	return nil