OpenTelemetry Collector本番運用ガイド：パイプラインアーキテクチャ、カスタムプロセッサ開発、スケーリング戦略

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  prometheus:
    config:
      scrape_configs:
        - job_name: 'kubernetes-pods'
          kubernetes_sd_configs:
            - role: pod
          relabel_configs:
            - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
              action: keep
              regex: true
  filelog:
    include:
      - /var/log/pods/*/*/*.log
    operators:
      - type: json_parser
        timestamp:
          parse_from: attributes.time
          layout: '%Y-%m-%dT%H:%M:%S.%LZ'

processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 1800
    spike_limit_mib: 400
  batch:
    timeout: 1s
    send_batch_size: 1024
    send_batch_max_size: 2048
  attributes:
    actions:
      - key: environment
        value: production
        action: upsert
      - key: collector.version
        value: '0.98.0'
        action: insert
  resource:
    attributes:
      - key: service.namespace
        value: platform
        action: upsert

exporters:
  otlp/tempo:
    endpoint: tempo-distributor.observability:4317
    tls:
      insecure: true
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 300s
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 5000
  prometheusremotewrite:
    endpoint: http://mimir-distributor.observability:9009/api/v1/push
    resource_to_telemetry_conversion:
      enabled: true
  otlp/loki:
    endpoint: loki-distributor.observability:3100
    tls:
      insecure: true

connectors:
  spanmetrics:
    histogram:
      explicit:
        buckets: [5ms, 10ms, 25ms, 50ms, 100ms, 250ms, 500ms, 1s, 2.5s, 5s, 10s]
    dimensions:
      - name: http.method
      - name: http.status_code
    metrics_flush_interval: 15s

service:
  telemetry:
    logs:
      level: info
    metrics:
      address: 0.0.0.0:8888
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [otlp/tempo, spanmetrics]
    metrics:
      receivers: [otlp, prometheus, spanmetrics]
      processors: [memory_limiter, batch]
      exporters: [prometheusremotewrite]
    logs:
      receivers: [otlp, filelog]
      processors: [memory_limiter, attributes, batch]
      exporters: [otlp/loki]

processors:
  batch:
    # この数のアイテムが蓄積されたらバッチを送信
    send_batch_size: 1024
    # 過大なバッチを防ぐハードキャップ
    send_batch_max_size: 2048
    # この期間後にバッファされたものを送信
    timeout: 1s

# Agentティア構成：トレースIDでGatewayにルーティング
exporters:
  loadbalancing:
    protocol:
      otlp:
        tls:
          insecure: true
    resolver:
      dns:
        hostname: otel-gateway-headless.observability
        port: 4317

# Gatewayティア構成：テールサンプリング
processors:
  tail_sampling:
    decision_wait: 10s
    num_traces: 100000
    expected_new_traces_per_sec: 2000
    policies:
      # ポリシー1：エラートレースを常に保持
      - name: errors-policy
        type: status_code
        status_code:
          status_codes: [ERROR]
      # ポリシー2：遅いトレースを常に保持（2秒超）
      - name: latency-policy
        type: latency
        latency:
          threshold_ms: 2000
      # ポリシー3：重要サービスに関連するトレースを保持
      - name: critical-services
        type: string_attribute
        string_attribute:
          key: service.name
          values: [payment-service, auth-service, order-service]
      # ポリシー4：それ以外の確率的サンプリング
      - name: probabilistic-policy
        type: probabilistic
        probabilistic:
          sampling_percentage: 5
      # 複合ポリシー：上記を組み合わせる
      - name: composite-policy
        type: composite
        composite:
          max_total_spans_per_second: 5000
          policy_order: [errors-policy, latency-policy, critical-services, probabilistic-policy]
          rate_allocation:
            - policy: errors-policy
              percent: 40
            - policy: latency-policy
              percent: 25
            - policy: critical-services
              percent: 25
            - policy: probabilistic-policy
              percent: 10

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, tail_sampling, batch]
      exporters: [otlp/tempo]

# Processorモジュールの作成
mkdir -p processor/deploymetadataprocessor
cd processor/deploymetadataprocessor

# Goモジュールの初期化
go mod init github.com/myorg/otelcol-custom/processor/deploymetadataprocessor
go get go.opentelemetry.io/collector/processor
go get go.opentelemetry.io/collector/pdata

// config.go
package deploymetadataprocessor

type Config struct {
    ClusterName    string `mapstructure:"cluster_name"`
    Region         string `mapstructure:"region"`
    DeploymentEnv  string `mapstructure:"deployment_env"`
    AddBuildInfo   bool   `mapstructure:"add_build_info"`
}

func createDefaultConfig() *Config {
    return &Config{
        ClusterName:   "default",
        Region:        "us-east-1",
        DeploymentEnv: "production",
        AddBuildInfo:  true,
    }
}

// factory.go
package deploymetadataprocessor

import (
    "context"

    "go.opentelemetry.io/collector/component"
    "go.opentelemetry.io/collector/consumer"
    "go.opentelemetry.io/collector/processor"
)

const (
    typeStr   = "deploy_metadata"
    stability = component.StabilityLevelDevelopment
)

func NewFactory() processor.Factory {
    return processor.NewFactory(
        component.MustNewType(typeStr),
        func() component.Config {
            return createDefaultConfig()
        },
        processor.WithTraces(createTracesProcessor, stability),
        processor.WithMetrics(createMetricsProcessor, stability),
        processor.WithLogs(createLogsProcessor, stability),
    )
}

func createTracesProcessor(
    ctx context.Context,
    set processor.Settings,
    cfg component.Config,
    next consumer.Traces,
) (processor.Traces, error) {
    pCfg := cfg.(*Config)
    return &deployMetadataTracesProcessor{
        config: pCfg,
        next:   next,
        logger: set.Logger,
    }, nil
}

// createMetricsProcessorとcreateLogsProcessorも同じパターンに従います

// processor_traces.go
package deploymetadataprocessor

import (
    "context"

    "go.opentelemetry.io/collector/consumer"
    "go.opentelemetry.io/collector/pdata/ptrace"
    "go.uber.org/zap"
)

type deployMetadataTracesProcessor struct {
    config *Config
    next   consumer.Traces
    logger *zap.Logger
}

func (p *deployMetadataTracesProcessor) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
    rss := td.ResourceSpans()
    for i := 0; i < rss.Len(); i++ {
        rs := rss.At(i)
        attrs := rs.Resource().Attributes()
        attrs.PutStr("deploy.cluster", p.config.ClusterName)
        attrs.PutStr("deploy.region", p.config.Region)
        attrs.PutStr("deploy.env", p.config.DeploymentEnv)
        if p.config.AddBuildInfo {
            attrs.PutStr("deploy.collector_build", "custom-v1.2.0")
        }
    }
    return p.next.ConsumeTraces(ctx, td)
}

func (p *deployMetadataTracesProcessor) Capabilities() consumer.Capabilities {
    return consumer.Capabilities{MutatesData: true}
}

func (p *deployMetadataTracesProcessor) Start(_ context.Context, _ component.Host) error {
    p.logger.Info("deploy_metadata processor started",
        zap.String("cluster", p.config.ClusterName),
        zap.String("region", p.config.Region))
    return nil
}

func (p *deployMetadataTracesProcessor) Shutdown(_ context.Context) error {
    return nil
}

# builder-config.yaml
dist:
  name: custom-otelcol
  description: デプロイメタデータProcessor付きカスタムOTel Collector
  output_path: ./dist
  version: 1.0.0

receivers:
  - gomod: go.opentelemetry.io/collector/receiver/otlpreceiver v0.98.0
  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/receiver/prometheusreceiver v0.98.0

processors:
  - gomod: go.opentelemetry.io/collector/processor/batchprocessor v0.98.0
  - gomod: go.opentelemetry.io/collector/processor/memorylimiterprocessor v0.98.0
  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.98.0
  - gomod: github.com/myorg/otelcol-custom/processor/deploymetadataprocessor v0.0.1

exporters:
  - gomod: go.opentelemetry.io/collector/exporter/otlpexporter v0.98.0

# カスタムCollectorのビルド
ocb --config builder-config.yaml

# 設定ファイルで実行
./dist/custom-otelcol --config otel-collector-config.yaml

# agent-daemonset.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: otel-agent
  namespace: observability
spec:
  selector:
    matchLabels:
      app: otel-agent
  template:
    metadata:
      labels:
        app: otel-agent
    spec:
      containers:
        - name: otel-agent
          image: otel/opentelemetry-collector-contrib:0.98.0
          args: ['--config=/etc/otel/config.yaml']
          resources:
            requests:
              cpu: 100m
              memory: 256Mi
            limits:
              cpu: 500m
              memory: 512Mi
          ports:
            - containerPort: 4317 # OTLP gRPC
            - containerPort: 4318 # OTLP HTTP
          volumeMounts:
            - name: config
              mountPath: /etc/otel
            - name: varlog
              mountPath: /var/log
              readOnly: true
      volumes:
        - name: config
          configMap:
            name: otel-agent-config
        - name: varlog
          hostPath:
            path: /var/log

# gateway-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: otel-gateway
  namespace: observability
spec:
  replicas: 3
  selector:
    matchLabels:
      app: otel-gateway
  template:
    metadata:
      labels:
        app: otel-gateway
    spec:
      containers:
        - name: otel-gateway
          image: custom-otelcol:1.0.0
          args: ['--config=/etc/otel/config.yaml']
          resources:
            requests:
              cpu: '1'
              memory: 4Gi
            limits:
              cpu: '2'
              memory: 6Gi
          ports:
            - containerPort: 4317
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: otel-gateway-hpa
  namespace: observability
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: otel-gateway
  minReplicas: 3
  maxReplicas: 12
  metrics:
    - type: Resource
      resource:
        name: cpu
        target:
          type: Utilization
          averageUtilization: 60
    - type: Resource
      resource:
        name: memory
        target:
          type: Utilization
          averageUtilization: 70

# 1. 現在のメモリ使用量を確認
kubectl top pods -n observability -l app=otel-gateway

# 2. OOMイベントを調査
kubectl describe pod otel-gateway-xxx -n observability | grep -A5 "Last State"

# 3. 即時緩和：decision_waitを短縮しレプリカを増加
kubectl scale deployment otel-gateway -n observability --replicas=6

# 4. 長期対策：テールサンプリング設定を調整
# decision_waitを10sから5sに短縮
# num_tracesを100000から200000に増加
# Podメモリリミットを8Giに、memory_limiterのlimit_mibを5600に増加

extensions:
  file_storage:
    directory: /var/otel/queue
    timeout: 10s
    compaction:
      on_start: true
      directory: /var/otel/queue/compaction

exporters:
  otlp/tempo:
    endpoint: tempo-distributor:4317
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
      storage: file_storage
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 60s
      max_elapsed_time: 600s

service:
  extensions: [file_storage]

processors:
  filter:
    metrics:
      exclude:
        match_type: regexp
        metric_names:
          - '.*_temp_.*'
        resource_attributes:
          - key: service.instance.id
            value: '.*debug.*'
  transform:
    metric_statements:
      - context: datapoint
        statements:
          - delete_key(attributes, "http.url")
          - delete_key(attributes, "user.id")

extensions:
  health_check:
    endpoint: 0.0.0.0:13133
    path: /health

exporters:
  otlp:
    endpoint: otel-gateway.observability:4317
    sending_queue:
      enabled: true
      queue_size: 2000
    retry_on_failure:
      enabled: true
      initial_interval: 1s
      max_interval: 30s
      max_elapsed_time: 120s

service:
  extensions: [health_check]