feat: propagate credential rotation to the collectors

docs/content/docs/user-guide/target.md

@@ -122,6 +122,28 @@ spec:
   credentialsRef: device-credentials
 ```
 
+### Rotating Credentials
+
+Update the Secret and the operator pushes the new credentials to the collectors
+on its own — no need to touch the TargetProfile, the Targets or the Cluster:
+
+```bash
+kubectl create secret generic device-credentials \
+  --from-literal=username=admin \
+  --from-literal=password=newpassword \
+  --dry-run=client -o yaml | kubectl apply -f -
+```
+
+The operator watches Secrets referenced by a TargetProfile and reconciles every
+Cluster collecting with them. Only a change to the Secret's data triggers this;
+adding a label or an annotation does not.
+
+Rotation is not atomic across a cluster. The collectors are reconfigured one
+after another, so for a short window some pods present the old credentials and
+some the new. Where the device rejects the old ones, expect connection errors on
+the targets that have not been reconfigured yet; they clear as the rollout
+completes.
+
 ## TLS Configuration
 
 The `TargetProfile` controls **connection-level TLS settings** for gNMI connections. For **client certificate authentication (mTLS)**, see [Cluster Client TLS]({{< ref "../user-guide/cluster#gnmi-client-tls-target-connections" >}}).

internal/controller/cluster_controller.go

@@ -1006,6 +1006,27 @@ func (p generationOrLabelsChangedPredicate) Update(e event.UpdateEvent) bool {
 	return !maps.Equal(e.ObjectOld.GetLabels(), e.ObjectNew.GetLabels())
 }
 
+// secretDataChangedPredicate triggers reconciliation only when a Secret's
+// contents change.
+//
+// Secrets carry no generation, so the predicates used for the CRDs do not apply
+// and the default would wake the controller on every write to every Secret in
+// scope — annotations, ownership churn, cert-manager renewals of unrelated
+// material. Only Data decides what gets baked into a TargetConfig, so only Data
+// is worth a reconcile.
+type secretDataChangedPredicate struct {
+	predicate.Funcs
+}
+
+func (secretDataChangedPredicate) Update(e event.UpdateEvent) bool {
+	oldSecret, okOld := e.ObjectOld.(*corev1.Secret)
+	newSecret, okNew := e.ObjectNew.(*corev1.Secret)
+	if !okOld || !okNew {
+		return false
+	}
+	return !maps.EqualFunc(oldSecret.Data, newSecret.Data, bytes.Equal)
+}
+
 // SetupWithManager sets up the controller with the Manager.
 func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	r.m = &sync.RWMutex{}
@@ -1061,6 +1082,11 @@ func (r *ClusterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 			handler.EnqueueRequestsFromMapFunc(r.findClustersForTunnelTargetPolicy),
 			builder.WithPredicates(specOrLabelsPredicate),
 		).
+		Watches(
+			&corev1.Secret{},
+			handler.EnqueueRequestsFromMapFunc(r.findClustersForSecret),
+			builder.WithPredicates(secretDataChangedPredicate{}),
+		).
 		Complete(r)
 }
 
@@ -1153,32 +1179,114 @@ func (r *ClusterReconciler) findClustersForTargetProfile(ctx context.Context, ob
 	if !ok {
 		return nil
 	}
+	return r.findClustersUsingProfiles(ctx, profile.Namespace, map[string]struct{}{profile.Name: {}})
+}
+
+// findClustersForSecret finds all Clusters collecting with the credentials this
+// Secret holds.
+//
+// Without this the credentials baked into each TargetConfig are only rebuilt
+// when something else happens to wake the Cluster reconciler, so a rotated
+// password reaches the pods whenever the next unrelated event does — or at the
+// resync, which is the framework default of about ten hours.
+func (r *ClusterReconciler) findClustersForSecret(ctx context.Context, obj client.Object) []reconcile.Request {
+	secret, ok := obj.(*corev1.Secret)
+	if !ok {
+		return nil
+	}
+	// A TargetProfile is the only thing that turns a Secret into target
+	// credentials. Most Secrets in a namespace belong to something else
+	// entirely, and they stop here at the cost of one cached list.
+	var profileList gnmicv1alpha1.TargetProfileList
+	if err := r.List(ctx, &profileList, client.InNamespace(secret.Namespace)); err != nil {
+		return nil
+	}
+	profiles := make(map[string]struct{})
+	for i := range profileList.Items {
+		if profileList.Items[i].Spec.CredentialsRef == secret.Name {
+			profiles[profileList.Items[i].Name] = struct{}{}
+		}
+	}
+	if len(profiles) == 0 {
+		return nil
+	}
+	return r.findClustersUsingProfiles(ctx, secret.Namespace, profiles)
+}
 
-	// list all targets in the same namespace
+// profileUser is something that names a TargetProfile and can itself be
+// selected by a Pipeline, which is what makes it a path from a profile to a
+// cluster.
+type profileUser struct {
+	name   string
+	labels map[string]string
+	kind   string // as understood by pipelineReferencesResource
+}
+
+// findClustersUsingProfiles resolves a set of TargetProfile names to the
+// Clusters that collect with them.
+//
+// Three cached lists regardless of the size of the set. The obvious
+// implementation calls findClustersReferencingResource once per matching
+// target, which re-lists every Pipeline each time and turns a single event into
+// O(targets x pipelines) work.
+func (r *ClusterReconciler) findClustersUsingProfiles(ctx context.Context, namespace string, profiles map[string]struct{}) []reconcile.Request {
 	var targetList gnmicv1alpha1.TargetList
-	if err := r.List(ctx, &targetList, client.InNamespace(profile.Namespace)); err != nil {
+	if err := r.List(ctx, &targetList, client.InNamespace(namespace)); err != nil {
 		return nil
 	}
+	var users []profileUser
+	for i := range targetList.Items {
+		t := &targetList.Items[i]
+		if _, ok := profiles[t.Spec.Profile]; ok {
+			users = append(users, profileUser{name: t.Name, labels: t.Labels, kind: "target"})
+		}
+	}
 
-	// find clusters for each target that uses this profile
-	seen := make(map[types.NamespacedName]struct{})
-	var results []reconcile.Request
+	// Tunnel targets are discovered at runtime rather than declared, so their
+	// credentials come from the policy's profile and no Target object exists to
+	// find them by.
+	var policyList gnmicv1alpha1.TunnelTargetPolicyList
+	if err := r.List(ctx, &policyList, client.InNamespace(namespace)); err != nil {
+		return nil
+	}
+	for i := range policyList.Items {
+		p := &policyList.Items[i]
+		if _, ok := profiles[p.Spec.Profile]; ok {
+			users = append(users, profileUser{name: p.Name, labels: p.Labels, kind: "tunnel-target-policy"})
+		}
+	}
+	if len(users) == 0 {
+		return nil
+	}
 
-	for _, target := range targetList.Items {
-		if target.Spec.Profile != profile.Name {
+	var pipelineList gnmicv1alpha1.PipelineList
+	if err := r.List(ctx, &pipelineList, client.InNamespace(namespace)); err != nil {
+		return nil
+	}
+	clusterSet := make(map[string]struct{})
+	for i := range pipelineList.Items {
+		pipeline := &pipelineList.Items[i]
+		if !pipeline.Spec.Enabled {
 			continue
 		}
-		// find clusters referencing this target
-		targetResults := r.findClustersReferencingResource(ctx, target.Namespace, target.Name, target.Labels, "target")
-		for _, req := range targetResults {
-			if _, ok := seen[req.NamespacedName]; !ok {
-				seen[req.NamespacedName] = struct{}{}
-				results = append(results, req)
+		if _, done := clusterSet[pipeline.Spec.ClusterRef]; done {
+			continue
+		}
+		for _, u := range users {
+			if pipelineReferencesResource(pipeline, u.name, u.labels, u.kind) {
+				clusterSet[pipeline.Spec.ClusterRef] = struct{}{}
+				break
 			}
 		}
 	}
 
-	return results
+	requests := make([]reconcile.Request, 0, len(clusterSet))
+	for clusterName := range clusterSet {
+		requests = append(requests, reconcile.Request{
+			NamespacedName: types.NamespacedName{Name: clusterName, Namespace: namespace},
+		})
+	}
+	return requests
 }
 
 // findClustersForTunnelTargetPolicy finds all Clusters that have Pipelines referencing this TunnelTargetPolicy

internal/controller/secret_watch_test.go

@@ -0,0 +1,196 @@
+package controller
+
+import (
+	"context"
+	"sort"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+
+	gnmicv1alpha1 "github.com/gnmic/operator/api/v1alpha1"
+)
+
+func secretWatchScheme(t *testing.T) *runtime.Scheme {
+	t.Helper()
+	scheme := runtime.NewScheme()
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	if err := gnmicv1alpha1.AddToScheme(scheme); err != nil {
+		t.Fatal(err)
+	}
+	return scheme
+}
+
+func reconcilerWith(t *testing.T, objs ...client.Object) *ClusterReconciler {
+	t.Helper()
+	scheme := secretWatchScheme(t)
+	cl := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build()
+	return &ClusterReconciler{Client: cl, Scheme: scheme}
+}
+
+func clusterNames(t *testing.T, r *ClusterReconciler, obj client.Object) []string {
+	t.Helper()
+	reqs := r.findClustersForSecret(context.Background(), obj)
+	out := make([]string, 0, len(reqs))
+	for _, req := range reqs {
+		out = append(out, req.Name)
+	}
+	sort.Strings(out)
+	return out
+}
+
+func secret(name string) *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"},
+		Data:       map[string][]byte{"username": []byte("u"), "password": []byte("p")},
+	}
+}
+
+func profile(name, credsRef string) *gnmicv1alpha1.TargetProfile {
+	return &gnmicv1alpha1.TargetProfile{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"},
+		Spec:       gnmicv1alpha1.TargetProfileSpec{CredentialsRef: credsRef},
+	}
+}
+
+func target(name, profileName string, labels map[string]string) *gnmicv1alpha1.Target {
+	return &gnmicv1alpha1.Target{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default", Labels: labels},
+		Spec:       gnmicv1alpha1.TargetSpec{Profile: profileName, Address: "10.0.0.1:57400"},
+	}
+}
+
+func pipelineSelectingTargets(name, clusterRef string, enabled bool, labels map[string]string) *gnmicv1alpha1.Pipeline {
+	return &gnmicv1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "default"},
+		Spec: gnmicv1alpha1.PipelineSpec{
+			ClusterRef:      clusterRef,
+			Enabled:         enabled,
+			TargetSelectors: []metav1.LabelSelector{{MatchLabels: labels}},
+		},
+	}
+}
+
+// A rotated Secret must reach the cluster whose pipeline collects with it. This
+// is the whole point of the watch: without it the new credentials sit in the
+// API until an unrelated event happens to trigger a reconcile.
+func TestFindClustersForSecret_ReachesCollectingCluster(t *testing.T) {
+	r := reconcilerWith(t,
+		secret("creds"),
+		profile("default", "creds"),
+		target("leaf1", "default", map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p1", "c1", true, map[string]string{"tag": "prod"}),
+	)
+	if got := clusterNames(t, r, secret("creds")); len(got) != 1 || got[0] != "c1" {
+		t.Fatalf("clusters = %v, want [c1]", got)
+	}
+}
+
+// Most Secrets in a namespace have nothing to do with the operator. They must
+// cost one cached list and no reconcile.
+func TestFindClustersForSecret_UnreferencedSecretEnqueuesNothing(t *testing.T) {
+	r := reconcilerWith(t,
+		secret("creds"),
+		secret("unrelated"),
+		profile("default", "creds"),
+		target("leaf1", "default", map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p1", "c1", true, map[string]string{"tag": "prod"}),
+	)
+	if got := clusterNames(t, r, secret("unrelated")); len(got) != 0 {
+		t.Fatalf("clusters = %v, want none", got)
+	}
+}
+
+// Tunnel targets are discovered at runtime, so no Target object names the
+// profile. Resolving only through Targets misses them entirely.
+func TestFindClustersForSecret_ReachesTunnelTargetPolicy(t *testing.T) {
+	policy := &gnmicv1alpha1.TunnelTargetPolicy{
+		ObjectMeta: metav1.ObjectMeta{Name: "tp1", Namespace: "default", Labels: map[string]string{"tag": "tunnel"}},
+		Spec:       gnmicv1alpha1.TunnelTargetPolicySpec{Profile: "default"},
+	}
+	pipeline := &gnmicv1alpha1.Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "p1", Namespace: "default"},
+		Spec: gnmicv1alpha1.PipelineSpec{
+			ClusterRef:                  "c1",
+			Enabled:                     true,
+			TunnelTargetPolicySelectors: []metav1.LabelSelector{{MatchLabels: map[string]string{"tag": "tunnel"}}},
+		},
+	}
+	r := reconcilerWith(t, secret("creds"), profile("default", "creds"), policy, pipeline)
+	if got := clusterNames(t, r, secret("creds")); len(got) != 1 || got[0] != "c1" {
+		t.Fatalf("clusters = %v, want [c1]", got)
+	}
+}
+
+// A disabled pipeline is not collecting, so nothing needs re-pushing.
+func TestFindClustersForSecret_SkipsDisabledPipeline(t *testing.T) {
+	r := reconcilerWith(t,
+		secret("creds"),
+		profile("default", "creds"),
+		target("leaf1", "default", map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p1", "c1", false, map[string]string{"tag": "prod"}),
+	)
+	if got := clusterNames(t, r, secret("creds")); len(got) != 0 {
+		t.Fatalf("clusters = %v, want none", got)
+	}
+}
+
+// One Secret can back several profiles, and several clusters can collect with
+// them. Each cluster must be enqueued once.
+func TestFindClustersForSecret_DeduplicatesClusters(t *testing.T) {
+	r := reconcilerWith(t,
+		secret("creds"),
+		profile("a", "creds"),
+		profile("b", "creds"),
+		target("leaf1", "a", map[string]string{"tag": "prod"}),
+		target("leaf2", "b", map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p1", "c1", true, map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p2", "c1", true, map[string]string{"tag": "prod"}),
+		pipelineSelectingTargets("p3", "c2", true, map[string]string{"tag": "prod"}),
+	)
+	got := clusterNames(t, r, secret("creds"))
+	if len(got) != 2 || got[0] != "c1" || got[1] != "c2" {
+		t.Fatalf("clusters = %v, want [c1 c2]", got)
+	}
+}
+
+// Secrets carry no generation, so the predicate is the only thing standing
+// between the controller and a reconcile per unrelated Secret write.
+func TestSecretDataChangedPredicate(t *testing.T) {
+	p := secretDataChangedPredicate{}
+
+	old := secret("creds")
+	unchanged := secret("creds")
+	if p.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: unchanged}) {
+		t.Error("identical data triggered a reconcile")
+	}
+
+	relabelled := secret("creds")
+	relabelled.Labels = map[string]string{"touched": "true"}
+	if p.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: relabelled}) {
+		t.Error("metadata-only change triggered a reconcile")
+	}
+
+	rotated := secret("creds")
+	rotated.Data["password"] = []byte("new")
+	if !p.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: rotated}) {
+		t.Error("rotated password did not trigger a reconcile")
+	}
+
+	removed := secret("creds")
+	delete(removed.Data, "password")
+	if !p.Update(event.UpdateEvent{ObjectOld: old, ObjectNew: removed}) {
+		t.Error("removed key did not trigger a reconcile")
+	}
+
+	if p.Update(event.UpdateEvent{ObjectOld: &corev1.ConfigMap{}, ObjectNew: &corev1.ConfigMap{}}) {
+		t.Error("non-Secret objects should not pass the predicate")
+	}
+}