データベースOperatorを作る — バックアップ・フェイルオーバー・スケールをコードで

apiVersion: db.example.com/v1alpha1
kind: Database
metadata:
  name: orders-db
spec:
  replicas: 3
  version: "16.3"
  storage:
    size: 50Gi
    storageClassName: fast-ssd
  backup:
    enabled: true
    schedule: "0 3 * * *"
    retention: 14
    destination: s3://my-backups/orders-db
status:
  phase: Running
  primary: orders-db-0
  readyReplicas: 3
  lastBackupTime: "2026-06-15T03:00:12Z"

// DatabaseSpec defines the desired state of Database.
type DatabaseSpec struct {
	// +kubebuilder:validation:Minimum=1
	// +kubebuilder:validation:Maximum=9
	// +kubebuilder:default=1
	Replicas int32 `json:"replicas"`

	// +kubebuilder:validation:Required
	Version string `json:"version"`

	Storage StorageSpec `json:"storage"`

	// +optional
	Backup *BackupSpec `json:"backup,omitempty"`
}

type StorageSpec struct {
	Size             string  `json:"size"`
	StorageClassName *string `json:"storageClassName,omitempty"`
}

type BackupSpec struct {
	Enabled     bool   `json:"enabled"`
	Schedule    string `json:"schedule"`
	Retention   int32  `json:"retention"`
	Destination string `json:"destination"`
}

// DatabaseStatus defines the observed state of Database.
type DatabaseStatus struct {
	Phase          string      `json:"phase,omitempty"`
	Primary        string      `json:"primary,omitempty"`
	ReadyReplicas  int32       `json:"readyReplicas,omitempty"`
	LastBackupTime *metav1.Time `json:"lastBackupTime,omitempty"`
	Conditions     []metav1.Condition `json:"conditions,omitempty"`
}

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := log.FromContext(ctx)

	// 1. 対象CRを取得
	var db dbv1alpha1.Database
	if err := r.Get(ctx, req.NamespacedName, &db); err != nil {
		// すでに削除されていれば無視
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// 2. 削除処理(finalizer)
	if !db.DeletionTimestamp.IsZero() {
		return r.reconcileDelete(ctx, &db)
	}
	if !controllerutil.ContainsFinalizer(&db, dbFinalizer) {
		controllerutil.AddFinalizer(&db, dbFinalizer)
		if err := r.Update(ctx, &db); err != nil {
			return ctrl.Result{}, err
		}
	}

	// 3. desiredなリソースを順に保証
	if err := r.reconcileHeadlessService(ctx, &db); err != nil {
		return ctrl.Result{}, err
	}
	if err := r.reconcileStatefulSet(ctx, &db); err != nil {
		return ctrl.Result{}, err
	}
	if err := r.reconcileBackupCronJob(ctx, &db); err != nil {
		return ctrl.Result{}, err
	}

	// 4. 状態更新
	if err := r.updateStatus(ctx, &db); err != nil {
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

func (r *DatabaseReconciler) reconcileStatefulSet(ctx context.Context, db *dbv1alpha1.Database) error {
	sts := &appsv1.StatefulSet{
		ObjectMeta: metav1.ObjectMeta{
			Name:      db.Name,
			Namespace: db.Namespace,
		},
	}

	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, sts, func() error {
		// 望ましいスペックを埋める(冪等)
		sts.Spec.Replicas = &db.Spec.Replicas
		sts.Spec.ServiceName = db.Name + "-headless"
		sts.Spec.Selector = &metav1.LabelSelector{
			MatchLabels: labelsFor(db),
		}
		sts.Spec.Template = podTemplateFor(db)
		sts.Spec.VolumeClaimTemplates = pvcTemplatesFor(db)

		// owner reference: CR削除時にStatefulSetもGC
		return controllerutil.SetControllerReference(db, sts, r.Scheme)
	})
	return err
}

+------------------------------------------+
|              Database CR                 |
+-------------------+----------------------+
                    | reconcile
        +-----------+-----------+-----------+
        v           v           v
  StatefulSet   Headless     CronJob
   (Pod N個)    Service      (バックアップ)
        |           |
        v           v
  安定した識別子   DNSレコード
  db-0, db-1     db-0.db-headless
  db-2 ...       (Pod別の固定アドレス)
        |
        v
  VolumeClaimTemplate
   -> Podごとに専用PVC
   -> Pod再起動でもデータ維持

apiVersion: v1
kind: Service
metadata:
  name: orders-db-headless
spec:
  clusterIP: None
  selector:
    app: orders-db
  ports:
    - port: 5432
      name: postgres

func (r *DatabaseReconciler) reconcileBackupCronJob(ctx context.Context, db *dbv1alpha1.Database) error {
	name := db.Name + "-backup"
	cj := &batchv1.CronJob{
		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: db.Namespace},
	}

	// バックアップがオフならCronJobを整理
	if db.Spec.Backup == nil || !db.Spec.Backup.Enabled {
		err := r.Delete(ctx, cj)
		return client.IgnoreNotFound(err)
	}

	_, err := controllerutil.CreateOrUpdate(ctx, r.Client, cj, func() error {
		cj.Spec.Schedule = db.Spec.Backup.Schedule
		cj.Spec.JobTemplate.Spec.Template.Spec = backupPodSpec(db)
		return controllerutil.SetControllerReference(db, cj, r.Scheme)
	})
	return err
}

フェイルオーバー手順(安全弁付き)
 1. プライマリのヘルス監視(例:5秒ごとのlivenessプローブ)
 2. N回連続失敗 -> 「疑い」状態に転換(即時昇格禁止)
 3. フェンシング(fencing):旧プライマリの書き込みを遮断
    - 書き込みServiceセレクタから旧プライマリを除去
    - またはネットワーク/ストレージレベルで隔離
 4. 最も先行したレプリカを選択(レプリケーション遅延が最小)
 5. そのレプリカをプライマリに昇格
 6. 書き込みServiceラベルを新プライマリへ移動
 7. 残りのレプリカを新プライマリに再接続
 8. status.primaryを更新、イベントを記録

アップグレード順序(Postgres系基準)
 1. すべてのレプリカが健康か確認(不健康なら中断)
 2. バックアップを一度強制実行(ロールバック対策)
 3. レプリカから先にアップグレード(db-2 -> db-1 ...)
    - 一台アップグレード -> Ready確認 -> レプリケーション追いつき確認 -> 次
 4. 最後にプライマリ:
    - レプリカ一つを新バージョンに昇格(計画されたフェイルオーバー)
    - 旧プライマリを新バージョンで再起動後レプリカとして合流
 5. status.versionを更新

// アップグレードを段階的に進める擬似コード
func (r *DatabaseReconciler) reconcileUpgrade(ctx context.Context, db *dbv1alpha1.Database) (ctrl.Result, error) {
	target := db.Spec.Version
	// 最も高いインデックスの旧バージョンレプリカを探す
	pod, found := r.findOutdatedReplica(ctx, db, target)
	if !found {
		return ctrl.Result{}, nil // すべて最新
	}
	if !r.isHealthyAndCaughtUp(ctx, db) {
		// まだ追いつき中 -> しばらく後に再確認
		return ctrl.Result{RequeueAfter: 15 * time.Second}, nil
	}
	if err := r.upgradePod(ctx, pod, target); err != nil {
		return ctrl.Result{}, err
	}
	// 一台処理したので再びreconcile
	return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
}

func (r *DatabaseReconciler) updateStatus(ctx context.Context, db *dbv1alpha1.Database) error {
	var sts appsv1.StatefulSet
	if err := r.Get(ctx, types.NamespacedName{Name: db.Name, Namespace: db.Namespace}, &sts); err != nil {
		return client.IgnoreNotFound(err)
	}

	db.Status.ReadyReplicas = sts.Status.ReadyReplicas
	if sts.Status.ReadyReplicas == db.Spec.Replicas {
		db.Status.Phase = "Running"
		meta.SetStatusCondition(&db.Status.Conditions, metav1.Condition{
			Type:    "Ready",
			Status:  metav1.ConditionTrue,
			Reason:  "AllReplicasReady",
			Message: "all database replicas are ready",
		})
	} else {
		db.Status.Phase = "Progressing"
		meta.SetStatusCondition(&db.Status.Conditions, metav1.Condition{
			Type:    "Ready",
			Status:  metav1.ConditionFalse,
			Reason:  "ReplicasNotReady",
			Message: "waiting for replicas to become ready",
		})
	}
	return r.Status().Update(ctx, db)
}

// +kubebuilder:printcolumn:name="Phase",type=string,JSONPath=`.status.phase`
// +kubebuilder:printcolumn:name="Replicas",type=integer,JSONPath=`.status.readyReplicas`
// +kubebuilder:printcolumn:name="Primary",type=string,JSONPath=`.status.primary`
// +kubebuilder:subresource:status

var (
	failoverTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "database_failover_total",
			Help: "Number of failovers performed per database",
		},
		[]string{"database", "namespace"},
	)
)

func init() {
	metrics.Registry.MustRegister(failoverTotal)
}

r.Recorder.Event(db, corev1.EventTypeNormal, "FailoverCompleted",
	fmt.Sprintf("promoted %s to primary", newPrimary))

シナリオ1. ノード障害でプライマリPodが消滅
  -> スケジューラが別ノードにPodを再作成、PVCが付いてくる
  -> その間HAエージェントがレプリカを昇格、Operatorが書き込みServiceラベルを移動
  -> 旧Podが新ノードで立つとレプリカとして合流

シナリオ2. ディスクが満杯
  -> statusに警告conditionを公開、イベント発生
  -> ユーザーがspec.storage.sizeを増加 -> OperatorがPVC拡張(許可時)

シナリオ3. 負荷増加で読み取りレプリカ追加が必要
  -> spec.replicas 3 -> 5 -> StatefulSetスケール -> 新レプリカがベースバックアップで同期

シナリオ4. 定期バックアップ失敗(S3権限の期限切れ)
  -> バックアップJob失敗、status.lastBackupTime未更新
  -> アラートルールが「X時間以上バックアップなし」を検知して呼び出し

[ ] PVCをむやみに削除 — owner referenceにPVCを縛るとCR削除時にデータが蒸発。
     データ保存が必要ならPVCはGC対象から除外し別途ポリシーで。
[ ] reconcileでブロッキング — 長い作業(バックアップなど)をreconcile内で同期的に
     待つとワーカーキューが詰まる。別のJob/CronJobに委譲すること。
[ ] 冪等性違反 — reconcile呼び出しごとに新しいバックアップJobを作ると無限増殖。
     常に「すでにあるか」を先に確認。
[ ] statusを入力として使用 — split-brainの判断をstatusだけに依存すると危険。
[ ] 無限requeue — エラー時に即時再試行すると暴走。指数バックオフを活用。
[ ] 過度なRBAC — データベースOperatorだからとcluster-adminを要求しないこと。
[ ] バージョンスキップ放置 — メジャーバージョンジャンプを防がないとデータ破損。