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 {

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)
 	}
 

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,7 @@ func (r *MySQLClusterReconciler) reconcileV1Secret(ctx context.Context, req ctrl
 
 	// During credential rotation, the CredentialRotationReconciler owns
 	// secret distribution. Skip to prevent overwriting pending passwords.
+	// Fail closed on transient errors to avoid distributing stale passwords.
 	var cr mocov1beta2.CredentialRotation
 	if err := r.Get(ctx, client.ObjectKey{
 		Namespace: cluster.Namespace,
@@ -358,6 +359,8 @@ func (r *MySQLClusterReconciler) reconcileV1Secret(ctx context.Context, req ctrl
 			log.Info("credential rotation in progress, skipping secret distribution", "phase", phase)
 			return nil
 		}
+	} else if !apierrors.IsNotFound(err) && !meta.IsNoMatchError(err) {
+		return fmt.Errorf("failed to check CredentialRotation status: %w", err)
 	}
 
 	if err := r.reconcileUserSecret(ctx, req, cluster, secret); err != nil {

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)

pkg/password/rotation.go

@@ -63,6 +63,14 @@ var userToPendingKey = map[string]string{
 	constants.WritableUser:    WritablePasswordPendingKey,
 }
 
+// GetRotationID returns the stored ROTATION_ID from the secret, or "" if absent.
+func GetRotationID(secret *corev1.Secret) string {
+	if secret.Data == nil {
+		return ""
+	}
+	return string(secret.Data[RotationIDKey])
+}
+
 // HasPendingPasswords validates the pending state of a source secret.
 //
 // Returns:
@@ -217,11 +225,11 @@ func ConfirmPendingPasswords(secret *corev1.Secret) error {
 // Returns error if pending keys are not all present.
 func PendingKeyMap(secret *corev1.Secret) (map[string]string, error) {
 	if secret.Data == nil {
-		return nil, fmt.Errorf("secret has no data")
+		return nil, fmt.Errorf("secret %s/%s has no data", secret.Namespace, secret.Name)
 	}
 	for _, key := range allPendingKeys {
 		if _, ok := secret.Data[key]; !ok {
-			return nil, fmt.Errorf("pending key %s not found in secret", key)
+			return nil, fmt.Errorf("pending key %s not found in secret %s/%s", key, secret.Namespace, secret.Name)
 		}
 	}