operator: user adoption and add credential sync for externally-managed secrets

.changes/unreleased/operator-Added-20260409-120000.yaml

@@ -0,0 +1,4 @@
+project: operator
+kind: Added
+body: Support importing existing Redpanda users into operator management and ongoing credential sync via spec.authentication.syncCredentials for external secret rotation (e.g. ESO).
+time: 2026-04-09T12:00:00.000000-04:00

operator/api/redpanda/v1alpha2/testdata/crd-docs.adoc

@@ -4748,6 +4748,11 @@ UserAuthenticationSpec defines the authentication mechanism enabled for this Red
 - scram-sha-256 + | scram-sha-512 | Enum: [scram-sha-256 scram-sha-512 SCRAM-SHA-256 SCRAM-SHA-512] +
 
 | *`password`* __xref:{anchor_prefix}-github-com-redpanda-data-redpanda-operator-operator-api-redpanda-v1alpha2-password[$$Password$$]__ | Password specifies where a password is read from. + |  | 
+| *`syncCredentials`* __boolean__ | SyncCredentials when set to true causes the operator to re-read the +
+password from the referenced Secret on every reconciliation cycle +
+(default: every 5 minutes) and upsert the credentials to Redpanda. +
+This enables password rotation via external systems like the External +
+Secrets Operator (ESO) without requiring user recreation. + |  | 
 |===
 
 

operator/api/redpanda/v1alpha2/user_types.go

@@ -72,6 +72,28 @@ func (u *User) HasManagedUser() bool {
 	return u.Status.ManagedUser
 }
 
+// ShouldSyncCredentials returns true if the user has credential sync enabled,
+// meaning the operator should re-read the password from the referenced Secret
+// and upsert it to Redpanda on each reconciliation cycle.
+func (u *User) ShouldSyncCredentials() bool {
+	return u.Spec.Authentication != nil && u.Spec.Authentication.SyncCredentials
+}
+
+// GetPasswordSecretName returns the name of the Secret referenced by the
+// user's password configuration, or empty string if no Secret is referenced.
+func (u *User) GetPasswordSecretName() string {
+	if u.Spec.Authentication == nil {
+		return ""
+	}
+	if u.Spec.Authentication.Password.ValueFrom == nil {
+		return ""
+	}
+	if u.Spec.Authentication.Password.ValueFrom.SecretKeyRef == nil {
+		return ""
+	}
+	return u.Spec.Authentication.Password.ValueFrom.SecretKeyRef.Name
+}
+
 func (u *User) ShouldManageACLs() bool {
 	return u.Spec.Authorization != nil
 }
@@ -125,6 +147,12 @@ type UserAuthenticationSpec struct {
 	Type *SASLMechanism `json:"type,omitempty"`
 	// Password specifies where a password is read from.
 	Password Password `json:"password"`
+	// SyncCredentials when set to true causes the operator to re-read the
+	// password from the referenced Secret on every reconciliation cycle
+	// (default: every 5 minutes) and upsert the credentials to Redpanda.
+	// This enables password rotation via external systems like the External
+	// Secrets Operator (ESO) without requiring user recreation.
+	SyncCredentials bool `json:"syncCredentials,omitempty"`
 }
 
 // Password specifies a password for the user.

operator/config/crd/bases/cluster.redpanda.com_users.yaml

@@ -110,6 +110,14 @@ spec:
                     x-kubernetes-validations:
                     - message: valueFrom must not be empty if no value supplied
                       rule: self.value != "" || has(self.valueFrom)
+                  syncCredentials:
+                    description: |-
+                      SyncCredentials when set to true causes the operator to re-read the
+                      password from the referenced Secret on every reconciliation cycle
+                      (default: every 5 minutes) and upsert the credentials to Redpanda.
+                      This enables password rotation via external systems like the External
+                      Secrets Operator (ESO) without requiring user recreation.
+                    type: boolean
                   type:
                     default: scram-sha-512
                     description: |-

operator/internal/controller/redpanda/user_controller.go

@@ -18,8 +18,12 @@ import (
 	"github.com/twmb/franz-go/pkg/kgo"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/fields"
+	"k8s.io/apimachinery/pkg/types"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"
+	mchandler "sigs.k8s.io/multicluster-runtime/pkg/handler"
 
 	redpandav1alpha2ac "github.com/redpanda-data/redpanda-operator/operator/api/applyconfiguration/redpanda/v1alpha2"
 	redpandav1alpha2 "github.com/redpanda-data/redpanda-operator/operator/api/redpanda/v1alpha2"
@@ -87,14 +91,25 @@ func (r *UserReconciler) SyncResource(ctx context.Context, request ResourceReque
 	defer usersClient.Close()
 	defer syncer.Close()
 
-	if !hasUser && shouldManageUser {
+	switch {
+	case shouldManageUser && !hasManagedUser:
+		// Create a new user or adopt an existing one. UpsertSCRAM is
+		// idempotent so this is safe regardless of whether the user
+		// already exists in Redpanda.
 		if err := usersClient.Create(ctx, user); err != nil {
 			return createPatch(err)
 		}
 		hasManagedUser = true
-	}
 
-	if hasUser && !shouldManageUser {
+	case shouldManageUser && hasManagedUser && user.ShouldSyncCredentials():
+		// Re-read the password from the referenced Secret and upsert
+		// credentials. This enables external rotation (e.g. via ESO)
+		// to propagate on each reconciliation cycle.
+		if err := usersClient.Update(ctx, user); err != nil {
+			return createPatch(err)
+		}
+
+	case !shouldManageUser && hasUser && hasManagedUser:
 		if err := usersClient.Delete(ctx, user); err != nil {
 			return createPatch(err)
 		}
@@ -171,6 +186,8 @@ func (r *UserReconciler) userAndACLClients(ctx context.Context, request Resource
 	return usersClient, syncer, hasUser, nil
 }
 
+const userPasswordSecretIndex = "__user_referencing_password_secret"
+
 func SetupUserController(ctx context.Context, mgr multicluster.Manager, expander *secrets.CloudExpander, includeV1, includeV2 bool, namespace string) error {
 	factory := internalclient.NewFactory(mgr, expander)
 
@@ -194,12 +211,67 @@ func SetupUserController(ctx context.Context, mgr multicluster.Manager, expander
 			}
 			builder.Watches(&redpandav1alpha2.Redpanda{}, enqueueV2User, controller.WatchOptions(clusterName)...)
 		}
+
+		// Index Users by the password Secret they reference so we can
+		// watch for external Secret changes (e.g. from ESO) and
+		// immediately reconcile the referencing User.
+		cluster, err := mgr.GetCluster(ctx, clusterName)
+		if err != nil {
+			return err
+		}
+		if err := cluster.GetFieldIndexer().IndexField(ctx, &redpandav1alpha2.User{}, userPasswordSecretIndex, func(o client.Object) []string {
+			user, ok := o.(*redpandav1alpha2.User)
+			if !ok {
+				return nil
+			}
+			if name := user.GetPasswordSecretName(); name != "" {
+				return []string{types.NamespacedName{Namespace: user.Namespace, Name: name}.String()}
+			}
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		builder.Watches(&corev1.Secret{}, enqueueUsersForSecret(mgr, clusterName), controller.WatchOptions(clusterName)...)
 	}
 
-	controller := NewResourceController(mgr, factory, &UserReconciler{}, "UserReconciler")
+	ctrl := NewResourceController(mgr, factory, &UserReconciler{}, "UserReconciler")
 
 	// Every 5 minutes try and check to make sure no manual modifications
 	// happened on the resource synced to the cluster and attempt to correct
 	// any drift.
-	return builder.Complete(controller.PeriodicallyReconcile(5 * time.Minute).FilterNamespace(namespace))
+	return builder.Complete(ctrl.PeriodicallyReconcile(5 * time.Minute).FilterNamespace(namespace))
+}
+
+// enqueueUsersForSecret returns an event handler that, when a Secret changes,
+// looks up all User resources that reference that Secret via
+// spec.authentication.password.valueFrom.secretKeyRef and enqueues them for
+// reconciliation. This enables immediate reconciliation when an external
+// system (e.g. ESO) updates a password Secret.
+func enqueueUsersForSecret(mgr multicluster.Manager, clusterName string) mchandler.EventHandlerFunc {
+	return mchandler.EnqueueRequestsFromMapFunc(func(ctx context.Context, o client.Object) []reconcile.Request {
+		cluster, err := mgr.GetCluster(ctx, clusterName)
+		if err != nil {
+			return nil
+		}
+
+		var userList redpandav1alpha2.UserList
+		nn := types.NamespacedName{Namespace: o.GetNamespace(), Name: o.GetName()}
+		if err := cluster.GetClient().List(ctx, &userList, &client.ListOptions{
+			FieldSelector: fields.OneTermEqualSelector(userPasswordSecretIndex, nn.String()),
+		}); err != nil {
+			return nil
+		}
+
+		requests := make([]reconcile.Request, 0, len(userList.Items))
+		for i := range userList.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: userList.Items[i].Namespace,
+					Name:      userList.Items[i].Name,
+				},
+			})
+		}
+		return requests
+	})
 }