---
# Vector TRANSFORM tier (the "brain") — single source of truth, also validated
# by `vector test` in CI. Consumes the whole log stream from JetStream via the
# durable `transform` consumer (at-least-once; durable offsets tracked by
# JetStream), routes by subject, normalises into the logs.raw columns, and is
# the ONLY ClickHouse writer. Per-app parsing is added here as follow-ups:
# insert a transform and append its id to the clickhouse sink `inputs` — no edge
# or VM rollout required.
#
# Durability model: JetStream (3d / 130 GiB, S2-compressed) is the SOLE
# durability layer and the replay window. This
# tier is stateless (no PVC, memory buffer). If ClickHouse is down the sink
# blocks (when_full=block); back-pressure stops the source pulling, so unpulled
# messages stay in JetStream and are redelivered. NB: Vector's NATS source has
# no end-to-end acks (acks on receipt), so a pod killed mid-outage can lose the
# in-memory buffer's worth of already-pulled events — accepted for a stateless,
# autoscalable tier.
data_dir: /vector-data-dir

api:
  enabled: true
  address: 0.0.0.0:8686

sources:
  js_in:
    type: nats
    url: nats://nats.logging.svc.cluster.local:4222
    connection_name: vector-transform
    subject: "logs.>"
    jetstream:
      stream: LOGS
      consumer: transform
    auth:
      strategy: user_password
      user_password:
        user: log-consumer
        password: ${NATS_CONSUMER_PASSWORD}
    decoding:
      codec: json

transforms:
  route:
    type: route
    inputs:
      - js_in
    route:
      k8s: 'starts_with(to_string(.subject) ?? "", "logs.k8s.")'
      vm: 'starts_with(to_string(.subject) ?? "", "logs.vm.")'

  k8s_shape:
    type: remap
    inputs:
      - route.k8s
    source: |
      ts = .timestamp || now()
      node = to_string(.kubernetes.pod_node_name || "") ?? ""
      ns = to_string(.kubernetes.pod_namespace || "") ?? ""
      pod = to_string(.kubernetes.pod_name || "") ?? ""
      container = to_string(.kubernetes.container_name || "") ?? ""
      strm = to_string(.stream || "") ?? ""
      msg = to_string(.message || "") ?? ""
      lbls = object(.kubernetes.pod_labels) ?? {}
      . = {
        "timestamp": ts,
        "host": node,
        "source": "k8s",
        "namespace": ns,
        "pod": pod,
        "container": container,
        "stream": strm,
        "severity": "",
        "message": msg,
        "labels": lbls,
        "fields": {}
      }

  vm_shape:
    type: remap
    inputs:
      - route.vm
    source: |
      ts = .timestamp || .ts || now()
      host = to_string(.host || .hostname || "") ?? ""
      msg = to_string(.message || .msg || "") ?? ""
      sev = to_string(.severity || .level || "") ?? ""
      role = to_string(.role || "") ?? ""
      lbls = {}
      if role != "" {
        lbls = {"role": role}
      }
      . = {
        "timestamp": ts,
        "host": host,
        "source": "vm",
        "namespace": "",
        "pod": "",
        "container": "",
        "stream": "",
        "severity": sev,
        "message": msg,
        "labels": lbls,
        "fields": {}
      }

sinks:
  clickhouse:
    type: clickhouse
    inputs:
      - k8s_shape
      - vm_shape
    endpoint: http://clickhouse-logs.logging.svc.cluster.local:8123
    database: logs
    table: raw
    skip_unknown_fields: true
    date_time_best_effort: true
    auth:
      strategy: basic
      user: "${CLICKHOUSE_USER}"
      password: "${CLICKHOUSE_PASSWORD}"
    batch:
      max_events: 500000
      max_bytes: 134217728
      timeout_secs: 10
    # Stateless: in-memory buffer, block on full so back-pressure reaches the
    # JetStream pull source (which then stops acking). JetStream is durability.
    buffer:
      type: memory
      max_events: 2000
      when_full: block
    healthcheck:
      enabled: true