ログを検索可能に、そして破産しないように — 構造化、マッピング爆発、保持、そして本当のコスト

# 悪い例 — 検索には正規表現が必要で、フォーマットが変わるとその正規表現が壊れる
log.info(f"order {order_id} for tenant {tenant} failed after {ms}ms: {err}")

# 良い例 — フィールドで検索と集計ができる
log.info(
    "order.failed",
    extra={
        "order.id": order_id,
        "tenant.id": tenant,
        "duration_ms": ms,
        "error.type": type(err).__name__,
        "error.message": str(err)[:500],
        "http.route": route,
    },
)

{
  "@timestamp": "2026-08-02T04:11:52.418Z",
  "severity_text": "ERROR",
  "severity_number": 17,
  "body": "order.failed",
  "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736",
  "span_id": "00f067aa0ba902b7",
  "service.name": "checkout-api",
  "service.version": "2.7.1",
  "deployment.environment.name": "prod",
  "http.route": "/v1/orders/:id",
  "http.response.status_code": 502,
  "error.type": "UpstreamTimeout",
  "order.id": "A-99183",
  "tenant.id": "t-8871",
  "duration_ms": 1421
}

# Pythonの自動計測がログにtrace_idとspan_idを注入する
export OTEL_PYTHON_LOG_CORRELATION=true
export OTEL_PYTHON_LOG_LEVEL=info

{
  "message": "cache stats",
  "cache": {
    "user_8871_hits": 12,
    "user_8871_misses": 3,
    "user_9902_hits": 7
  }
}

PUT _index_template/logs-app
{
  "index_patterns": ["logs-app-*"],
  "data_stream": {},
  "priority": 200,
  "template": {
    "settings": {
      "index.number_of_shards": 3,
      "index.number_of_replicas": 1,
      "index.refresh_interval": "30s",
      "index.codec": "zstd_no_dict",
      "index.mapping.total_fields.limit": 1500,
      "index.mapping.depth.limit": 8,
      "index.mapping.nested_fields.limit": 20
    },
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "@timestamp":        { "type": "date" },
        "severity_text":     { "type": "keyword" },
        "body":              { "type": "keyword" },
        "trace_id":          { "type": "keyword" },
        "span_id":           { "type": "keyword" },
        "service.name":      { "type": "keyword" },
        "service.version":   { "type": "keyword" },
        "http.route":        { "type": "keyword" },
        "http.response.status_code": { "type": "short" },
        "error.type":        { "type": "keyword" },
        "error.message":     { "type": "text", "index": true, "norms": false },
        "duration_ms":       { "type": "integer" },
        "tenant.id":         { "type": "keyword" },
        "order.id":          { "type": "keyword", "doc_values": true, "index": true },
        "stack_trace":       { "type": "text", "index": false },
        "attributes":        { "type": "flat_object" }
      }
    }
  }
}

# インデックスごとのフィールド数 — 四桁が出たらすでに問題がある
curl -s 'https://opensearch:9200/logs-app-000042/_mapping?pretty' \
  | jq '[paths(type=="object" and has("type")) | length] | length'

# クラスタ状態のサイズとマッピング更新のキュー
curl -s 'https://opensearch:9200/_cluster/stats?pretty' \
  | jq '.indices.mappings'

curl -s 'https://opensearch:9200/_cluster/health?pretty' \
  | jq '{status, number_of_pending_tasks, task_max_waiting_in_queue_millis}'

# どのインデックスがディスクを使っているか
curl -s 'https://opensearch:9200/_cat/indices/logs-*?v&s=store.size:desc&h=index,docs.count,store.size,pri,rep' \
  | head -20

PUT _plugins/_ism/policies/logs-app-lifecycle
{
  "policy": {
    "description": "アプリケーションログの30日保持",
    "default_state": "hot",
    "ism_template": [
      {
        "index_patterns": ["logs-app-*"],
        "priority": 200
      }
    ],
    "states": [
      {
        "name": "hot",
        "actions": [
          {
            "rollover": {
              "min_primary_shard_size": "30gb",
              "min_index_age": "1d"
            }
          }
        ],
        "transitions": [
          { "state_name": "warm", "conditions": { "min_index_age": "3d" } }
        ]
      },
      {
        "name": "warm",
        "actions": [
          { "replica_count": { "number_of_replicas": 0 } },
          { "force_merge": { "max_num_segments": 1 } }
        ],
        "transitions": [
          { "state_name": "cold", "conditions": { "min_index_age": "10d" } }
        ]
      },
      {
        "name": "cold",
        "actions": [
          { "read_only": {} }
        ],
        "transitions": [
          { "state_name": "delete", "conditions": { "min_index_age": "30d" } }
        ]
      },
      {
        "name": "delete",
        "actions": [{ "delete": {} }]
      }
    ]
  }
}

# ポリシーが付いたインデックスと現在の状態
curl -s 'https://opensearch:9200/_plugins/_ism/explain/logs-app-*?pretty' \
  | jq 'to_entries[] | select(.value.index != null)
        | {index: .value.index, state: .value."policy_id", step: .value.step.name, failed: .value.failed}'

# 失敗した管理対象インデックスだけ
curl -s 'https://opensearch:9200/_plugins/_ism/explain/logs-app-*?pretty' \
  | jq '[to_entries[] | select(.value.failed == true) | .key]'

# otel-collector.yaml — ログパイプライン
processors:
  memory_limiter:
    check_interval: 1s
    limit_percentage: 80
    spike_limit_percentage: 20

  # 1) ヘルスチェックはまるごと捨てる
  filter/drop_health:
    error_mode: ignore
    logs:
      log_record:
        - 'attributes["http.route"] == "/healthz"'
        - 'attributes["http.route"] == "/readyz"'
        - 'attributes["http.route"] == "/metrics"'

  # 2) 正常経路のDEBUGは10%だけ残す。エラーには手を付けない
  probabilistic_sampler/debug:
    sampling_percentage: 10
    attribute_source: record
    from_attribute: trace_id

  # 3) 大きなフィールドは切り詰める
  transform/trim:
    error_mode: ignore
    log_statements:
      - context: log
        statements:
          - truncate_all(attributes, 4096)
          - delete_key(attributes, "http.request.body")
          - set(attributes["error.message"], Substring(attributes["error.message"], 0, 500))
            where attributes["error.message"] != nil

  batch:
    timeout: 5s
    send_batch_size: 8192

exporters:
  opensearch:
    http:
      endpoint: https://opensearch.observability.svc:9200
    logs_index: logs-app
    sending_queue:
      enabled: true
      queue_size: 5000
    retry_on_failure:
      enabled: true
      max_elapsed_time: 300s

service:
  pipelines:
    # エラーと警告は無条件で全量保持
    logs/critical:
      receivers: [otlp]
      processors: [memory_limiter, transform/trim, batch]
      exporters: [opensearch]
    # 正常経路はサンプリングする
    logs/routine:
      receivers: [otlp]
      processors: [memory_limiter, filter/drop_health, probabilistic_sampler/debug, transform/trim, batch]
      exporters: [opensearch]

# 規模の想定
rps            = 5_000          # 秒間リクエスト数
log_bytes      = 800            # ログ1件の索引後サイズ (バイト)
seconds_of_day = 86_400

daily_gb = rps * log_bytes * seconds_of_day / 1024**3
print(f"1日の索引量: {daily_gb:.1f} GB")
# 1日の索引量: 321.8 GB

# 同じ情報をメトリクスで表現すると
routes, status_classes, pods = 120, 5, 40
series = routes * status_classes * pods
samples_per_day = series * (seconds_of_day / 15)      # 15秒スクレイプ
metric_bytes = samples_per_day * 2                    # 圧縮後サンプルあたり約2バイト
print(f"時系列 {series:,}本、1日 {metric_bytes/1024**3:.3f} GB")
# 時系列 24,000本、1日 0.257 GB

# ログからカウントメトリクスを作りPrometheusへ送る
connectors:
  count:
    logs:
      log.error.count:
        description: 重大度別のログ発生数
        conditions:
          - 'severity_number >= 17'
        attributes:
          - key: service.name
          - key: error.type
          - key: http.route

service:
  pipelines:
    logs/in:
      receivers: [otlp]
      processors: [memory_limiter, batch]
      exporters: [opensearch, count]
    metrics/from_logs:
      receivers: [count]
      processors: [batch]
      exporters: [prometheusremotewrite]

exporters:
  opensearch:
    sending_queue:
      enabled: true
      queue_size: 5000
      # キューが満杯なら新しいデータを捨てる。アプリへバックプレッシャーを戻さない
      block_on_overflow: false