Operator の本番アップグレードとマイグレーション — 無停止で進化させる

1. Operator 自体のアップグレード   - コントローラ Deployment のローリング、リーダー選出、graceful shutdown
2. CRD スキーマの進化              - マルチバージョン CRD、conversion webhook、storage version
3. 管理ワークロードの安全な展開    - 段階的/パーティションロールアウト、observedGeneration、status conditions
4. ロールバック戦略               - ダウングレードの落とし穴、不可逆な変換、フィールド消失
5. マルチバージョン運用           - 複数 API バージョンの同時配信、deprecation ポリシー
6. カナリア Operator             - 一部ネームスペースにのみ新バージョンをデプロイ
7. バージョン互換性              - K8s 1.36 / Go 1.26、controller-runtime v0.24.x、version skew
8. 大規模 CR マイグレーション      - 一回限りのバッチジョブ、storage version migrator、レート制限
9. 障害対応                     - ロールアウトの途中で問題が起きたとき
10. チェックリスト               - デプロイ前後の点検項目

package main

import (
	"crypto/tls"
	"flag"
	"os"

	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/healthz"
	"sigs.k8s.io/controller-runtime/pkg/log/zap"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
)

func main() {
	var enableLeaderElection bool
	var probeAddr string
	flag.BoolVar(&enableLeaderElection, "leader-elect", true,
		"Enable leader election for controller manager.")
	flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081",
		"The address the probe endpoint binds to.")
	flag.Parse()

	ctrl.SetLogger(zap.New(zap.UseDevMode(false)))

	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
		Scheme: scheme,
		Metrics: metricsserver.Options{
			BindAddress:   ":8443",
			SecureServing: true,
			// 2026: kube-rbac-proxy は削除。認証/認可をマネージャに内蔵。
			FilterProvider: filters.WithAuthenticationAndAuthorization,
			TLSOpts: []func(*tls.Config){
				func(c *tls.Config) { c.MinVersion = tls.VersionTLS13 },
			},
		},
		HealthProbeBindAddress: probeAddr,
		LeaderElection:         enableLeaderElection,
		LeaderElectionID:       "db-operator.example.com",
		// リーダー停止時にスタンバイが素早く引き継げるよう lease を調整。
		LeaseDuration: durationPtr(15 * time.Second),
		RenewDeadline: durationPtr(10 * time.Second),
		RetryPeriod:   durationPtr(2 * time.Second),
		// 終了時にリーダーシップを即座に返却し、ダウンタイムを減らす。
		LeaderElectionReleaseOnCancel: true,
	})
	if err != nil {
		os.Exit(1)
	}

	_ = mgr.AddHealthzCheck("healthz", healthz.Ping)
	_ = mgr.AddReadyzCheck("readyz", healthz.Ping)

	if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
		os.Exit(1)
	}
}

apiVersion: apps/v1
kind: Deployment
metadata:
  name: db-operator-controller-manager
  namespace: db-operator-system
spec:
  replicas: 2
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      terminationGracePeriodSeconds: 30
      containers:
        - name: manager
          image: registry.example.com/db-operator:v1.4.0
          args:
            - --leader-elect=true
            - --health-probe-bind-address=:8081
          livenessProbe:
            httpGet:
              path: /healthz
              port: 8081
            initialDelaySeconds: 15
            periodSeconds: 20
          readinessProbe:
            httpGet:
              path: /readyz
              port: 8081
            initialDelaySeconds: 5
            periodSeconds: 10
          resources:
            requests:
              cpu: 100m
              memory: 128Mi
            limits:
              memory: 512Mi
          lifecycle:
            preStop:
              exec:
                command: ["/bin/sh", "-c", "sleep 5"]

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

	var db examplev1.Database
	if err := r.Get(ctx, req.NamespacedName, &db); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// 外部呼び出しに ctx を渡すと終了時に即キャンセルされ、graceful shutdown が保証される。
	if err := r.provisioner.EnsureReady(ctx, &db); err != nil {
		if ctx.Err() != nil {
			// 終了中なら次のリーダーが再処理するよう静かに返す。
			return ctrl.Result{}, nil
		}
		log.Error(err, "failed to ensure database ready")
		return ctrl.Result{RequeueAfter: 30 * time.Second}, nil
	}

	return ctrl.Result{}, nil
}

              +----------------------------------------------+
              |                 API Server                    |
  v1alpha1 -->|  served:true   -+                             |
  v1beta1  -->|  served:true    +-> conversion webhook -> etcd| (storage: v1)
  v1       -->|  served:true,   |                             |
              |  storage:true --+                             |
              +----------------------------------------------+

   v1alpha1 (spoke)              v1 (hub, storage)              v1beta1 (spoke)
        |                            ^   |                            |
        |  ConvertTo(hub) -----------+   +---- ConvertFrom(hub) <-----+
        +---------- ConvertFrom(hub) <--- ConvertTo(hub) -------------+

package v1

// Hub はこの型が変換 hub であることを示す。追加の実装は不要。
func (*Database) Hub() {}

package v1beta1

import (
	"sigs.k8s.io/controller-runtime/pkg/conversion"
	dbv1 "github.com/example/db-operator/api/v1"
)

// ConvertTo はこの spoke(v1beta1)を hub(v1)へ変換する。
func (src *Database) ConvertTo(dstRaw conversion.Hub) error {
	dst := dstRaw.(*dbv1.Database)

	dst.ObjectMeta = src.ObjectMeta

	// 単純なフィールドはそのままコピー。
	dst.Spec.Engine = src.Spec.Engine
	dst.Spec.Replicas = src.Spec.Replicas

	// v1beta1 の StorageGB(int) を v1 の Storage(構造体)へ再編。
	dst.Spec.Storage = dbv1.StorageSpec{
		SizeGB:    src.Spec.StorageGB,
		ClassName: "standard", // 新規フィールドのデフォルト値
	}

	// status も変換。
	dst.Status.Phase = src.Status.Phase
	dst.Status.ObservedGeneration = src.Status.ObservedGeneration
	return nil
}

// ConvertFrom は hub(v1)をこの spoke(v1beta1)へ変換する。
func (dst *Database) ConvertFrom(srcRaw conversion.Hub) error {
	src := srcRaw.(*dbv1.Database)

	dst.ObjectMeta = src.ObjectMeta

	dst.Spec.Engine = src.Spec.Engine
	dst.Spec.Replicas = src.Spec.Replicas

	// v1 の Storage 構造体を v1beta1 の StorageGB(int)へ縮約。
	// 注意: ClassName 情報は v1beta1 で表現できないため失われる。
	dst.Spec.StorageGB = src.Spec.Storage.SizeGB

	dst.Status.Phase = src.Status.Phase
	dst.Status.ObservedGeneration = src.Status.ObservedGeneration
	return nil
}

func (r *Database) SetupWebhookWithManager(mgr ctrl.Manager) error {
	return ctrl.NewWebhookManagedBy(mgr).
		For(r).
		Complete()
}

apiVersion: apiextensions.k8s.io/v1
kind: CustomResourceDefinition
metadata:
  name: databases.example.com
spec:
  group: example.com
  names:
    kind: Database
    plural: databases
  scope: Namespaced
  conversion:
    strategy: Webhook
    webhook:
      conversionReviewVersions: ["v1"]
      clientConfig:
        service:
          namespace: db-operator-system
          name: db-operator-webhook-service
          path: /convert
  versions:
    - name: v1alpha1
      served: true
      storage: false
      deprecated: true
      deprecationWarning: "example.com/v1alpha1 Database is deprecated; use v1"
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
    - name: v1beta1
      served: true
      storage: false
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object
    - name: v1
      served: true
      storage: true
      schema:
        openAPIV3Schema:
          type: object
          properties:
            spec:
              type: object

ステップ 1: v1 を served:true, storage:false で追加(まだ保存は v1beta1)
            conversion webhook をデプロイ + 変換関数を検証
ステップ 2: コントローラの新バージョンをデプロイ(v1 を認識)
ステップ 3: storage version を v1 に切り替え(v1beta1 storage:false, v1 storage:true)
ステップ 4: storage version migrator で既存 CR を v1 として再保存
ステップ 5: v1alpha1/v1beta1 を deprecated 表記し、十分な猶予期間の後 served:false
ステップ 6: 参照がなくなったら旧バージョンを削除

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var db examplev1.Database
	if err := r.Get(ctx, req.NamespacedName, &db); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// すでに最新の generation を処理済みなら追加作業なし。
	if db.Status.ObservedGeneration == db.Generation &&
		meta.IsStatusConditionTrue(db.Status.Conditions, "Ready") {
		return ctrl.Result{}, nil
	}

	// ... 実際の reconcile ロジック ...

	db.Status.ObservedGeneration = db.Generation
	meta.SetStatusCondition(&db.Status.Conditions, metav1.Condition{
		Type:               "Ready",
		Status:             metav1.ConditionTrue,
		ObservedGeneration: db.Generation,
		Reason:             "Reconciled",
		Message:            "database is ready",
	})
	if err := r.Status().Update(ctx, &db); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

// RolloutPolicy は CR の spec に定義され、段階的移行を制御する。
type RolloutPolicy struct {
	// MaxUnavailable: 同時に置き換え可能なインスタンスの割合(%)。
	MaxUnavailable int `json:"maxUnavailable"`
	// Partition: このインデックス未満は新バージョンに変えない(漸進的に拡大)。
	Partition int `json:"partition"`
	// Paused: true ならロールアウトを止め、現状を維持する。
	Paused bool `json:"paused"`
}

func (r *DatabaseReconciler) rolloutManagedPods(
	ctx context.Context, db *examplev1.Database, desiredImage string,
) (bool, error) {
	pods, err := r.listManagedPods(ctx, db)
	if err != nil {
		return false, err
	}

	// 一時停止状態なら何も変えない。
	if db.Spec.Rollout.Paused {
		return false, nil
	}

	// インデックスの降順で、partition 以上のものだけを対象にする。
	updating := 0
	maxConcurrent := percentToCount(db.Spec.Rollout.MaxUnavailable, len(pods))

	for i := len(pods) - 1; i >= db.Spec.Rollout.Partition; i-- {
		p := pods[i]
		if podImage(p) == desiredImage {
			continue // すでに新バージョン
		}
		if !isHealthy(p) {
			// 直前に変えた Pod がまだ健全でなければ、これ以上進めない。
			return true, nil
		}
		if updating >= maxConcurrent {
			return true, nil // 同時置換の上限に到達、次の reconcile で継続
		}
		if err := r.recreatePodWithImage(ctx, p, desiredImage); err != nil {
			return false, err
		}
		updating++
	}

	// すべての対象が新バージョンなら完了。
	return updating > 0, nil
}

func (r *DatabaseReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	var db examplev1.Database
	if err := r.Get(ctx, req.NamespacedName, &db); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// 運用者が緊急停止をかけたら reconcile をスキップ。
	if db.Annotations["example.com/reconcile-paused"] == "true" {
		ctrl.LoggerFrom(ctx).Info("reconcile paused by annotation")
		return ctrl.Result{}, nil
	}
	// ...
	return ctrl.Result{}, nil
}

[正しいロールバック順序 — ストレージバージョンを上げた場合]

  1. conversion webhook はそのまま維持(決して先に下げない)
  2. コントローラだけを以前のバージョンにロールバック
  3. 旧バージョンが認識するバージョンへ storage version を戻すか慎重に判断
  4. すでに v1 として保存された CR があれば migrator で旧バージョンに再保存が必要

func TestConversionRoundTrip(t *testing.T) {
	original := &v1beta1.Database{
		Spec: v1beta1.DatabaseSpec{
			Engine:    "postgres",
			Replicas:  3,
			StorageGB: 100,
		},
	}

	// v1beta1 -> v1 -> v1beta1
	hub := &v1.Database{}
	if err := original.ConvertTo(hub); err != nil {
		t.Fatal(err)
	}
	roundTripped := &v1beta1.Database{}
	if err := roundTripped.ConvertFrom(hub); err != nil {
		t.Fatal(err)
	}

	// v1beta1 が表現できるフィールドは無損失でなければならない。
	if roundTripped.Spec.StorageGB != original.Spec.StorageGB {
		t.Errorf("StorageGB lost: got %d want %d",
			roundTripped.Spec.StorageGB, original.Spec.StorageGB)
	}
}

mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
	Scheme: scheme,
	Cache: cache.Options{
		// このコントローラは canary ネームスペースだけを監視する。
		DefaultNamespaces: map[string]cache.Config{
			"team-canary": {},
		},
	},
	// カナリアは別のリーダー選出 ID を使い、既存コントローラと衝突しないようにする。
	LeaderElection:   true,
	LeaderElectionID: "db-operator-canary.example.com",
})

[互換性点検の順序]

  1. go.mod で controller-runtime, client-go, apimachinery のバージョンを整列
  2. 対象クラスタの K8s バージョンがサポート範囲内か確認
  3. 廃止予定(deprecated)API を使っていないか点検(例: 旧 admissionregistration バージョン)
  4. controller-tools を上げた後 CRD を再生成しスキーマ diff を確認
  5. envtest で統合テストを回し回帰を確認

package main

import (
	"context"
	"time"

	"golang.org/x/time/rate"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// migrateStorageVersion はすべての Database CR を一度ずつ no-op 更新し、
// 現在の storage version として再保存する。
func migrateStorageVersion(ctx context.Context, c client.Client) error {
	// 毎秒 10 件に制限(API サーバー/webhook を保護)。
	limiter := rate.NewLimiter(rate.Limit(10), 1)

	var continueToken string
	for {
		var list examplev1.DatabaseList
		opts := []client.ListOption{client.Limit(100)}
		if continueToken != "" {
			opts = append(opts, client.Continue(continueToken))
		}
		if err := c.List(ctx, &list, opts...); err != nil {
			return err
		}

		for i := range list.Items {
			if err := limiter.Wait(ctx); err != nil {
				return err
			}
			db := &list.Items[i]

			// no-op annotation のトグルで update を誘発し再保存。
			if db.Annotations == nil {
				db.Annotations = map[string]string{}
			}
			db.Annotations["example.com/storage-migrated-at"] =
				time.Now().UTC().Format(time.RFC3339)

			if err := c.Update(ctx, db); err != nil {
				// 競合は次のラウンドで再試行されるようログだけ残す。
				continue
			}
		}

		if list.Continue == "" {
			break
		}
		continueToken = list.Continue
	}
	return nil
}

var _ = metav1.ObjectMeta{} // import 使用の明示

1. レート制限は必須          - API サーバー/webhook の過負荷を防止(毎秒 N 件)
2. ページネーション         - List に Limit + Continue を使い、メモリ爆発を防止
3. 冪等性                   - 途中で落ちても再実行すればよいよう設計
4. 競合の許容               - 同時更新の競合は次のラウンドで再試行
5. 進行の可視化            - 処理件数/失敗件数をメトリクスで公開
6. webhook ヘルス確認       - 変換が失敗したら即座に中断

[障害対応プレイブック]

症状の検知
   |
   v
1) ロールアウトを即停止
   - CR に reconcile-paused=true annotation(拡散を遮断)
   - または Rollout.Paused=true で段階的移行を中断
   |
   v
2) 影響範囲の把握
   - status conditions / observedGeneration でどこまで進んだか確認
   - 新バージョンに変わったワークロード数を集計
   |
   v
3) 可逆性の判断
   - ストレージバージョンをまだ上げていないか? -> コントローライメージのロールバックで十分
   - すでに上げたか? -> conversion webhook を維持しつつ慎重にロールバック
   |
   v
4) コントローラのロールバック
   - Deployment イメージを直前の正常タグに
   - ただし CRD/webhook をむやみに一緒に下げない
   |
   v
5) データ整合性の確認
   - 消失しうるフィールド(ClassName など)が影響を受けたか点検
   - 必要ならバックアップから復旧
   |
   v
6) 事後分析
   - round-trip テストの欠落、webhook ヘルス未点検などの原因を記録

[アップグレード前]
[] go.mod の controller-runtime/client-go/apimachinery バージョン整列を確認
[] 対象 K8s バージョンがサポート範囲(1.36)内か確認
[] CRD 再生成後、既存とのスキーマ diff を確認
[] conversion 関数の round-trip テスト通過
[] envtest 統合テスト通過
[] リーダー選出 + LeaderElectionReleaseOnCancel 設定を確認
[] Deployment maxUnavailable=0 / maxSurge=1 を確認
[] ロールバック手順の文書化 + バックアップ確認
[] 新スキーマが旧バージョンのコントローラと互換か(カナリア可否)を判断

[アップグレード中]
[] 新バージョンを served:true, storage:false で先に追加
[] conversion webhook のヘルス確認
[] コントローラのローリング後、リーダー引き継ぎが正常動作するか確認
[] 管理ワークロードは partition で漸進的に拡大
[] status conditions / observedGeneration で進行を追跡

[アップグレード後]
[] storage version の切り替えはコントローラ安定後に別途
[] storage version migrator で既存 CR を再保存(レート制限)
[] 旧バージョンを deprecated 表記し猶予期間を付与
[] メトリクス(reconcile エラー率、webhook 遅延)が正常範囲か確認
[] 参照のなくなった旧バージョンを served:false -> 削除