관측 데이터를 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")