Operator の可観測性 — メトリクス、イベント、ロギング、そして SLO

可観測性の五つのシグナル
┌──────────────┬───────────────────────────────────────────┐
│ シグナル       │ 誰のためのものか                            │
├──────────────┼───────────────────────────────────────────┤
│ メトリクス      │ SRE / 運用者 — 傾向、アラート、SLO           │
│ イベント        │ エンドユーザー — kubectl describe の文脈      │
│ ロギング        │ 開発者 / 運用者 — 単一 reconcile の詳細      │
│ トレーシング     │ 開発者 — 複数段階にまたがる遅延の分析         │
│ status/条件     │ エンドユーザー / コントローラー — 宣言的状態   │
└──────────────┴───────────────────────────────────────────┘

controller_runtime_reconcile_total{controller, result}
    reconcile 呼び出し回数。result ラベルは success / error /
    requeue / requeue_after で区別される。成功率計算の分母・分子
    になる。

controller_runtime_reconcile_errors_total{controller}
    Reconcile がエラーを返した回数。この値が着実に増えるなら、
    何かが繰り返し失敗しているという意味。

controller_runtime_reconcile_time_seconds{controller}
    reconcile 1回の処理時間のヒストグラム。_bucket / _sum /
    _count 系列を生成し、分位数(p50/p95/p99)の計算に使う。

controller_runtime_active_workers{controller}
    現在 reconcile を同時実行中のワーカー数。

controller_runtime_max_concurrent_reconciles{controller}
    設定された最大同時 reconcile 数(MaxConcurrentReconciles)。

workqueue_depth{name}
    キューで待機中の項目数。継続的に高いと、コントローラーが
    入ってくるイベント速度に追いつけていないという意味。

workqueue_adds_total{name}
    キューに追加された累積項目数。rate() で流入速度を見る。

workqueue_queue_duration_seconds{name}
    項目がキューに滞在した時間(ヒストグラム)。処理開始までの待ち。

workqueue_work_duration_seconds{name}
    項目処理にかかった時間(ヒストグラム)。reconcile の実作業時間。

workqueue_retries_total{name}
    キューから再試行された累積回数。エラーによる再試行が多いと増加。

workqueue_unfinished_work_seconds{name}
    まだ処理されていない作業がキューに溜まっている時間。

workqueue_longest_running_processor_seconds{name}
    最も長く実行中の処理タスクの時間。止まった reconcile の検知に有用。

イベント発生 -> predicate 通過 -> workqueue.Add()
                                    │
                              [workqueue_adds_total]
                                    │
                                    ▼
                            ┌──────────────┐
                            │   workqueue  │  <- [workqueue_depth]
                            │  (待機項目)    │  <- [queue_duration]
                            └──────┬───────┘
                                   │ Get()
                                   ▼
                            Reconcile() 実行
                              │  <- [active_workers]
                              │  <- [work_duration / reconcile_time]
                              ▼
                  成功 -> Forget()        エラー -> AddRateLimited()
              [reconcile_total            [reconcile_errors_total
               result=success]             workqueue_retries_total]

import (
	"sigs.k8s.io/controller-runtime/pkg/manager"
	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
	"sigs.k8s.io/controller-runtime/pkg/metrics/filters"
)

func newManager(cfg *rest.Config) (manager.Manager, error) {
	return ctrl.NewManager(cfg, ctrl.Options{
		Metrics: metricsserver.Options{
			BindAddress:   ":8443",
			SecureServing: true,
			// メトリクスエンドポイントに認証・認可を適用する。
			// かつての kube-rbac-proxy サイドカーの役割を代替する。
			FilterProvider: filters.WithAuthenticationAndAuthorization,
		},
	})
}

package controller

import (
	"github.com/prometheus/client_golang/prometheus"
	"sigs.k8s.io/controller-runtime/pkg/metrics"
)

var (
	// 管理中リソースの現在の状態別カウント(ゲージ)
	managedResources = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "myoperator_managed_resources",
			Help: "現在管理中のリソース数(phase ラベル別)",
		},
		[]string{"namespace", "phase"},
	)

	// 外部 API 呼び出し結果のカウンター
	externalAPICalls = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "myoperator_external_api_calls_total",
			Help: "外部 API 呼び出し回数(結果ラベル別)",
		},
		[]string{"endpoint", "result"},
	)

	// 最後の成功同期からの経過時間を推論するためのタイムスタンプゲージ
	lastSyncTimestamp = prometheus.NewGaugeVec(
		prometheus.GaugeOpts{
			Name: "myoperator_last_successful_sync_timestamp_seconds",
			Help: "リソースごとの最後の成功同期の Unix タイムスタンプ",
		},
		[]string{"namespace", "name"},
	)

	// reconcile 段階別の所要時間ヒストグラム
	reconcileStageDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "myoperator_reconcile_stage_duration_seconds",
			Help:    "reconcile 内部段階別の所要時間",
			Buckets: prometheus.DefBuckets,
		},
		[]string{"stage"},
	)
)

func init() {
	// controller-runtime のグローバルレジストリに登録する。
	// こうするとデフォルトメトリクスと同じ /metrics で公開される。
	metrics.Registry.MustRegister(
		managedResources,
		externalAPICalls,
		lastSyncTimestamp,
		reconcileStageDuration,
	)
}

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

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

	// 段階別タイマー開始
	stageTimer := prometheus.NewTimer(
		reconcileStageDuration.WithLabelValues("fetch_external"),
	)
	state, err := r.External.FetchState(ctx, obj.Spec.ID)
	stageTimer.ObserveDuration()

	if err != nil {
		externalAPICalls.WithLabelValues("fetch", "error").Inc()
		log.Error(err, "外部状態の取得に失敗")
		return ctrl.Result{}, err
	}
	externalAPICalls.WithLabelValues("fetch", "success").Inc()

	// desired state へ収束させる適用段階
	applyTimer := prometheus.NewTimer(
		reconcileStageDuration.WithLabelValues("apply"),
	)
	phase, err := r.applyDesiredState(ctx, &obj, state)
	applyTimer.ObserveDuration()
	if err != nil {
		return ctrl.Result{}, err
	}

	// ゲージ更新
	managedResources.WithLabelValues(obj.Namespace, phase).Set(1)
	lastSyncTimestamp.WithLabelValues(obj.Namespace, obj.Name).
		SetToCurrentTime()

	return ctrl.Result{RequeueAfter: 5 * time.Minute}, nil
}

// finalizer または NotFound 処理時に、該当ラベルの系列を片付ける。
managedResources.DeleteLabelValues(obj.Namespace, oldPhase)
lastSyncTimestamp.DeleteLabelValues(obj.Namespace, obj.Name)

apiVersion: monitoring.coreos.com/v1
kind: ServiceMonitor
metadata:
  name: myoperator-controller-manager-metrics
  namespace: myoperator-system
  labels:
    app.kubernetes.io/name: myoperator
    release: kube-prometheus-stack
spec:
  selector:
    matchLabels:
      control-plane: controller-manager
  endpoints:
    - port: https
      scheme: https
      path: /metrics
      interval: 30s
      bearerTokenSecret:
        name: metrics-reader-token
        key: token
      tlsConfig:
        insecureSkipVerify: true
  namespaceSelector:
    matchNames:
      - myoperator-system

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: myoperator-metrics-reader
rules:
  - nonResourceURLs:
      - /metrics
    verbs:
      - get

scrape_configs:
  - job_name: myoperator
    scheme: https
    metrics_path: /metrics
    tls_config:
      insecure_skip_verify: true
    authorization:
      type: Bearer
      credentials_file: /var/run/secrets/tokens/metrics-token
    kubernetes_sd_configs:
      - role: endpoints
        namespaces:
          names:
            - myoperator-system
    relabel_configs:
      - source_labels: [__meta_kubernetes_endpoint_port_name]
        action: keep
        regex: https

sum(rate(controller_runtime_reconcile_total{controller="myresource", result!="error"}[5m]))
/
sum(rate(controller_runtime_reconcile_total{controller="myresource"}[5m]))

sum(rate(controller_runtime_reconcile_errors_total{controller="myresource"}[5m]))

histogram_quantile(
  0.95,
  sum(rate(controller_runtime_reconcile_time_seconds_bucket{controller="myresource"}[5m])) by (le)
)

workqueue_depth{name="myresource"}

histogram_quantile(
  0.95,
  sum(rate(workqueue_queue_duration_seconds_bucket{name="myresource"}[5m])) by (le)
)

time() - max by (namespace, name) (myoperator_last_successful_sync_timestamp_seconds)

type MyResourceReconciler struct {
	client.Client
	Scheme   *runtime.Scheme
	Recorder record.EventRecorder
}

// SetupWithManager で Recorder を注入する。
func (r *MyResourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
	r.Recorder = mgr.GetEventRecorderFor("myresource-controller")
	return ctrl.NewControllerManagedBy(mgr).
		For(&appsv1alpha1.MyResource{}).
		Complete(r)
}

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

	deployment, err := r.ensureDeployment(ctx, &obj)
	if err != nil {
		// ユーザーに見える Warning イベント:何が、なぜ失敗したか
		r.Recorder.Eventf(&obj, corev1.EventTypeWarning, "DeploymentFailed",
			"子 Deployment の作成に失敗: %v", err)
		return ctrl.Result{}, err
	}

	if deployment.Status.ReadyReplicas == *deployment.Spec.Replicas {
		// 正常状態を知らせる Normal イベント
		r.Recorder.Eventf(&obj, corev1.EventTypeNormal, "Ready",
			"すべてのレプリカ(%d個)が準備完了しました", deployment.Status.ReadyReplicas)
	}

	return ctrl.Result{}, nil
}

イベント記録の原則
- 状態の「遷移」だけに記録せよ。毎 reconcile ごとに同じイベントを
  打つと、etcd にとってもユーザーにとってもノイズになる。
- Reason は PascalCase の語で、機械が grep できるようにせよ。
  (例: DeploymentFailed, Ready, Scaling, QuotaExceeded)
- メッセージは人間のための文に。次の行動を示唆できればなお良い。
- イベントはデフォルトの TTL(約1時間)で消える。永続的な状態は
  イベントではなく status.conditions に置け。
- Warning は本当に注意が必要なときだけ。乱発すると鈍る。

import (
	logf "sigs.k8s.io/controller-runtime/pkg/log"
)

func (r *MyResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	// コンテキストからロガーを取得する。controller-runtime が
	// すでに reconcileID, controller, namespace, name などを注入済み。
	log := logf.FromContext(ctx)

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

	// この reconcile の間ずっと付いて回る共通フィールドを束ねる。
	log = log.WithValues("generation", obj.Generation, "phase", obj.Status.Phase)

	log.Info("reconcile 開始")

	if err := r.doWork(ctx, &obj); err != nil {
		// エラーは Error レベルで。err を第一引数に渡すと構造化される。
		log.Error(err, "作業の実行に失敗", "step", "doWork")
		return ctrl.Result{}, err
	}

	// 詳細なデバッグ情報は V-level を活用し、普段はオフにする。
	log.V(1).Info("desired state の計算完了", "replicas", obj.Spec.Replicas)

	log.Info("reconcile 完了")
	return ctrl.Result{}, nil
}

ロギングレベルガイド
  Error          : reconcile が失敗し再試行される状況。常に出力。
  Info / V(0)    : 状態遷移、reconcile 開始/完了のようなマクロな流れ。
  V(1)           : 判断の根拠 — なぜこの経路を選んだか。
  V(2) 以上       : ループ内部、個別項目単位の非常に詳細な流れ。

何をロギングしないか
  - Secret の値、トークン、パスワード(絶対禁止)
  - 毎 reconcile ごとの同じ「変化なし」ログ(ノイズ)
  - 巨大なオブジェクト全体のダンプ(必要なフィールドだけ WithValues で)

import (
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
)

var tracer = otel.Tracer("myoperator/controller")

func (r *MyResourceReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	// reconcile 全体を包むルートスパンを開始する。
	ctx, span := tracer.Start(ctx, "Reconcile",
		trace.WithAttributes(
			attribute.String("resource.namespace", req.Namespace),
			attribute.String("resource.name", req.Name),
		),
	)
	defer span.End()

	var obj appsv1alpha1.MyResource
	if err := r.Get(ctx, req.NamespacedName, &obj); err != nil {
		span.RecordError(err)
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// 外部呼び出しを子スパンで包む。ctx をそのまま渡して
	// 親子関係が伝播するようにする。
	if err := r.fetchExternal(ctx, &obj); err != nil {
		span.SetStatus(codes.Error, "external fetch failed")
		span.RecordError(err)
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

func (r *MyResourceReconciler) fetchExternal(ctx context.Context, obj *appsv1alpha1.MyResource) error {
	// 同じ ctx から子スパンを開始すると自動的に親に紐づく。
	ctx, span := tracer.Start(ctx, "fetchExternal")
	defer span.End()
	span.SetAttributes(attribute.String("external.id", obj.Spec.ID))
	return r.External.Call(ctx, obj.Spec.ID)
}

1回の reconcile トレース
Reconcile [120ms] ──────────────────────────────────────────
  ├─ Get(MyResource) [3ms] ─
  ├─ fetchExternal [95ms] ────────────────────────────
  │     └─ HTTP GET cloud-api [92ms] ──────────────────
  └─ applyDesiredState [20ms] ───────
        └─ Patch(Deployment) [18ms] ──────

import (
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/api/meta"
)

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

	ready, reason, msg := r.evaluateReadiness(ctx, &obj)

	// meta.SetStatusCondition は同じ Type の条件を更新するか、
	// なければ追加する。LastTransitionTime を自動で管理する。
	meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{
		Type:               "Ready",
		Status:             conditionStatus(ready),
		Reason:             reason,             // PascalCase、機械判読可能
		Message:            msg,                // 人間のための説明
		ObservedGeneration: obj.Generation,     // どの世代を評価したか
	})

	// status サブリソースを更新する。
	if err := r.Status().Update(ctx, &obj); err != nil {
		return ctrl.Result{}, err
	}
	return ctrl.Result{}, nil
}

func conditionStatus(ok bool) metav1.ConditionStatus {
	if ok {
		return metav1.ConditionTrue
	}
	return metav1.ConditionFalse
}

status.conditions の設計原則
- Type は名詞的な能力/状態にせよ: Ready, Available,
  Progressing, Degraded など。標準的な意味に従うとツール互換性が良い。
- Status は True/False/Unknown の三つだけ。
- Reason は PascalCase の語。Message は人間のための文。
- ObservedGeneration を必ず埋めよ。適用有無判断の核心。
- conditions は冪等に set せよ。毎回 append してはいけない。

Operator のための三つの SLI / SLO 例
┌────────────────┬─────────────────────────────┬──────────────┐
│ SLI            │ 定義                          │ SLO 目標(例)  │
├────────────────┼─────────────────────────────┼──────────────┤
│ reconcile      │ エラー以外の reconcile /        │ 30日 99.5%   │
│ 成功率          │ 全 reconcile                  │              │
├────────────────┼─────────────────────────────┼──────────────┤
│ reconcile      │ reconcile p99 処理時間         │ p99 < 2s     │
│ 遅延           │                              │ (5分ウィンドウ)│
├────────────────┼─────────────────────────────┼──────────────┤
│ 新鮮さ          │ 最後の成功同期からの経過時間     │ 95% が        │
│ (freshness)    │                              │ < 10分        │
└────────────────┴─────────────────────────────┴──────────────┘

1 - (
  sum(rate(controller_runtime_reconcile_errors_total{controller="myresource"}[30d]))
  /
  sum(rate(controller_runtime_reconcile_total{controller="myresource"}[30d]))
)

apiVersion: monitoring.coreos.com/v1
kind: PrometheusRule
metadata:
  name: myoperator-slo-alerts
  namespace: myoperator-system
  labels:
    release: kube-prometheus-stack
spec:
  groups:
    - name: myoperator.slo
      rules:
        - alert: ReconcileErrorRateHigh
          expr: |
            sum(rate(controller_runtime_reconcile_errors_total{controller="myresource"}[5m]))
            /
            sum(rate(controller_runtime_reconcile_total{controller="myresource"}[5m]))
            > 0.05
          for: 10m
          labels:
            severity: warning
          annotations:
            summary: "MyResource reconcile エラー率が 5% を超えました"
            description: "直近5分間で reconcile エラー率が SLO しきい値を超えました。"

        - alert: ReconcileLatencyHigh
          expr: |
            histogram_quantile(0.99,
              sum(rate(controller_runtime_reconcile_time_seconds_bucket{controller="myresource"}[5m])) by (le)
            ) > 2
          for: 15m
          labels:
            severity: warning
          annotations:
            summary: "MyResource reconcile p99 遅延が 2秒を超えました"

        - alert: ResourceStaleness
          expr: |
            time() - max by (namespace, name) (
              myoperator_last_successful_sync_timestamp_seconds
            ) > 900
          for: 5m
          labels:
            severity: critical
          annotations:
            summary: "リソースが15分以上同期されていません"
            description: "特定の MyResource が SLO 新鮮さ目標に違反しています。"

        - alert: WorkqueueBacklog
          expr: workqueue_depth{name="myresource"} > 50
          for: 10m
          labels:
            severity: warning
          annotations:
            summary: "workqueue の滞留が発生しました"
            description: "コントローラーが入ってくるイベントに追いついていません。"

ランブック:「reconcile がトリガーされない」の診断

1) まずメトリクスで reconcile が回っているか確認
   - controller_runtime_reconcile_total{controller="..."} が
     増えているか?
   - 全く増えない -> コントローラーがイベントを受け取っていない。
     以下の 2-6 へ。
   - 増えてはいるが errors_total も増える -> reconcile は呼ばれて
     いるが失敗。ログ/イベントで原因を追跡(このランブックの対象外)。

2) コントローラーマネージャーは生きていてリーダーか
   - Pod は Running か? クラッシュループではないか?
   - リーダー選出(leader election)使用時:リーダーでない複製は
     reconcile を回さない。リーダーロックの Lease を確認。
       kubectl get lease -n myoperator-system
   - リーダー Pod のログを見ているか確認(別の Pod のログを見ると
     「静か」に見えることがある)。

3) 情報がキャッシュに入ってきているか(informer / RBAC)
   - コントローラー SA が対象リソースを watch/list/get する RBAC が
     あるか?なければ informer が静かに失敗するか権限エラー。
       kubectl auth can-i list myresources \
         --as=system:serviceaccount:myoperator-system:controller-manager
   - ログに "failed to list" / "forbidden" 系のエラーがないか。

4) predicate がイベントをフィルタしていないか
   - SetupWithManager の WithEventFilter / Owns / Watches に
     付いた predicate がその変更を通過させるか?
   - 例: GenerationChangedPredicate は metadata/label だけ変わった
     更新を無視する -> status/annotation 変更だけでは reconcile が
     回らない。
   - 疑わしければ predicate を一時的に外して再現してみる。

5) イベントソース(watch)が実際に登録されているか
   - For()/Owns()/Watches() に対象の型が抜けていないか。
   - 別の namespace / クラスタスコープのリソースを見るのに、
     マネージャーが namespace でキャッシュを制限していないか
     (Cache.DefaultNamespaces 設定を確認)。

6) オブジェクトは実際に変わったか
   - kubectl get myresource -o yaml で generation と
     status.observedGeneration を比較。
   - resourceVersion がそのままなら apiserver から見て変更が
     ない -> イベント自体がない。

決定木の要約
  reconcile_total が増えない?
    ├─ Pod/リーダーの問題   -> 2)
    ├─ RBAC 不足           -> 3)
    ├─ predicate フィルタ   -> 4)
    └─ watch 抜け          -> 5)

日常点検(ダッシュボードで毎日/週次)
  - reconcile 成功率が SLO(例: 99.5%)を上回っているか
  - reconcile p99 遅延が目標(例: 2s)を下回っているか
  - workqueue_depth が 0 近くに収束しているか
  - 新鮮さアラートが静かか

デプロイ後の確認
  - reconcile_errors_total がデプロイ直後に跳ねていないか
  - active_workers が max_concurrent_reconciles に
    張り付き続けていないか(飽和シグナル)
  - 新バージョンのログに新規 Error がないか

障害対応
  - エラー率急増 -> 最近のデプロイのロールバックが第一候補。
    外部依存の状態も同時に確認。
  - 遅延急増 -> トレースでどのスパンが遅いかを特定。
    外部 API のレート制限 / タイムアウトを疑う。
  - workqueue 滞留 -> MaxConcurrentReconciles を引き上げ、
    predicate で不要イベントを遮断、reconcile を軽量化。
  - 特定のリソースだけ stale -> 当該オブジェクトを kubectl describe
    して Warning イベントと conditions を確認。observedGeneration 比較。

容量計画
  - 管理リソース数の増加に伴う reconcile_total rate の傾向
  - メトリクスのカーディナリティ(特に name ラベル)の増加傾向を監視
  - Prometheus TSDB のメモリ/ディスク使用量