feat: enable filtered secret watch with feature flag

Signed-off-by: Anish Ramasekar <anish.ramasekar@gmail.com>

Author: aramase

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

@@ -52,7 +52,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 +61,12 @@ var (
 	enableProfile          = flag.Bool("enable-pprof", false, "enable pprof profiling")
 	profilePort            = flag.Int("pprof-port", 6065, "port for pprof profiling")
 
+	// enable filtered watch for secrets. The filtering is done on the csi driver label: secrets-store.csi.k8s.io/managed=true
+	// this label is set for all Kubernetes secrets created by the CSI driver. 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/managed=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 secrets with label secrets-store.csi.k8s.io/managed=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 secret based on secrets-store.csi.k8s.io/managed=true label enabled")
+	}
 
 	// initialize metrics exporter before creating measurements
 	err := metrics.InitMetricsExporter()
@@ -100,18 +108,29 @@ 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[string]string{
+		"pods": fields.OneTermEqualSelector("spec.nodeName", *nodeID).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
+	labelSelectorByResource := map[string]string{
+		"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
+	if *filteredWatchSecret {
+		labelSelectorByResource["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: map[string]string{
-				"pods": fields.OneTermEqualSelector("spec.nodeName", *nodeID).String(),
-			},
-			LabelSelectorByResource: map[string]string{
-				"secretproviderclasspodstatuses": fmt.Sprintf("%s=%s", v1alpha1.InternalNodeLabel, *nodeID),
-				"secrets":                        "secrets-store.csi.k8s.io/managed=true",
-			},
+			FieldSelectorByResource: fieldSelectorByResource,
+			LabelSelectorByResource: labelSelectorByResource,
 		}),
 	})
 	if err != nil {
@@ -161,7 +180,7 @@ 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)
 		}

controllers/secretproviderclasspodstatus_controller.go

@@ -24,6 +24,7 @@ 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"
 
@@ -40,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"
@@ -54,7 +54,7 @@ import (
 )
 
 const (
-	secretManagedLabel         = "secrets-store.csi.k8s.io/managed"
+	SecretManagedLabel         = "secrets-store.csi.k8s.io/managed"
 	secretCreationFailedReason = "FailedToCreateSecret"
 )
 
@@ -110,7 +110,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 {
@@ -121,21 +121,56 @@ func (r *SecretProviderClassPodStatusReconciler) Patcher(ctx context.Context) er
 	for i := range spcPodStatuses {
 		spcName := spcPodStatuses[i].Status.SecretProviderClassName
 		spc := &v1alpha1.SecretProviderClass{}
+		namespace := spcPodStatuses[i].Namespace
+
 		if val, exists := spcMap[spcPodStatuses[i].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}
 			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
 			}
 		}
 	}
@@ -282,7 +317,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 {
@@ -384,7 +419,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,
@@ -401,23 +436,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