OpenTelemetry Operator CRD編
現在、OpenTelemetryCollectorは OpenTelemetry Operatorの中核となるカスタムリソースです。OpenTelemetry Collector のデプロイ・管理・自動スケーリング・アップグレードなどを宣言的に行えます。
code:md
┌─────────────────────────────────────────────────────────────────────────────────┐
│ OpenTelemetry Operator Architecture │
└─────────────────────────────────────────────────────────────────────────────────┘
User Kubernetes API Server
│ │
│ kubectl apply │
▼ ▼
┌─────────────────┐ ┌─────────────────────┐
│ OpenTelemetry │ ◄─── Webhook ────► │ Admission Controller│
│ Collector CR │ Validation/ │ (Mutating/Validating│
│ (v1beta1) │ Defaulting │ Webhooks) │
└────────┬────────┘ └─────────────────────┘
│
│ Watch
▼
┌─────────────────────────────────────────────────────────────┐
│ OpenTelemetryCollectorReconciler │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Reconcile Loop │ │
│ │ 1. Get CR Instance │ │
│ │ 2. Check Deletion / ManagementState │ │
│ │ 3. Version Upgrade (if needed) │ │
│ │ 4. BuildCollector() → Generate Manifests │ │
│ │ 5. reconcileDesiredObjects() → Create/Update/Delete │ │
│ │ 6. Update Status │ │
│ └─────────────────────────────────────────────────────┘ │
└────────────────────────────┬────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Generated Resources │
├─────────────────┬─────────────────┬─────────────────┬───────────────────────┤
│ Workload │ Networking │ Monitoring │ RBAC │
├─────────────────┼─────────────────┼─────────────────┼───────────────────────┤
│ • Deployment │ • Service │ • ServiceMonitor│ • ServiceAccount │
│ • DaemonSet │ • HeadlessService│• PodMonitor │ • ClusterRole │
│ • StatefulSet │ • Ingress │ │ • ClusterRoleBinding │
│ │ • Route (OCP) │ │ │
│ │ • NetworkPolicy │ │ │
├─────────────────┼─────────────────┼─────────────────┼───────────────────────┤
│ Scaling │ Config │ Disruption │ Sub-CR │
├─────────────────┼─────────────────┼─────────────────┼───────────────────────┤
│ • HPA │ • ConfigMap │ • PDB │ • TargetAllocator │
└─────────────────┴─────────────────┴─────────────────┴───────────────────────┘
Mode別にCollectorが生成されるようになってます。
code:go
func Build(params manifests.Params) ([]client.Object, error) {
var resourceManifests []client.Object
switch params.OtelCol.Spec.Mode {
case v1beta1.ModeDeployment:
manifestFactories = append(manifestFactories, manifests.Factory(Deployment))
manifestFactories = append(manifestFactories, manifests.Factory(PodDisruptionBudget))
case v1beta1.ModeStatefulSet:
manifestFactories = append(manifestFactories, manifests.Factory(StatefulSet))
manifestFactories = append(manifestFactories, manifests.Factory(PodDisruptionBudget))
case v1beta1.ModeDaemonSet:
manifestFactories = append(manifestFactories, manifests.Factory(DaemonSet))
case v1beta1.ModeSidecar:
params.Log.V(5).Info("not building sidecar...")
}
manifestFactories = append(manifestFactories, []manifests.K8sManifestFactorymanifests.Params{ manifests.Factory(ConfigMap),
manifests.Factory(HorizontalPodAutoscaler),
manifests.Factory(ServiceAccount),
manifests.Factory(Service),
manifests.Factory(HeadlessService),
manifests.Factory(MonitoringService),
manifests.Factory(ExtensionService),
manifests.Factory(Ingress),
manifests.Factory(NetworkPolicy),
manifests.Factory(TargetAllocator),
}...)
if params.OtelCol.Spec.Observability.Metrics.EnableMetrics {
if params.OtelCol.Spec.Mode == v1beta1.ModeSidecar {
manifestFactories = append(manifestFactories, manifests.Factory(PodMonitor))
} else {
manifestFactories = append(manifestFactories, manifests.Factory(ServiceMonitor), manifests.Factory(ServiceMonitorMonitoring))
}
}
Deployment Mode
Replicas設定可能
DeploymentUpdateStrategyでアップデート戦略を指定
PodDisruptionBudget生成（デフォルト: MaxUnavailable=1）
code:go
func Deployment(params manifests.Params) (*appsv1.Deployment, error) {
name := naming.Collector(params.OtelCol.Name)
labels := manifestutils.Labels(params.OtelCol.ObjectMeta, name, params.OtelCol.Spec.Image, ComponentOpenTelemetryCollector, params.Config.LabelsFilter)
annotations, err := manifestutils.Annotations(params.OtelCol, params.Config.AnnotationsFilter)
// ...
return &appsv1.Deployment{
Spec: appsv1.DeploymentSpec{
Replicas: manifestutils.GetDesiredReplicas(params.OtelCol),
Strategy: params.OtelCol.Spec.DeploymentUpdateStrategy,
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
Containers: append([]corev1.Container{Container(params.Config, params.Log, params.OtelCol, true)}, params.OtelCol.Spec.AdditionalContainers...),
Volumes: Volumes(params.Config, params.OtelCol),
// ... PodSpec共通フィールド
},
},
},
}, nil
}
StatefulSet
HeadlessServiceと連携
VolumeClaimTemplatesでPersistentVolumeを自動作成
PodManagementPolicy: Parallel
code:go
func StatefulSet(params manifests.Params) (*appsv1.StatefulSet, error) {
// ...
serviceName := params.OtelCol.Spec.ServiceName
if serviceName == "" {
serviceName = naming.HeadlessService(params.OtelCol.Name)
}
return &appsv1.StatefulSet{
Spec: appsv1.StatefulSetSpec{
ServiceName: serviceName, // HeadlessServiceと連携
PodManagementPolicy: "Parallel", // 並列起動
VolumeClaimTemplates: VolumeClaimTemplates(params.OtelCol),
PersistentVolumeClaimRetentionPolicy: params.OtelCol.Spec.PersistentVolumeClaimRetentionPolicy,
// ...
},
}, nil
}
DaemonSet
全ノード（または一部ノード）で1 Podずつ実行
DaemonSetUpdateStrategyでローリングアップデート設定
Replicas指定不可
Service生成時: InternalTrafficPolicy: Local
code:go
func DaemonSet(params manifests.Params) (*appsv1.DaemonSet, error) {
return &appsv1.DaemonSet{
Spec: appsv1.DaemonSetSpec{
UpdateStrategy: params.OtelCol.Spec.DaemonSetUpdateStrategy,
Template: corev1.PodTemplateSpec{
Spec: corev1.PodSpec{
// ... Deployment/StatefulSetと同様
},
},
},
}, nil
}
Sidecar Mode
Webhookによる Pod への自動注入
ConfigMapではなく環境変数(OTEL_CONFIG)で設定を渡す
Native Sidecar対応（K8s 1.28+）: InitContainers + RestartPolicy: Always
Operatorはワークロードを作成しない（Build()でスキップ）
code:go
func add(cfg config.Config, logger logr.Logger, otelcol v1beta1.OpenTelemetryCollector, pod corev1.Pod, attributes []corev1.EnvVar) (corev1.Pod, error) {
otelColCfg, err := collector.ReplaceConfig(otelcol, nil)
if err != nil {
return pod, err
}
container := collector.Container(cfg, logger, otelcol, false)
container.Args = append(container.Args, fmt.Sprintf("--config=env:%s", confEnvVar))
container.Env = append(container.Env, corev1.EnvVar{Name: confEnvVar, Value: otelColCfg})
if !hasResourceAttributeEnvVar(container.Env) {
container.Env = append(container.Env, attributes...)
}
pod.Spec.InitContainers = append(pod.Spec.InitContainers, otelcol.Spec.InitContainers...)
if cfg.Internal.NativeSidecarSupport {
policy := corev1.ContainerRestartPolicyAlways
container.RestartPolicy = &policy
// NOTE: Use StartupProbe if available, otherwise fallback to ReadinessProbe as startup probe.
if container.StartupProbe == nil {
container.StartupProbe = container.ReadinessProbe
}
pod.Spec.InitContainers = append(pod.Spec.InitContainers, container)
} else {
pod.Spec.Containers = append(pod.Spec.Containers, container)
}
pod.Spec.Volumes = append(pod.Spec.Volumes, otelcol.Spec.Volumes...)
for _, cm := range otelcol.Spec.ConfigMaps {
pod.Spec.Volumes = append(
pod.Spec.Volumes,
corev1.Volume{
Name: naming.ConfigMapExtra(cm.Name),
VolumeSource: corev1.VolumeSource{
ConfigMap: &corev1.ConfigMapVolumeSource{
LocalObjectReference: corev1.LocalObjectReference{
Name: cm.Name,
},
},
},
},
)
}
if pod.Labels == nil {
pod.Labels = mapstringstring{} }
pod.LabelsinjectedLabel = naming.Truncate("%s.%s", 63, otelcol.Namespace, otelcol.Name) return pod, nil
}
これらのmodeをユースケースで使い分けるのが基本になると思います。
table:ユースケース
OTLPゲートウェイ Deployment スケーラブル、ステートレス
ログ収集（ファイルテール） DaemonSet 各ノードのログにアクセス必要
Prometheus互換スクレイプ StatefulSet TargetAllocator連携、重複防止
アプリトレース収集 Sidecar アプリ単位、自動注入
Kubernetes Events収集 Deployment (1 replica) クラスタワイド、単一で十分
ノードメトリクス収集 DaemonSet hostmetrics receiver使用
テレメトリ集約/加工 Deployment 処理能力でスケール
まとめ
明日はAutoInstrumentionを見ます。