필사 모드: 観測データをClickHouseに入れるということ — スキーマ、ロールアップ、TTL、そして役割分担

SELECT
    table,
    formatReadableSize(sum(data_uncompressed_bytes)) AS raw,
    formatReadableSize(sum(data_compressed_bytes))   AS compressed,
    round(sum(data_uncompressed_bytes) / sum(data_compressed_bytes), 1) AS ratio
FROM system.columns
WHERE database = 'otel'
GROUP BY table
ORDER BY sum(data_compressed_bytes) DESC;

-- カラム単位でも見られる。圧縮が効かないカラムがコストの大半を占める
SELECT
    name,
    type,
    formatReadableSize(data_compressed_bytes) AS compressed,
    round(data_uncompressed_bytes / data_compressed_bytes, 1) AS ratio
FROM system.columns
WHERE database = 'otel' AND table = 'otel_traces'
ORDER BY data_compressed_bytes DESC
LIMIT 15;

CREATE TABLE otel.otel_traces
(
    Timestamp          DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    TraceId            String CODEC(ZSTD(1)),
    SpanId             String CODEC(ZSTD(1)),
    ParentSpanId       String CODEC(ZSTD(1)),
    TraceState         String CODEC(ZSTD(1)),
    SpanName           LowCardinality(String) CODEC(ZSTD(1)),
    SpanKind           LowCardinality(String) CODEC(ZSTD(1)),
    ServiceName        LowCardinality(String) CODEC(ZSTD(1)),
    ResourceAttributes Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    ScopeName          String CODEC(ZSTD(1)),
    ScopeVersion       String CODEC(ZSTD(1)),
    SpanAttributes     Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    Duration           UInt64 CODEC(ZSTD(1)),
    StatusCode         LowCardinality(String) CODEC(ZSTD(1)),
    StatusMessage      String CODEC(ZSTD(1)),

    Events Nested (
        Timestamp  DateTime64(9),
        Name       LowCardinality(String),
        Attributes Map(LowCardinality(String), String)
    ) CODEC(ZSTD(1)),

    Links Nested (
        TraceId    String,
        SpanId     String,
        TraceState String,
        Attributes Map(LowCardinality(String), String)
    ) CODEC(ZSTD(1)),

    INDEX idx_trace_id       TraceId TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_res_attr_key   mapKeys(ResourceAttributes)   TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_span_attr_key  mapKeys(SpanAttributes)       TYPE bloom_filter(0.01) GRANULARITY 1,
    INDEX idx_duration       Duration TYPE minmax GRANULARITY 1
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, SpanName, toDateTime(Timestamp))
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

CREATE TABLE otel.otel_logs
(
    Timestamp          DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    TraceId            String CODEC(ZSTD(1)),
    SpanId             String CODEC(ZSTD(1)),
    TraceFlags         UInt8,
    SeverityText       LowCardinality(String) CODEC(ZSTD(1)),
    SeverityNumber     UInt8,
    ServiceName        LowCardinality(String) CODEC(ZSTD(1)),
    Body               String CODEC(ZSTD(1)),
    ResourceAttributes Map(LowCardinality(String), String) CODEC(ZSTD(1)),
    LogAttributes      Map(LowCardinality(String), String) CODEC(ZSTD(1)),

    -- よくフィルタリングするキーは物理カラムに昇格させる
    HttpRoute    LowCardinality(String) MATERIALIZED LogAttributes['http.route'],
    HttpStatus   UInt16                 MATERIALIZED toUInt16OrZero(LogAttributes['http.response.status_code']),
    ErrorType    LowCardinality(String) MATERIALIZED LogAttributes['error.type'],

    INDEX idx_trace_id  TraceId    TYPE bloom_filter(0.001) GRANULARITY 1,
    INDEX idx_body      Body       TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 1,
    INDEX idx_severity  SeverityNumber TYPE set(16) GRANULARITY 4
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, toStartOfFiveMinutes(Timestamp), Timestamp)
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;

-- JSON型の使用例 — サブカラム数に上限を設ける
CREATE TABLE otel.otel_logs_json
(
    Timestamp    DateTime64(9) CODEC(Delta(8), ZSTD(1)),
    ServiceName  LowCardinality(String) CODEC(ZSTD(1)),
    SeverityText LowCardinality(String) CODEC(ZSTD(1)),
    TraceId      String CODEC(ZSTD(1)),
    Body         String CODEC(ZSTD(1)),
    Attributes   JSON(max_dynamic_paths = 512) CODEC(ZSTD(1))
)
ENGINE = MergeTree
PARTITION BY toDate(Timestamp)
ORDER BY (ServiceName, toStartOfFiveMinutes(Timestamp), Timestamp)
TTL toDateTime(Timestamp) + toIntervalDay(30)
SETTINGS ttl_only_drop_parts = 1;

-- サブパスに型を明示してアクセスすると、インデックスと統計情報が活用される
SELECT
    ServiceName,
    Attributes.http.route::LowCardinality(String) AS route,
    count() AS c,
    quantile(0.99)(Attributes.duration_ms::Float64) AS p99
FROM otel.otel_logs_json
WHERE Timestamp >= now() - INTERVAL 1 HOUR
  AND SeverityText = 'ERROR'
GROUP BY ServiceName, route
ORDER BY c DESC
LIMIT 20;

-- ストレージポリシーの定義 (config.xml または別の設定ファイル)
-- hot: ローカルNVMe、cold: オブジェクトストレージ

ALTER TABLE otel.otel_traces
MODIFY TTL
    toDateTime(Timestamp) + INTERVAL 3 DAY  TO VOLUME 'hot',
    toDateTime(Timestamp) + INTERVAL 14 DAY TO VOLUME 'cold',
    toDateTime(Timestamp) + INTERVAL 90 DAY DELETE;

-- パーティションごとのサイズと最も古いデータ
SELECT
    table,
    partition,
    formatReadableSize(sum(bytes_on_disk)) AS size,
    sum(rows) AS rows,
    min(min_time) AS oldest
FROM system.parts
WHERE database = 'otel' AND active
GROUP BY table, partition
ORDER BY partition
LIMIT 10;

-- マージの待機と進行中のもの
SELECT table, elapsed, progress, num_parts, formatReadableSize(memory_usage) AS mem
FROM system.merges
WHERE database = 'otel';

-- パート数が多いと挿入が拒否され始める
SELECT table, count() AS parts
FROM system.parts
WHERE database = 'otel' AND active
GROUP BY table
ORDER BY parts DESC;

-- 1) 集計結果を格納するテーブル
CREATE TABLE otel.trace_rollup_1m
(
    Bucket      DateTime,
    ServiceName LowCardinality(String),
    SpanName    LowCardinality(String),
    SpanKind    LowCardinality(String),
    Calls       AggregateFunction(count),
    Errors      AggregateFunction(countIf, UInt8),
    DurationQ   AggregateFunction(quantiles(0.5, 0.9, 0.99), Float64),
    DurationSum AggregateFunction(sum, Float64)
)
ENGINE = AggregatingMergeTree
PARTITION BY toYYYYMM(Bucket)
ORDER BY (ServiceName, SpanName, SpanKind, Bucket)
TTL Bucket + toIntervalDay(400);

-- 2) 生テーブルに挿入されたタイミングで自動的に集計する
CREATE MATERIALIZED VIEW otel.trace_rollup_1m_mv TO otel.trace_rollup_1m AS
SELECT
    toStartOfMinute(Timestamp) AS Bucket,
    ServiceName,
    SpanName,
    SpanKind,
    countState()                                   AS Calls,
    countIfState(StatusCode = 'Error')             AS Errors,
    quantilesState(0.5, 0.9, 0.99)(Duration / 1e6) AS DurationQ,
    sumState(Duration / 1e6)                       AS DurationSum
FROM otel.otel_traces
GROUP BY Bucket, ServiceName, SpanName, SpanKind;

-- 3) クエリ時に状態をマージする
SELECT
    ServiceName,
    SpanName,
    countMerge(Calls)                          AS calls,
    countIfMerge(Errors)                       AS errors,
    round(countIfMerge(Errors) / countMerge(Calls), 4) AS error_ratio,
    arrayElement(quantilesMerge(0.5, 0.9, 0.99)(DurationQ), 3) AS p99_ms
FROM otel.trace_rollup_1m
WHERE Bucket >= now() - INTERVAL 30 DAY
GROUP BY ServiceName, SpanName
ORDER BY calls DESC
LIMIT 30;

# otel-collector.yaml
exporters:
  clickhouse:
    endpoint: tcp://clickhouse.observability.svc:9000?dial_timeout=10s
    database: otel
    username: otel_writer
    password: ${env:CLICKHOUSE_PASSWORD}
    # 本番ではDDLを自分で管理する
    create_schema: false
    logs_table_name: otel_logs
    traces_table_name: otel_traces
    compress: lz4
    async_insert: true
    timeout: 10s
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_elapsed_time: 300s

processors:
  # ClickHouseは大きなバッチを好む。小さな挿入が頻繁だとパートが急増する
  batch:
    timeout: 10s
    send_batch_size: 20000
    send_batch_max_size: 50000

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

EXPLAIN indexes = 1
SELECT count()
FROM otel.otel_traces
WHERE ServiceName = 'checkout-api'
  AND Timestamp >= now() - INTERVAL 1 HOUR;
-- 読み込んだマーク数が全体のごく一部であるべき

processors:
  transform/normalize:
    error_mode: ignore
    trace_statements:
      - context: resource
        statements:
          # 古い名前で入ってきたものを現在の規約に統一する
          - set(attributes["deployment.environment.name"], attributes["deployment.environment"])
            where attributes["deployment.environment.name"] == nil
                  and attributes["deployment.environment"] != nil
          - delete_key(attributes, "deployment.environment")