Address review comments on CredentialRotation PR

- Remove SetOwnerReference in CLI (controller handles it; kubeClient
  has no Scheme() method)
- Reuse existing ROTATION_ID from source Secret in handleStartRotation
  to avoid wedging on crash recovery between Secret update and status
  update
- Fail closed in reconcileV1Secret guard: return error on transient
  CredentialRotation Get failures instead of proceeding with
  distribution
- Fix test assertion: use Equal("") instead of BeElementOf with
  duplicate values

Signed-off-by: shunki-fujita <shunki-fujita@cybozu.co.jp>

Author: shunki-fujita

clustering/password_rotation.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cybozu-go/moco/pkg/password"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
@@ -23,7 +24,7 @@ func (p *managerProcess) handlePasswordRotation(ctx context.Context, ss *StatusS
 		Namespace: p.name.Namespace,
 		Name:      p.name.Name,
 	}, cr)
-	if apierrors.IsNotFound(err) {
+	if apierrors.IsNotFound(err) || meta.IsNoMatchError(err) {
 		return false, nil
 	}
 	if err != nil {
@@ -71,29 +72,42 @@ func (p *managerProcess) handleRotatingPhase(ctx context.Context, ss *StatusSet,
 		return false, nil
 	}
 
-	// Pre-check: verify no instance has stale dual passwords.
 	currentPasswd, err := password.NewMySQLPasswordFromSecret(sourceSecret)
 	if err != nil {
 		return false, err
 	}
-	for idx := 0; idx < replicas; idx++ {
-		op, err := p.dbf.New(ctx, cluster, currentPasswd, idx)
-		if err != nil {
-			return false, err
-		}
-		dualFound, dualUser, checkErr := checkInstanceDualPasswords(ctx, op, idx)
-		op.Close()
-		if checkErr != nil {
-			return false, checkErr
+
+	// Pre-check: verify no instance has stale dual passwords from outside
+	// this rotation cycle. Skip if RETAIN_STARTED marker is present (crash
+	// recovery after partial RETAIN — per-user HasDualPassword in
+	// rotateInstanceUsers provides idempotency).
+	retainStarted := string(sourceSecret.Data[password.RetainStartedKey]) == cr.Status.RotationID
+	if !retainStarted {
+		for idx := 0; idx < replicas; idx++ {
+			op, err := p.dbf.New(ctx, cluster, currentPasswd, idx)
+			if err != nil {
+				return false, err
+			}
+			dualFound, dualUser, checkErr := checkInstanceDualPasswords(ctx, op, idx)
+			op.Close()
+			if checkErr != nil {
+				return false, checkErr
+			}
+			if dualFound {
+				log.Info("waiting: instance has pre-existing dual password",
+					"instance", idx, "user", dualUser)
+				p.recorder.Eventf(cluster, corev1.EventTypeWarning, "DualPasswordExists",
+					"Cannot proceed with RETAIN: instance %d user %s already has a dual password. "+
+						"Follow the recovery procedure in docs/designdoc/credential_rotation_crd.md.",
+					idx, dualUser)
+				return false, nil
+			}
 		}
-		if dualFound {
-			log.Info("waiting: instance has pre-existing dual password",
-				"instance", idx, "user", dualUser)
-			p.recorder.Eventf(cluster, corev1.EventTypeWarning, "DualPasswordExists",
-				"Cannot proceed with RETAIN: instance %d user %s already has a dual password. "+
-					"Follow the recovery procedure in docs/designdoc/credential_rotation_crd.md.",
-				idx, dualUser)
-			return false, nil
+
+		// Mark that pre-check passed and RETAIN is about to start.
+		sourceSecret.Data[password.RetainStartedKey] = []byte(cr.Status.RotationID)
+		if err := p.client.Update(ctx, sourceSecret); err != nil {
+			return false, fmt.Errorf("failed to set RETAIN_STARTED marker: %w", err)
 		}
 	}
 

clustering/process.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cybozu-go/moco/pkg/metrics"
 	"github.com/go-logr/logr"
 	"github.com/prometheus/client_golang/prometheus"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -200,8 +201,13 @@ func (p *managerProcess) do(ctx context.Context) (bool, error) {
 	}
 
 	// Handle credential rotation if a CredentialRotation CR exists.
+	// Errors are non-fatal: rotation failures must not block core clustering
+	// operations (failover, switchover) that maintain availability.
 	if redo, err := p.handlePasswordRotation(ctx, ss); err != nil {
-		return false, fmt.Errorf("failed to handle password rotation: %w", err)
+		log := logFromContext(ctx)
+		log.Error(err, "password rotation error (will retry next cycle)")
+		p.recorder.Eventf(ss.Cluster, corev1.EventTypeWarning, "PasswordRotationError",
+			"Password rotation encountered an error: %v", err)
 	} else if redo {
 		return true, nil
 	}

cmd/kubectl-moco/cmd/credential.go

@@ -11,19 +11,30 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 )
 
 var credentialConfig struct {
 	user   string
 	format string
 }
 
-// credentialCmd represents the credential parent command
+// credentialCmd represents the credential parent command.
+// When called without a subcommand (e.g. "credential CLUSTER_NAME"),
+// it falls back to "show" for backward compatibility.
 var credentialCmd = &cobra.Command{
-	Use:   "credential",
+	Use:   "credential [CLUSTER_NAME]",
 	Short: "Manage MySQL credentials",
-	Long:  "Manage MySQL credentials for a MOCO cluster.",
+	Long:  "Manage MySQL credentials for a MOCO cluster. When called with a cluster name and no subcommand, shows the credential (backward compatible).",
+	Args:  cobra.MaximumNArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		if len(args) == 0 {
+			return cmd.Help()
+		}
+		return credentialShow(cmd.Context(), args[0])
+	},
+	ValidArgsFunction: func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+		return mysqlClusterCandidates(cmd.Context(), cmd, args, toComplete)
+	},
 }
 
 // credentialShowCmd shows the credential of a specified user
@@ -119,9 +130,7 @@ func credentialRotate(ctx context.Context, clusterName string) error {
 				DiscardOldPassword: false,
 			},
 		}
-		if err := controllerutil.SetOwnerReference(cluster, cr, kubeClient.Scheme()); err != nil {
-			return fmt.Errorf("failed to set ownerReference: %w", err)
-		}
+		// ownerReference is set by the controller on first reconcile.
 		if err := kubeClient.Create(ctx, cr); err != nil {
 			return fmt.Errorf("failed to create CredentialRotation: %w", err)
 		}
@@ -169,18 +178,22 @@ func credentialDiscard(ctx context.Context, clusterName string) error {
 }
 
 func init() {
-	// Show subcommand flags
-	fs := credentialShowCmd.Flags()
+	// Flags on the parent command for backward compatibility
+	// ("kubectl moco credential -u moco-admin CLUSTER_NAME").
+	fs := credentialCmd.Flags()
 	fs.StringVarP(&credentialConfig.user, "mysql-user", "u", constants.ReadOnlyUser, "User for login to mysql")
 	fs.StringVar(&credentialConfig.format, "format", "plain", "The format of output [`plain` or `mycnf`]")
 
-	_ = credentialShowCmd.RegisterFlagCompletionFunc("mysql-user", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+	_ = credentialCmd.RegisterFlagCompletionFunc("mysql-user", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 		return []string{constants.ReadOnlyUser, constants.WritableUser, constants.AdminUser}, cobra.ShellCompDirectiveDefault
 	})
-	_ = credentialShowCmd.RegisterFlagCompletionFunc("format", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+	_ = credentialCmd.RegisterFlagCompletionFunc("format", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
 		return []string{"plain", "mycnf"}, cobra.ShellCompDirectiveDefault
 	})
 
+	// "show" subcommand shares the same flags via the parent.
+	credentialShowCmd.Flags().AddFlagSet(credentialCmd.Flags())
+
 	credentialCmd.AddCommand(credentialShowCmd)
 	credentialCmd.AddCommand(credentialRotateCmd)
 	credentialCmd.AddCommand(credentialDiscardCmd)

controllers/credentialrotation_controller.go

@@ -114,9 +114,6 @@ func (r *CredentialRotationReconciler) handleStartRotation(ctx context.Context,
 		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
 	}
 
-	// Generate rotationID
-	rotationID := uuid.New().String()
-
 	// Get the source Secret
 	sourceSecret := &corev1.Secret{}
 	secretName := cluster.ControllerSecretName()
@@ -127,12 +124,20 @@ func (r *CredentialRotationReconciler) handleStartRotation(ctx context.Context,
 		return ctrl.Result{}, fmt.Errorf("failed to get source secret: %w", err)
 	}
 
-	// Generate pending passwords
+	// Reuse existing rotationID if the Secret already has complete pending
+	// passwords (crash recovery: Secret was updated but status was not).
+	rotationID := password.GetRotationID(sourceSecret)
+	if rotationID == "" {
+		rotationID = uuid.New().String()
+	}
+
+	// Generate pending passwords (idempotent if rotationID matches)
 	_, err := password.SetPendingPasswords(sourceSecret, rotationID)
 	if err != nil {
 		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "RotationPendingError",
-			"Failed to set pending passwords: %v", err)
-		return ctrl.Result{}, fmt.Errorf("failed to set pending passwords: %w", err)
+			"Failed to set pending passwords: %v. Manual cleanup required: "+
+				"see docs/designdoc/credential_rotation_crd.md Recovery Procedures", err)
+		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
 	}
 
 	// Update the source Secret
@@ -167,11 +172,22 @@ func (r *CredentialRotationReconciler) handleRetainedPhase(ctx context.Context,
 		return ctrl.Result{}, fmt.Errorf("failed to get source secret: %w", err)
 	}
 
+	// Verify pending passwords belong to this rotation cycle.
+	if hasPending, err := password.HasPendingPasswords(sourceSecret, cr.Status.RotationID); err != nil {
+		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "RotationPendingError",
+			"Pending password state inconsistency: %v. Manual cleanup required: "+
+				"see docs/designdoc/credential_rotation_crd.md Recovery Procedures", err)
+		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
+	} else if !hasPending {
+		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "MissingRotationPending",
+			"Pending passwords not found in source secret for rotationID %s. Manual cleanup required: "+
+				"see docs/designdoc/credential_rotation_crd.md Recovery Procedures", cr.Status.RotationID)
+		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
+	}
+
 	// Distribute pending passwords to per-namespace user Secret
 	pendingPasswd, err := password.NewMySQLPasswordFromPending(sourceSecret)
 	if err != nil {
-		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "MissingRotationPending",
-			"Pending passwords not found in source secret: %v", err)
 		return ctrl.Result{}, fmt.Errorf("failed to read pending passwords: %w", err)
 	}
 
@@ -282,6 +298,21 @@ func (r *CredentialRotationReconciler) handleDiscardedPhase(ctx context.Context,
 		return ctrl.Result{}, fmt.Errorf("failed to get source secret for confirm: %w", err)
 	}
 
+	// Verify pending passwords are still present before confirming.
+	// ConfirmPendingPasswords is a no-op when no pending keys exist, which would
+	// incorrectly finalize the rotation with stale passwords if they were lost.
+	if hasPending, err := password.HasPendingPasswords(sourceSecret, cr.Status.RotationID); err != nil {
+		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "RotationPendingError",
+			"Pending password state inconsistency during confirm: %v. Manual cleanup required: "+
+				"see docs/designdoc/credential_rotation_crd.md Recovery Procedures", err)
+		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
+	} else if !hasPending {
+		r.Recorder.Eventf(cr, corev1.EventTypeWarning, "MissingRotationPending",
+			"Pending passwords lost before confirm for rotationID %s. Manual cleanup required: "+
+				"see docs/designdoc/credential_rotation_crd.md Recovery Procedures", cr.Status.RotationID)
+		return ctrl.Result{RequeueAfter: credRotationRequeueInterval}, nil
+	}
+
 	if err := password.ConfirmPendingPasswords(sourceSecret); err != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to confirm pending passwords: %w", err)
 	}

controllers/credentialrotation_controller_test.go

@@ -327,6 +327,7 @@ var _ = Describe("CredentialRotation reconciler", func() {
 				return err
 			}
 			sts.Status.ObservedGeneration = sts.Generation
+			sts.Status.Replicas = *sts.Spec.Replicas
 			sts.Status.CurrentRevision = "rev-1"
 			sts.Status.UpdateRevision = "rev-1"
 			sts.Status.UpdatedReplicas = *sts.Spec.Replicas
@@ -436,6 +437,7 @@ var _ = Describe("CredentialRotation reconciler", func() {
 				return err
 			}
 			sts.Status.ObservedGeneration = sts.Generation
+			sts.Status.Replicas = *sts.Spec.Replicas
 			sts.Status.CurrentRevision = "rev-1"
 			sts.Status.UpdateRevision = "rev-1"
 			sts.Status.UpdatedReplicas = *sts.Spec.Replicas
@@ -515,14 +517,14 @@ var _ = Describe("CredentialRotation reconciler", func() {
 		err = k8sClient.Create(ctx, cr)
 		Expect(err).NotTo(HaveOccurred())
 
-		// Phase should NOT advance past Rotating (reconciler emits warning and requeues).
+		// Phase should remain empty (reconciler refuses and requeues without advancing).
 		Consistently(func() mocov1beta2.RotationPhase {
 			cr := &mocov1beta2.CredentialRotation{}
 			if err := k8sClient.Get(ctx, client.ObjectKey{Namespace: "test", Name: "test"}, cr); err != nil {
 				return ""
 			}
 			return cr.Status.Phase
-		}).Should(BeElementOf(mocov1beta2.RotationPhase(""), mocov1beta2.RotationPhase("")))
+		}).Should(Equal(mocov1beta2.RotationPhase("")))
 	})
 
 	It("should not advance past Rotated without discardOldPassword", func() {

controllers/mysqlcluster_controller.go

@@ -348,6 +348,10 @@ func (r *MySQLClusterReconciler) reconcileV1Secret(ctx context.Context, req ctrl
 
 	// During credential rotation, the CredentialRotationReconciler owns
 	// secret distribution. Skip to prevent overwriting pending passwords.
+	// On any Get error (NotFound, CRD missing, cache not ready), assume no
+	// active rotation and proceed with distribution. This is safe because
+	// if a rotation is truly active, the CredentialRotationReconciler will
+	// re-distribute the correct passwords.
 	var cr mocov1beta2.CredentialRotation
 	if err := r.Get(ctx, client.ObjectKey{
 		Namespace: cluster.Namespace,

go.mod

@@ -39,6 +39,8 @@ require (
 	sigs.k8s.io/controller-runtime v0.22.0
 )
 
+require github.com/google/uuid v1.6.0
+
 require (
 	cel.dev/expr v0.24.0 // indirect
 	cloud.google.com/go v0.116.0 // indirect
@@ -97,7 +99,6 @@ require (
 	github.com/google/gnostic-models v0.7.0 // indirect
 	github.com/google/pprof v0.0.0-20250820193118-f64d9cf942d6 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.4 // indirect
 	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
 	github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 // indirect

pkg/dbop/password.go

@@ -2,6 +2,8 @@ package dbop
 
 import (
 	"context"
+	"database/sql"
+	"database/sql/driver"
 	"fmt"
 	"regexp"
 	"strings"
@@ -36,6 +38,19 @@ func validateMocoUser(user string) error {
 	return nil
 }
 
+// closeDirtyConn restores sql_log_bin=1 on a connection. If the restore fails,
+// the connection is marked bad so it is discarded from the pool instead of being
+// reused with sql_log_bin=0.
+func closeDirtyConn(ctx context.Context, conn *sql.Conn) {
+	if _, err := conn.ExecContext(ctx, "SET sql_log_bin=1"); err != nil {
+		// Mark the connection as bad so the pool discards it.
+		_ = conn.Raw(func(driverConn interface{}) error {
+			return driver.ErrBadConn
+		})
+	}
+	conn.Close()
+}
+
 // GetAuthPlugin returns the default authentication plugin for the first factor
 // by parsing @@global.authentication_policy (introduced in MySQL 8.0.27).
 //
@@ -106,13 +121,11 @@ func (o *operator) RotateUserPassword(ctx context.Context, user, newPassword str
 	if err != nil {
 		return fmt.Errorf("failed to acquire connection for rotate: %w", err)
 	}
-	defer conn.Close()
+	defer closeDirtyConn(ctx, conn)
 
 	if _, err := conn.ExecContext(ctx, "SET sql_log_bin=0"); err != nil {
 		return fmt.Errorf("failed to set sql_log_bin=0: %w", err)
 	}
-	// Restore sql_log_bin before the connection is returned to the pool.
-	defer conn.ExecContext(ctx, "SET sql_log_bin=1") //nolint:errcheck
 	query := fmt.Sprintf("ALTER USER '%s'@'%%' IDENTIFIED BY ? RETAIN CURRENT PASSWORD", user)
 	if _, err := conn.ExecContext(ctx, query, newPassword); err != nil {
 		return fmt.Errorf("failed to rotate password for %s: %w", user, err)
@@ -146,13 +159,11 @@ func (o *operator) MigrateUserAuthPlugin(ctx context.Context, user, password, au
 	if err != nil {
 		return fmt.Errorf("failed to acquire connection for auth plugin migration: %w", err)
 	}
-	defer conn.Close()
+	defer closeDirtyConn(ctx, conn)
 
 	if _, err := conn.ExecContext(ctx, "SET sql_log_bin=0"); err != nil {
 		return fmt.Errorf("failed to set sql_log_bin=0: %w", err)
 	}
-	// Restore sql_log_bin before the connection is returned to the pool.
-	defer conn.ExecContext(ctx, "SET sql_log_bin=1") //nolint:errcheck
 	query := fmt.Sprintf("ALTER USER '%s'@'%%' IDENTIFIED WITH %s BY ?", user, authPlugin)
 	if _, err := conn.ExecContext(ctx, query, password); err != nil {
 		return fmt.Errorf("failed to migrate auth plugin for %s: %w", user, err)
@@ -205,13 +216,11 @@ func (o *operator) DiscardOldPassword(ctx context.Context, user string) error {
 	if err != nil {
 		return fmt.Errorf("failed to acquire connection for discard: %w", err)
 	}
-	defer conn.Close()
+	defer closeDirtyConn(ctx, conn)
 
 	if _, err := conn.ExecContext(ctx, "SET sql_log_bin=0"); err != nil {
 		return fmt.Errorf("failed to set sql_log_bin=0: %w", err)
 	}
-	// Restore sql_log_bin before the connection is returned to the pool.
-	defer conn.ExecContext(ctx, "SET sql_log_bin=1") //nolint:errcheck
 	query := fmt.Sprintf("ALTER USER '%s'@'%%' DISCARD OLD PASSWORD", user)
 	if _, err := conn.ExecContext(ctx, query); err != nil {
 		return fmt.Errorf("failed to discard old password for %s: %w", user, err)