Observability 데이터 파이프라인 비용 최적화: 샘플링·필터링·티어링 전략

[생성] --> [수집/전송] --> [처리/가공] --> [인덱싱] --> [저장] --> [쿼리]
  |            |              |             |            |          |
 SDK       네트워크       Collector      백엔드       스토리지    컴퓨팅
 오버헤드    대역폭        CPU/메모리    인덱스 I/O   디스크/S3   쿼리 비용

월간 비용 = (인제스트 볼륨 x 인제스트 단가) + (저장 볼륨 x 스토리지 단가) + (쿼리 수 x 쿼리 단가)

인제스트 볼륨 = 서비스 수 x 서비스당 RPS x 스팬/로그 비율 x 평균 페이로드 크기 x 보존 기간

# OpenTelemetry Collector - Probabilistic Sampler Processor
processors:
  probabilistic_sampler:
    # 전체 트레이스의 10%만 샘플링
    sampling_percentage: 10
    # hash_seed: 트레이스 ID 해싱 시드 (다중 Collector 인스턴스에서 동일 결정을 위해 고정)
    hash_seed: 22

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

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased, ParentBasedTraceIdRatio
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# 10% Head Sampling - ParentBased는 부모 스팬의 결정을 따름
sampler = ParentBasedTraceIdRatio(0.1)

provider = TracerProvider(sampler=sampler)

# Batch processor로 효율적 전송
provider.add_span_processor(
    BatchSpanProcessor(
        OTLPSpanExporter(endpoint="http://otel-collector:4317"),
        max_queue_size=2048,
        max_export_batch_size=512,
        schedule_delay_millis=5000,
    )
)
trace.set_tracer_provider(provider)

# OpenTelemetry Collector - Tail Sampling Processor
# 반드시 Gateway 모드의 Collector에서 사용할 것
processors:
  tail_sampling:
    # 첫 번째 스팬 수신 후 결정까지 대기 시간 (트레이스 완료 대기)
    decision_wait: 30s
    # 메모리에 보관할 최대 트레이스 수
    num_traces: 100000
    # 초당 예상 신규 트레이스 수 (내부 메모리 할당 최적화)
    expected_new_traces_per_sec: 1000
    policies:
      # 정책 1: 에러가 있는 트레이스는 100% 보존
      - name: error-policy
        type: status_code
        status_code:
          status_codes:
            - ERROR

      # 정책 2: 500ms 이상 지연된 트레이스는 100% 보존
      - name: latency-policy
        type: latency
        latency:
          threshold_ms: 500

      # 정책 3: 특정 서비스의 트레이스 100% 보존
      - name: critical-service-policy
        type: string_attribute
        string_attribute:
          key: service.name
          values:
            - payment-service
            - auth-service
            - order-service

      # 정책 4: 나머지 정상 트레이스는 5%만 샘플링
      - name: normal-traffic-policy
        type: probabilistic
        probabilistic:
          sampling_percentage: 5

      # 정책 5: 복합 정책 - AND 조건
      - name: composite-policy
        type: and
        and:
          and_sub_policy:
            - name: is-health-check
              type: string_attribute
              string_attribute:
                key: http.route
                values:
                  - /healthz
                  - /readyz
                  - /livez
            - name: drop-most
              type: probabilistic
              probabilistic:
                sampling_percentage: 0.1

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

[서비스 A] --OTLP--> [Agent Collector (DaemonSet)]
[서비스 B] --OTLP--> [Agent Collector (DaemonSet)]  --트레이스ID 기반 라우팅-->
[서비스 C] --OTLP--> [Agent Collector (DaemonSet)]

                                                    [Gateway Collector #1] --> [Tempo]
                                                    [Gateway Collector #2] --> [Tempo]
                                                    [Gateway Collector #3] --> [Tempo]
                                                    (Tail Sampling 수행)

# Agent Collector 설정 - LoadBalancing Exporter
exporters:
  loadbalancing:
    protocol:
      otlp:
        tls:
          insecure: true
    resolver:
      dns:
        hostname: otel-gateway-headless.observability.svc.cluster.local
        port: 4317

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [loadbalancing]

# OpenTelemetry Collector - Log Level Filtering
processors:
  # DEBUG/TRACE 로그 드롭
  filter/drop-debug:
    error_mode: ignore
    logs:
      log_record:
        - 'severity_number < SEVERITY_NUMBER_INFO'

  # 특정 로그 본문 패턴 드롭 (헬스체크, readiness probe)
  filter/drop-healthcheck:
    error_mode: ignore
    logs:
      log_record:
        - 'IsMatch(body, ".*GET /health.*")'
        - 'IsMatch(body, ".*GET /readyz.*")'
        - 'IsMatch(body, ".*GET /livez.*")'
        - 'IsMatch(body, ".*kube-probe.*")'

  # 특정 네임스페이스의 로그만 보존
  filter/namespace:
    error_mode: ignore
    logs:
      log_record:
        - 'resource.attributes["k8s.namespace.name"] == "kube-system"'

  # 속성 변환 - 불필요한 필드 제거로 페이로드 축소
  transform/reduce-attributes:
    error_mode: ignore
    log_statements:
      - context: log
        statements:
          - delete_key(attributes, "log.file.path")
          - delete_key(attributes, "log.iostream")
          - truncate_all(attributes, 256)
          - limit(attributes, 20)

service:
  pipelines:
    logs:
      receivers: [otlp, filelog]
      processors:
        - memory_limiter
        - filter/drop-debug
        - filter/drop-healthcheck
        - filter/namespace
        - transform/reduce-attributes
        - batch
      exporters: [otlp/loki]

# OpenTelemetry Collector - Group By Attributes로 로그 집계
processors:
  groupbyattrs:
    keys:
      - service.name
      - severity_text
      - log.template

  # 중복 로그 집계 후 카운트 속성 추가
  transform/aggregate-count:
    error_mode: ignore
    log_statements:
      - context: log
        statements:
          - set(attributes["log.dedup_count"], 1) where attributes["log.dedup_count"] == nil

#!/bin/bash
# Collector 내부 메트릭으로 필터링 효과 측정
# Prometheus 엔드포인트에서 수집

# 필터링 전 수신된 로그 수
RECEIVED=$(curl -s http://localhost:8888/metrics | \
  grep 'otelcol_receiver_accepted_log_records' | \
  awk '{sum += $2} END {print sum}')

# 필터링 후 전송된 로그 수
EXPORTED=$(curl -s http://localhost:8888/metrics | \
  grep 'otelcol_exporter_sent_log_records' | \
  awk '{sum += $2} END {print sum}')

# 드롭률 계산
if [ "$RECEIVED" -gt 0 ]; then
  DROP_RATE=$(echo "scale=2; (1 - $EXPORTED / $RECEIVED) * 100" | bc)
  echo "수신 로그: $RECEIVED"
  echo "전송 로그: $EXPORTED"
  echo "드롭률: ${DROP_RATE}%"
  echo "예상 비용 절감: 월간 $(echo "scale=0; $DROP_RATE * 150 / 100" | bc) USD (기준: 150 USD/월)"
fi

# 프로세서별 드롭 카운트 확인
echo ""
echo "=== 프로세서별 드롭 현황 ==="
curl -s http://localhost:8888/metrics | \
  grep 'otelcol_processor_dropped_log_records' | \
  sort -t' ' -k2 -rn

# OpenTelemetry Collector - 메트릭 카디널리티 관리
processors:
  # 고카디널리티 속성 제거
  metricstransform/drop-high-cardinality:
    transforms:
      - include: http.server.request.duration
        action: update
        operations:
          # user_id, session_id 등 고카디널리티 레이블 제거
          - action: delete_label_value
            label: user_id
          - action: delete_label_value
            label: session_id

  # URL 경로 정규화
  transform/normalize-paths:
    error_mode: ignore
    metric_statements:
      - context: datapoint
        statements:
          - replace_pattern(attributes["url.path"], "^/api/users/[0-9]+", "/api/users/:id")
          - replace_pattern(attributes["url.path"], "^/api/orders/[0-9]+", "/api/orders/:id")
          - replace_pattern(attributes["url.path"], "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", ":uuid")

  # 속성 수 제한
  attributes/limit:
    actions:
      - key: http.request.header.x-request-id
        action: delete
      - key: http.request.header.authorization
        action: delete

  # 메트릭 집계 - 세밀한 레이블을 제거하고 집계된 메트릭만 전송
  filter/metrics-allowlist:
    error_mode: ignore
    metrics:
      metric:
        # 특정 메트릭만 허용 (화이트리스트 방식)
        - 'name == "http.server.request.duration"'
        - 'name == "http.server.active_requests"'
        - 'name == "process.runtime.go.goroutines"'
        - 'IsMatch(name, "^system\\..*")'

service:
  pipelines:
    metrics:
      receivers: [otlp, prometheus]
      processors:
        - memory_limiter
        - metricstransform/drop-high-cardinality
        - transform/normalize-paths
        - attributes/limit
        - filter/metrics-allowlist
        - batch
      exporters: [prometheusremotewrite/mimir]

#!/usr/bin/env python3
"""
Observability Budget Enforcer
서비스별 텔레메트리 예산을 모니터링하고 경고를 발생시키는 스크립트
"""

import requests
import yaml
import sys
from datetime import datetime

# 서비스별 Observability Budget 정의 (일간 기준)
BUDGETS = {
    "payment-service": {
        "traces_per_day": 5_000_000,
        "logs_per_day": 10_000_000,
        "metrics_series": 50_000,
        "alert_threshold": 0.8,  # 80% 도달 시 경고
    },
    "user-service": {
        "traces_per_day": 2_000_000,
        "logs_per_day": 5_000_000,
        "metrics_series": 30_000,
        "alert_threshold": 0.8,
    },
    "default": {
        "traces_per_day": 1_000_000,
        "logs_per_day": 3_000_000,
        "metrics_series": 20_000,
        "alert_threshold": 0.8,
    },
}

PROMETHEUS_URL = "http://prometheus:9090"


def query_prometheus(query: str) -> float:
    """Prometheus에서 현재 값을 조회한다."""
    resp = requests.get(
        f"{PROMETHEUS_URL}/api/v1/query",
        params={"query": query},
    )
    result = resp.json().get("data", {}).get("result", [])
    if result:
        return float(result[0]["value"][1])
    return 0.0


def check_budget(service_name: str) -> dict:
    """서비스의 현재 텔레메트리 사용량을 예산 대비 확인한다."""
    budget = BUDGETS.get(service_name, BUDGETS["default"])
    today = datetime.now().strftime("%Y-%m-%d")

    # 오늘 생성된 트레이스 수
    traces_today = query_prometheus(
        f'sum(increase(traces_spanmetrics_calls_total{{service_name="{service_name}"}}[24h]))'
    )

    # 오늘 생성된 로그 수
    logs_today = query_prometheus(
        f'sum(increase(loki_distributor_lines_received_total{{service="{service_name}"}}[24h]))'
    )

    # 현재 활성 메트릭 시리즈 수
    active_series = query_prometheus(
        f'count({{service_name="{service_name}"}})'
    )

    usage = {
        "service": service_name,
        "date": today,
        "traces": {
            "current": int(traces_today),
            "budget": budget["traces_per_day"],
            "usage_pct": round(traces_today / budget["traces_per_day"] * 100, 1),
        },
        "logs": {
            "current": int(logs_today),
            "budget": budget["logs_per_day"],
            "usage_pct": round(logs_today / budget["logs_per_day"] * 100, 1),
        },
        "metrics_series": {
            "current": int(active_series),
            "budget": budget["metrics_series"],
            "usage_pct": round(active_series / budget["metrics_series"] * 100, 1),
        },
    }

    # 예산 초과 여부 확인
    for signal_type, data in usage.items():
        if isinstance(data, dict) and "usage_pct" in data:
            if data["usage_pct"] >= budget["alert_threshold"] * 100:
                print(
                    f"[WARNING] {service_name}: {signal_type} 예산 "
                    f"{data['usage_pct']}% 사용 ({data['current']:,} / {data['budget']:,})"
                )

    return usage


if __name__ == "__main__":
    services = sys.argv[1:] if len(sys.argv) > 1 else list(BUDGETS.keys())
    for svc in services:
        if svc == "default":
            continue
        result = check_budget(svc)
        print(f"\n=== {svc} Observability Budget ===")
        for key, val in result.items():
            if isinstance(val, dict):
                print(f"  {key}: {val['current']:,} / {val['budget']:,} ({val['usage_pct']}%)")

# Elasticsearch ILM Policy - 로그 티어링
# PUT _ilm/policy/observability-logs-policy
{
  'policy':
    {
      'phases':
        {
          'hot':
            {
              'min_age': '0ms',
              'actions':
                {
                  'rollover': { 'max_primary_shard_size': '50gb', 'max_age': '1d' },
                  'set_priority': { 'priority': 100 },
                },
            },
          'warm':
            {
              'min_age': '7d',
              'actions':
                {
                  'shrink': { 'number_of_shards': 1 },
                  'forcemerge': { 'max_num_segments': 1 },
                  'allocate': { 'require': { 'data': 'warm' } },
                  'set_priority': { 'priority': 50 },
                },
            },
          'cold':
            {
              'min_age': '30d',
              'actions':
                {
                  'searchable_snapshot':
                    { 'snapshot_repository': 's3-repo', 'force_merge_index': true },
                  'allocate': { 'require': { 'data': 'cold' } },
                  'set_priority': { 'priority': 0 },
                },
            },
          'delete': { 'min_age': '365d', 'actions': { 'delete': {} } },
        },
    },
}

# Mimir 설정 - 메트릭 보존 및 다운샘플링
limits:
  # 원본 해상도 메트릭은 14일 보존
  compactor_blocks_retention_period: 14d

compactor:
  # 다운샘플링 활성화
  downsample:
    # 7일 이후 5분 해상도로 다운샘플
    - resolution: 5m
      retention: 7d
    # 30일 이후 1시간 해상도로 다운샘플
    - resolution: 1h
      retention: 30d

store_gateway:
  sharding_ring:
    replication_factor: 3

# S3 백엔드 설정
blocks_storage:
  backend: s3
  s3:
    bucket_name: mimir-metrics
    endpoint: s3.ap-northeast-2.amazonaws.com
    storage_class: INTELLIGENT_TIERING
  bucket_store:
    sync_dir: /data/mimir-sync
    # 인덱스 캐시로 쿼리 성능 유지
    index_cache:
      backend: memcached
      memcached:
        addresses: dns+memcached.observability.svc:11211

                    [마이크로서비스 클러스터]
                    (SDK Head Sampling 10%)
                            |
                            v
            +-------------------------------+
            |   Agent Collector (DaemonSet)  |
            |   - Memory Limiter            |
            |   - Filter/drop-debug         |
            |   - Filter/drop-healthcheck   |
            |   - Attributes/limit          |
            +-------------------------------+
                     |          |          |
                  Traces     Logs      Metrics
                     |          |          |
                     v          v          v
     +---------------+    +---------+  +-----------+
     | Gateway        |    | Gateway |  | Gateway   |
     | Collector      |    | (Logs)  |  | (Metrics) |
     | (Traces)       |    |         |  |           |
     | - Tail Sampling|    | - Group |  | - Cardina |
     | - LoadBalance  |    |   ByAttr|  |   lity    |
     +-------+--------+    +----+----+  +-----+-----+
             |                   |              |
             v                   v              v
     +-------+--------+  +------+------+  +----+------+
     | Tempo          |  | Loki        |  | Mimir     |
     | (S3 + Intelli- |  | (S3 + ILM)  |  | (S3 +     |
     |  gent Tiering) |  |             |  | Downsample|
     +----------------+  +-------------+  +-----------+
             |                   |              |
             v                   v              v
     +--------------------------------------------------+
     |              Grafana Dashboard                    |
     |  - 실시간 모니터링 (Hot Tier)                       |
     |  - 장애 분석 (Warm Tier)                           |
     |  - 컴플라이언스 감사 (Cold/Archive Tier)            |
     +--------------------------------------------------+

# OOM 방지를 위한 Memory Limiter + Tail Sampling 조합
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 3800 # 4GB 컨테이너에서 200MB 여유
    spike_limit_mib: 800 # 일시적 버스트 허용

  tail_sampling:
    decision_wait: 20s # 30s에서 20s로 축소
    num_traces: 50000 # 메모리 기반 계산
    expected_new_traces_per_sec: 2500
    policies:
      - name: error-always
        type: status_code
        status_code:
          status_codes: [ERROR]
      - name: latency-always
        type: latency
        latency:
          threshold_ms: 500
      - name: probabilistic-default
        type: probabilistic
        probabilistic:
          sampling_percentage: 5

service:
  pipelines:
    traces:
      receivers: [otlp]
      # memory_limiter가 반드시 첫 번째
      processors: [memory_limiter, tail_sampling, batch]
      exporters: [otlp/tempo]

# 1. Collector 메모리 사용량 확인
kubectl top pods -n observability -l app=otel-gateway

# 2. pprof로 메모리 프로파일링 (Extension 활성화 필요)
curl -s http://localhost:1777/debug/pprof/heap > heap.prof
go tool pprof -top heap.prof

# 3. zPages에서 파이프라인 상태 확인
curl -s http://localhost:55679/debug/tracez | jq .

# 4. Tail Sampling 큐 상태 확인
curl -s http://localhost:8888/metrics | grep tail_sampling

# 디버깅용 파이프라인 설정
exporters:
  debug:
    verbosity: detailed
    sampling_initial: 5
    sampling_thereafter: 200

service:
  pipelines:
    logs/debug:
      receivers: [otlp]
      processors: [filter/drop-debug]
      exporters: [debug]

# Prometheus에서 카디널리티 상위 메트릭 확인
curl -s http://prometheus:9090/api/v1/status/tsdb | \
  jq '.data.seriesCountByMetricName | sort_by(-.value) | .[0:10]'

# 특정 메트릭의 레이블 카디널리티 분석
curl -s 'http://prometheus:9090/api/v1/query?query=count(http_server_request_duration_seconds_bucket) by (service_name)' | \
  jq '.data.result | sort_by(-.value[1] | tonumber) | .[0:10]'

# Mimir에서 테넌트별 활성 시리즈 수 확인
curl -s http://mimir:8080/api/v1/cardinality/active_series | \
  jq '.data | sort_by(-.active_series) | .[0:10]'