OTel to ClickStack

# OTel -> ClickHouse (ClickStack / HyperDX compatible)
# Collector contrib v0.146.0. Exporter stability: traces/logs beta, metrics alpha.

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  memory_limiter:
    # Must be the first processor so backpressure reaches the receivers.
    check_interval: 1s
    limit_percentage: 80
    spike_limit_percentage: 20

  transform/parse_body:
    # Services that log JSON emit it as one string body, which makes every
    # field inside it unqueryable. Parse it into attributes (transform
    # README's documented pattern); "insert" never overwrites attributes
    # the SDK already set, and non-JSON bodies pass through untouched.
    error_mode: ignore
    log_statements:
      - merge_maps(log.cache, ParseJSON(log.body), "upsert") where IsMatch(log.body, "^\\s*\\{")
      - flatten(log.cache) where IsMap(log.cache)
      - merge_maps(log.attributes, log.cache, "insert") where IsMap(log.cache)

  transform/normalize:
    # Loggers disagree on field names (msg vs log, lvl vs level) and ship
    # severity as free text. Coalesce the variants, then fill the OTel
    # severity fields so HyperDX severity filters and the SeverityNumber
    # column work. Mapping only runs when severity_number is 0 (unset):
    # records the SDK already classified are left alone.
    error_mode: ignore
    log_statements:
      - set(log.attributes["message"], log.attributes["msg"]) where log.attributes["message"] == nil and log.attributes["msg"] != nil
      - set(log.attributes["message"], log.attributes["log"]) where log.attributes["message"] == nil and log.attributes["log"] != nil
      - delete_key(log.attributes, "msg")
      - delete_key(log.attributes, "log")
      - set(log.attributes["level"], log.attributes["lvl"]) where log.attributes["level"] == nil and log.attributes["lvl"] != nil
      - delete_key(log.attributes, "lvl")
      - set(log.cache["lvl"], ToUpperCase(log.attributes["level"])) where log.attributes["level"] != nil and log.severity_number == 0
      - set(log.severity_text, "TRACE") where log.cache["lvl"] == "TRACE"
      - set(log.severity_number, SEVERITY_NUMBER_TRACE) where log.cache["lvl"] == "TRACE"
      - set(log.severity_text, "DEBUG") where log.cache["lvl"] == "DEBUG"
      - set(log.severity_number, SEVERITY_NUMBER_DEBUG) where log.cache["lvl"] == "DEBUG"
      - set(log.severity_text, "INFO") where log.cache["lvl"] == "INFO"
      - set(log.severity_number, SEVERITY_NUMBER_INFO) where log.cache["lvl"] == "INFO"
      - set(log.severity_text, "WARN") where log.cache["lvl"] == "WARN" or log.cache["lvl"] == "WARNING"
      - set(log.severity_number, SEVERITY_NUMBER_WARN) where log.cache["lvl"] == "WARN" or log.cache["lvl"] == "WARNING"
      - set(log.severity_text, "ERROR") where log.cache["lvl"] == "ERROR" or log.cache["lvl"] == "ERR"
      - set(log.severity_number, SEVERITY_NUMBER_ERROR) where log.cache["lvl"] == "ERROR" or log.cache["lvl"] == "ERR"
      - set(log.severity_text, "FATAL") where log.cache["lvl"] == "FATAL" or log.cache["lvl"] == "CRITICAL" or log.cache["lvl"] == "PANIC"
      - set(log.severity_number, SEVERITY_NUMBER_FATAL) where log.cache["lvl"] == "FATAL" or log.cache["lvl"] == "CRITICAL" or log.cache["lvl"] == "PANIC"

  transform/service_name:
    # HyperDX's service filter keys off service.name (the source's Service Name
    # Expression); fill it in when missing so records can still be filtered by service.
    error_mode: ignore
    log_statements:
      - set(resource.attributes["service.name"], "unknown_service") where resource.attributes["service.name"] == nil
    trace_statements:
      - set(resource.attributes["service.name"], "unknown_service") where resource.attributes["service.name"] == nil
    metric_statements:
      - set(resource.attributes["service.name"], "unknown_service") where resource.attributes["service.name"] == nil

  redaction/mask:
    # Last stop before ClickHouse: keep every attribute, mask values that
    # look like PII. Runs on log attributes and bodies (body scanning since
    # contrib v0.126.0). Once a card number is in the warehouse it stays
    # until the table TTL expires; mask it here instead.
    allow_all_keys: true
    blocked_values:
      - '4[0-9]{12}(?:[0-9]{3})?'                          # Visa
      - '5[1-5][0-9]{14}'                                  # Mastercard
      - '\b\d{3}-\d{2}-\d{4}\b'                            # US SSN
      - 'AKIA[0-9A-Z]{16}'                                 # AWS access key id
      - '[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}'   # email
    summary: silent # set to debug during rollout to see redaction.* counts, then back

  batch:
    # ClickHouse wants few, large inserts (exporter README: >=1000 rows,
    # no more than ~1 request/s). The 8192/200ms defaults flush on timeout
    # at most volumes and create many small parts.
    send_batch_size: 5000
    send_batch_max_size: 10000
    timeout: 5s

exporters:
  clickhouse:
    # ClickHouse Cloud: clickhouse://<host>:9440?secure=true
    # Self-hosted:      tcp://<host>:9000  (append ?secure=true behind TLS)
    endpoint: ${env:CLICKHOUSE_ENDPOINT}
    # ClickStack/HyperDX's stock sources read the `default` database and the
    # exporter's default table names (otel_logs, otel_traces, otel_metrics_*).
    # If you change the database or table names, edit each HyperDX source's
    # Database/Table fields to match.
    database: default
    username: ${env:CLICKHOUSE_USERNAME}
    password: ${env:CLICKHOUSE_PASSWORD}
    create_schema: true # convenient first run; manage DDL yourself at fleet scale
    ttl: 720h           # 30d retention enforced by ClickHouse table TTL
    compress: lz4
    async_insert: true
    timeout: 15s        # large inserts to Cloud can exceed the 5s default
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 300s
    sending_queue:
      enabled: true
      # Few consumers = few concurrent inserts, matching ClickHouse guidance.
      num_consumers: 2
      sizer: items
      queue_size: 100000     # ~20 full batches buffered, counted in items
      storage: file_storage  # persistent queue on disk; survives restarts

extensions:
  file_storage:
    directory: /var/lib/otelcol/file_storage
    create_directory: true

service:
  extensions: [file_storage]
  pipelines:
    logs:
      receivers: [otlp]
      processors: [memory_limiter, transform/parse_body, transform/normalize, transform/service_name, redaction/mask, batch]
      exporters: [clickhouse]
    traces:
      receivers: [otlp]
      processors: [memory_limiter, transform/service_name, batch]
      exporters: [clickhouse]
    metrics:
      # clickhouse exporter metrics support is alpha at v0.146.0
      receivers: [otlp]
      processors: [memory_limiter, transform/service_name, batch]
      exporters: [clickhouse]

Validated against otelcol-contrib v0.147.0. Fill in the ${env:…} placeholders before running it.