logarchiver/docs/architecture.md

# Architecture

logarchiver is one Go binary with a service mode (`run`) and an operator CLI
(`search`/`fetch`/`init-schema`). Boundaries to external systems (NATS, S3,
ClickHouse, Vault) sit behind interfaces so they can be faked in tests.

## Packages

| Package | Responsibility |
|---|---|
| `internal/config` | Config struct, YAML+env loading, validation, defaults tuned for the logging stack. |
| `internal/event` | Extract host + timestamp from a raw JSON event (k8s and VM shapes), subject sanitization. |
| `internal/batcher` | Group events per subject into batches; decide flush by size/count/age; carry ack tokens. |
| `internal/crypto` | The LARC1 container: zstd + framed AES-256-GCM + OpenPGP-wrapped data key. Seal/Open. |
| `internal/vaultgpg` | Thin client for `vault-plugin-secrets-gpg`: fetch public key, decrypt wrapped data key. |
| `internal/s3store` | S3/Ceph RGW object storage behind an `ObjectStore` interface. |
| `internal/index` | ClickHouse archive index (`Index` interface), DDL, pure search-SQL builder. |
| `internal/archiver` | Orchestrates seal → S3 PUT → index write; object-key builder; public-key provider/refresh. |
| `internal/consumer` | JetStream binding + the fetch→batch→persist→ack loop (sink-conditional acks). |
| `internal/metrics` | Prometheus collectors and `/metrics`. |
| `internal/cli` | cobra command tree; wires everything for each subcommand. |

## Service data flow

1. **Consume.** A durable JetStream **pull** consumer (`LOGS` stream, durable
   `archiver`, subject filter default `logs.k8s.vault.>`) is created/updated at
   startup with explicit acks and independent offsets, so logarchiver's replay
   position is decoupled from the ClickHouse transform tier. Messages are pulled
   in batches.
2. **Route + batch.** Each message's payload is parsed just enough to extract
   the source **host** (`.host` → `.hostname` → `.kubernetes.pod_node_name`) and
   **timestamp** (`.timestamp` → `.ts`). Events are grouped by NATS **subject**.
   A batch is flushed when it hits `max_bytes` (64 MiB raw), `max_events`, or
   `max_age` (5 min) — whichever first — or on shutdown.
3. **Seal.** The batch is rendered to NDJSON, compressed with zstd, and sealed
   into a **LARC1** object (see below).
4. **Store.** The object is `PUT` to S3 at
   `archive/<subject>/YYYY/MM/DD/<UTCstamp>-<rand>.ndjson.zst.larc`.
5. **Index.** One row per object is inserted into ClickHouse `logs.archive_index`.
6. **Ack.** Only now are the batch's JetStream messages acknowledged. If seal,
   S3, or index fails, the messages are Nak'd (with a backoff) and JetStream
   redelivers them — nothing is lost on a sink outage.

## The LARC1 container format

The Vault GPG engine can only decrypt a **whole** OpenPGP message inline (no
session-key extraction, no streaming — see the retrieval runbook). To keep
objects any size while making retrieval cheap, logarchiver does its own hybrid
encryption instead of PGP-encrypting the whole object:

```
+-----------------------------------------------------------+
| magic         "LARC1\n"  (6 bytes)                        |
| header_len    uint32 big-endian                           |
| header        JSON  { v, key_name, key_fingerprint,       |
|                       wrapped_dek_len, nonce_prefix,       |
|                       frame_size, compression, cipher }    |
| wrapped_dek   OpenPGP message encrypting the 32-byte DEK   |  <- only this goes to Vault
|               to the engine's public key (~hundreds of B)  |
| frames        repeated:  uint32 ct_len | ciphertext        |  <- zstd(NDJSON) in AES-256-GCM
|               terminated by a zero-length frame            |     frames; nonce = prefix||counter
+-----------------------------------------------------------+
```

- A fresh random 256-bit **DEK** and 4-byte nonce prefix are generated per
  object. Each frame is AES-256-GCM sealed with nonce `prefix||counter` and AAD
  = the frame counter (binds frame order; tampering fails the GCM tag).
- The DEK is OpenPGP-encrypted to the engine's public key. This wrapped blob is
  a small, standard OpenPGP message — the only thing ever sent to Vault's
  decrypt endpoint, regardless of object size.
- Compression order is NDJSON → zstd → encrypt, so decryption streams frames,
  GCM-decrypts, and feeds a streaming zstd decoder to emit NDJSON.

Because objects are a logarchiver-specific container (not a bare `gpg` file),
retrieval must go through `logarchiver fetch`. This is the deliberate cost of
supporting arbitrary object sizes against an engine that only whole-payload
decrypts.

## Index schema

`logs.archive_index` (one row per object), `MergeTree`,
`PARTITION BY toYYYYMM(min_ts)`, `ORDER BY (subject, min_ts, object_key)`, with a
`bloom_filter` skip index on `hosts`:

| column | type | notes |
|---|---|---|
| object_key | String | S3 key |
| bucket | LowCardinality(String) | e.g. `logs-archive` |
| subject | LowCardinality(String) | NATS subject |
| hosts | Array(LowCardinality(String)) | distinct source hosts |
| min_ts / max_ts | DateTime64(3) | event time range |
| event_count | UInt64 | |
| raw_bytes / stored_bytes | UInt64 | pre/post compression+encryption |
| compression / cipher / container_format | LowCardinality(String) | `zstd` / `AES-256-GCM` / `LARC1` |
| key_name / key_fingerprint | LowCardinality(String) / String | Vault GPG key + 40-hex fingerprint |
| created_at | DateTime64(3) DEFAULT now64(3) | |

Search overlaps the `[from,to]` window (`max_ts >= from AND min_ts <= to`),
matches the subject glob with an anchored regex (`*` = one token, `>` = rest),
and filters hosts with `has(hosts, …)` (exact) or `arrayExists(… match …)` (glob).

The DDL is owned in-cluster by the argocd bootstrap Job; `logarchiver
init-schema` applies the same statements and `--print` emits them. Keep
`schema/archive_index.sql` and `internal/index/ddl.go` in sync.

## Delivery guarantees

- **At-least-once**, sink-conditional. Acks happen only after S3 + index
  success. A failure after S3 but before index leaves an orphan object (still
  retrievable by prefix; reaped by bucket lifecycle) and triggers redelivery,
  which produces a fresh object — so an event may appear in two objects, never
  zero. Downstream consumers of the archive should treat events as at-least-once.
- **Graceful shutdown** drains open batches (bounded timeout) so in-flight
  events are persisted and acked before exit.

## Configuration & secrets

Config loads defaults < YAML file < env. Secrets are never in the file:
NATS password from `NATS_CONSUMER_PASSWORD`, S3 creds from the standard `AWS_*`
env (cephrgw `logs-archive-s3` secret), ClickHouse password from
`CLICKHOUSE_PASSWORD`, Vault via `VAULT_ADDR` + k8s auth (service) or ambient
token (CLI). See [deployment.md](deployment.md).