Implement CredentialRotation CRD for system user password rotation

Add a dedicated CredentialRotation CRD and controller to rotate all 8
MOCO system user passwords without MySQL downtime, using MySQL's dual
password feature (8.0.14+).

Components:
- CRD types with 6-phase lifecycle (Rotating → Retained → Rotated →
  Discarding → Discarded → Completed)
- Validation webhook for create/update
- CredentialRotationReconciler for K8s resource operations (Secret
  distribution, StatefulSet rolling restart, phase transitions)
- ClusterManager handlers for MySQL operations (ALTER USER RETAIN,
  DISCARD OLD PASSWORD, auth plugin migration)
- Pending password helpers with idempotent set/confirm operations
- DB operations with sql_log_bin=0 to prevent cross-cluster propagation
- kubectl moco credential rotate/discard subcommands
- Guard in reconcileV1Secret to skip distribution during active rotation

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

Author: shunki-fujita

api/v1beta2/credentialrotation_types.go

@@ -0,0 +1,79 @@
+package v1beta2
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// CredentialRotationSpec defines the desired state of CredentialRotation.
+// The target MySQLCluster is identified by the CR's own name and namespace
+// (CredentialRotation name must equal MySQLCluster name).
+type CredentialRotationSpec struct {
+	// RotationGeneration is a monotonically increasing counter.
+	// Incrementing this value triggers a new rotation cycle.
+	// +optional
+	RotationGeneration int64 `json:"rotationGeneration,omitempty"`
+
+	// DiscardOldPassword triggers the discard phase.
+	// Can only be set to true when Phase is Rotated.
+	// Must be reset to false when incrementing rotationGeneration.
+	// +optional
+	DiscardOldPassword bool `json:"discardOldPassword,omitempty"`
+}
+
+// CredentialRotationStatus defines the observed state of CredentialRotation.
+type CredentialRotationStatus struct {
+	// Phase is the current rotation phase.
+	// +optional
+	Phase RotationPhase `json:"phase,omitempty"`
+
+	// RotationID is the UUID for this rotation cycle.
+	// +optional
+	RotationID string `json:"rotationID,omitempty"`
+
+	// ObservedRotationGeneration is the last rotationGeneration
+	// that completed successfully.
+	// +optional
+	ObservedRotationGeneration int64 `json:"observedRotationGeneration,omitempty"`
+}
+
+// RotationPhase represents the phase of a credential rotation.
+type RotationPhase string
+
+const (
+	RotationPhaseRotating   RotationPhase = "Rotating"
+	RotationPhaseRetained   RotationPhase = "Retained"
+	RotationPhaseRotated    RotationPhase = "Rotated"
+	RotationPhaseDiscarding RotationPhase = "Discarding"
+	RotationPhaseDiscarded  RotationPhase = "Discarded"
+	RotationPhaseCompleted  RotationPhase = "Completed"
+)
+
+// +kubebuilder:object:root=true
+// +kubebuilder:subresource:status
+// +kubebuilder:storageversion
+// +kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase"
+// +kubebuilder:printcolumn:name="Generation",type="integer",JSONPath=".spec.rotationGeneration"
+// +kubebuilder:printcolumn:name="Observed",type="integer",JSONPath=".status.observedRotationGeneration"
+// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp"
+
+// CredentialRotation is the Schema for the credentialrotations API
+type CredentialRotation struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   CredentialRotationSpec   `json:"spec,omitempty"`
+	Status CredentialRotationStatus `json:"status,omitempty"`
+}
+
+//+kubebuilder:object:root=true
+
+// CredentialRotationList contains a list of CredentialRotation
+type CredentialRotationList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []CredentialRotation `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&CredentialRotation{}, &CredentialRotationList{})
+}

api/v1beta2/credentialrotation_webhook.go

@@ -0,0 +1,145 @@
+package v1beta2
+
+import (
+	"context"
+
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+)
+
+func (r *CredentialRotation) SetupWebhookWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(r).
+		WithValidator(&credentialRotationAdmission{client: mgr.GetAPIReader()}).
+		Complete()
+}
+
+type credentialRotationAdmission struct {
+	client client.Reader
+}
+
+//+kubebuilder:webhook:path=/validate-moco-cybozu-com-v1beta2-credentialrotation,mutating=false,failurePolicy=fail,sideEffects=None,matchPolicy=Equivalent,groups=moco.cybozu.com,resources=credentialrotations,verbs=create;update,versions=v1beta2,name=vcredentialrotation.kb.io,admissionReviewVersions=v1
+
+var _ webhook.CustomValidator = &credentialRotationAdmission{}
+
+func (a *credentialRotationAdmission) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
+	cr := obj.(*CredentialRotation)
+
+	var errs field.ErrorList
+
+	// MySQLCluster with the same name must exist
+	cluster := &MySQLCluster{}
+	if err := a.client.Get(ctx, types.NamespacedName{
+		Namespace: cr.Namespace,
+		Name:      cr.Name,
+	}, cluster); err != nil {
+		if apierrors.IsNotFound(err) {
+			errs = append(errs, field.Invalid(field.NewPath("metadata", "name"), cr.Name,
+				"MySQLCluster with the same name must exist in the same namespace"))
+		} else {
+			errs = append(errs, field.InternalError(field.NewPath("metadata", "name"), err))
+		}
+	} else {
+		// Replicas must be > 0
+		if cluster.Spec.Replicas <= 0 {
+			errs = append(errs, field.Invalid(field.NewPath("metadata", "name"), cr.Name,
+				"target MySQLCluster must have replicas > 0"))
+		}
+	}
+
+	// rotationGeneration must be > 0
+	if cr.Spec.RotationGeneration <= 0 {
+		errs = append(errs, field.Invalid(field.NewPath("spec", "rotationGeneration"),
+			cr.Spec.RotationGeneration, "must be > 0"))
+	}
+
+	// discardOldPassword must be false
+	if cr.Spec.DiscardOldPassword {
+		errs = append(errs, field.Invalid(field.NewPath("spec", "discardOldPassword"),
+			cr.Spec.DiscardOldPassword, "must be false on creation"))
+	}
+
+	if len(errs) > 0 {
+		return nil, apierrors.NewInvalid(
+			schema.GroupKind{Group: GroupVersion.Group, Kind: "CredentialRotation"}, cr.Name, errs)
+	}
+	return nil, nil
+}
+
+func (a *credentialRotationAdmission) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
+	oldCR := oldObj.(*CredentialRotation)
+	newCR := newObj.(*CredentialRotation)
+
+	var errs field.ErrorList
+
+	// rotationGeneration must be monotonically increasing
+	if newCR.Spec.RotationGeneration < oldCR.Spec.RotationGeneration {
+		errs = append(errs, field.Invalid(field.NewPath("spec", "rotationGeneration"),
+			newCR.Spec.RotationGeneration, "must be >= previous value (monotonically increasing)"))
+	}
+
+	generationChanged := newCR.Spec.RotationGeneration > oldCR.Spec.RotationGeneration
+
+	if generationChanged {
+		// rotationGeneration can only increase when Phase is "" or Completed
+		phase := oldCR.Status.Phase
+		if phase != "" && phase != RotationPhaseCompleted {
+			errs = append(errs, field.Forbidden(field.NewPath("spec", "rotationGeneration"),
+				"can only increment rotationGeneration when phase is empty or Completed"))
+		}
+
+		// When rotationGeneration increases, discardOldPassword must be false
+		if newCR.Spec.DiscardOldPassword {
+			errs = append(errs, field.Invalid(field.NewPath("spec", "discardOldPassword"),
+				newCR.Spec.DiscardOldPassword, "must be false when incrementing rotationGeneration"))
+		}
+
+		// Check MySQLCluster replicas > 0
+		cluster := &MySQLCluster{}
+		if err := a.client.Get(ctx, types.NamespacedName{
+			Namespace: newCR.Namespace,
+			Name:      newCR.Name,
+		}, cluster); err != nil {
+			if apierrors.IsNotFound(err) {
+				errs = append(errs, field.Invalid(field.NewPath("metadata", "name"), newCR.Name,
+					"MySQLCluster with the same name must exist"))
+			} else {
+				errs = append(errs, field.InternalError(field.NewPath("metadata", "name"), err))
+			}
+		} else if cluster.Spec.Replicas <= 0 {
+			errs = append(errs, field.Invalid(field.NewPath("metadata", "name"), newCR.Name,
+				"target MySQLCluster must have replicas > 0"))
+		}
+	} else {
+		// When rotationGeneration is unchanged, discardOldPassword can only go false→true
+		if oldCR.Spec.DiscardOldPassword && !newCR.Spec.DiscardOldPassword {
+			errs = append(errs, field.Forbidden(field.NewPath("spec", "discardOldPassword"),
+				"cannot change from true to false without incrementing rotationGeneration"))
+		}
+
+		// discardOldPassword=true requires Phase==Rotated
+		if newCR.Spec.DiscardOldPassword && !oldCR.Spec.DiscardOldPassword {
+			if oldCR.Status.Phase != RotationPhaseRotated {
+				errs = append(errs, field.Forbidden(field.NewPath("spec", "discardOldPassword"),
+					"can only set to true when phase is Rotated"))
+			}
+		}
+	}
+
+	if len(errs) > 0 {
+		return nil, apierrors.NewInvalid(
+			schema.GroupKind{Group: GroupVersion.Group, Kind: "CredentialRotation"}, newCR.Name, errs)
+	}
+	return nil, nil
+}
+
+func (a *credentialRotationAdmission) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
+	return nil, nil
+}