benvin
c05ccfcb5d
logarchiver replaces the plain Vector archiver leg of the centralized logging stack (argocd-apps #296) with a Go service that archives raw logs from NATS JetStream to S3 as zstd-compressed, OpenPGP-encrypted, indexed objects, plus an operator CLI to search the index and retrieve/decrypt archived logs. It adds the things that outgrew Vector: zstd compression, encryption keyed from Ben's Vault GPG secrets engine, a searchable ClickHouse index, and sink-conditional acks (a batch is acknowledged to JetStream only after the object is durably in S3 AND indexed). Service (`logarchiver run`): - Durable JetStream pull consumer (stream LOGS, durable archiver, subject filter default logs.k8s.vault.>), explicit acks, independent offsets. - Batch per subject by size/count/time -> NDJSON -> zstd -> encrypt -> S3 PUT -> ClickHouse index row -> ack. On any failure the batch is Nak'd and redelivered, so nothing is lost on a sink outage. - Encryption is a wrapped-DEK envelope (container LARC1): the bulk is AES-256-GCM framed under a random data key, and only that 32-byte key is OpenPGP-encrypted to the engine's public key. This is because the Vault GPG engine does whole-payload decrypt only; retrieval round-trips just the tiny wrapped key regardless of object size. Public key fetched from the engine or a mounted file (configurable); key fingerprint recorded per object; periodic pubkey refresh for rotation. - Prometheus metrics, structured slog, graceful drain on shutdown. CLI: - `search` queries the index (subject/host/time) and lists matching objects. - `fetch` downloads, decrypts via the Vault GPG engine, unzstds and emits NDJSON (optionally re-filtered by host/time). - `init-schema` creates/prints the ClickHouse archive_index DDL. - cobra `completion` subcommands. Config via file+env (k8s-friendly, secrets from env), boundaries (NATS/S3/ ClickHouse/Vault) behind interfaces with unit tests (config, batching, host/subject extraction, crypto roundtrip with a test key, ack-after-persist with fakes, search query building). go build/vet/test -race clean; golangci-lint v2 clean. Woodpecker CI: build/test/pre-commit on PR; on v* tag a container image plus a Gitea binary release + rpm-internal RPM. Docs per subcommand + architecture + retrieval runbook + deployment drop-in. Claude-Session: https://claude.ai/code/session_015ur3i7D2azsMAWTSVABApv
81 lines
2.4 KiB
Go
81 lines
2.4 KiB
Go
package consumer
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"crypto/x509"
|
|
"fmt"
|
|
"os"
|
|
"time"
|
|
|
|
"git.unkin.net/unkin/logarchiver/internal/config"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// Connect dials NATS as the configured user and returns the connection and a
|
|
// JetStream context. Callers must Close the returned *nats.Conn.
|
|
func Connect(cfg config.NATSConfig) (*nats.Conn, jetstream.JetStream, error) {
|
|
opts := []nats.Option{
|
|
nats.Name("logarchiver"),
|
|
nats.MaxReconnects(-1),
|
|
nats.ReconnectWait(2 * time.Second),
|
|
}
|
|
if cfg.User != "" {
|
|
opts = append(opts, nats.UserInfo(cfg.User, cfg.Password))
|
|
}
|
|
if cfg.CAFile != "" {
|
|
pool := x509.NewCertPool()
|
|
pem, err := os.ReadFile(cfg.CAFile)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("read nats ca %s: %w", cfg.CAFile, err)
|
|
}
|
|
if !pool.AppendCertsFromPEM(pem) {
|
|
return nil, nil, fmt.Errorf("no certs parsed from nats ca %s", cfg.CAFile)
|
|
}
|
|
opts = append(opts, nats.Secure(&tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}))
|
|
}
|
|
nc, err := nats.Connect(cfg.URL, opts...)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("connect nats %s: %w", cfg.URL, err)
|
|
}
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
nc.Close()
|
|
return nil, nil, fmt.Errorf("jetstream context: %w", err)
|
|
}
|
|
return nc, js, nil
|
|
}
|
|
|
|
// EnsureConsumer creates or updates the durable pull consumer on the stream with
|
|
// the configured subject filters. Independent offsets and explicit acks give
|
|
// logarchiver at-least-once delivery decoupled from the transform tier.
|
|
func EnsureConsumer(ctx context.Context, js jetstream.JetStream, cfg config.NATSConfig) (jetstream.Consumer, error) {
|
|
ackWait := cfg.AckWait
|
|
if ackWait <= 0 {
|
|
ackWait = 2 * time.Minute
|
|
}
|
|
consCfg := jetstream.ConsumerConfig{
|
|
Durable: cfg.Durable,
|
|
Name: cfg.Durable,
|
|
AckPolicy: jetstream.AckExplicitPolicy,
|
|
DeliverPolicy: jetstream.DeliverAllPolicy,
|
|
AckWait: ackWait,
|
|
MaxDeliver: -1,
|
|
ReplayPolicy: jetstream.ReplayInstantPolicy,
|
|
}
|
|
switch len(cfg.Subjects) {
|
|
case 0:
|
|
return nil, fmt.Errorf("no subject filters configured")
|
|
case 1:
|
|
consCfg.FilterSubject = cfg.Subjects[0]
|
|
default:
|
|
consCfg.FilterSubjects = cfg.Subjects
|
|
}
|
|
cons, err := js.CreateOrUpdateConsumer(ctx, cfg.Stream, consCfg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ensure consumer %s on stream %s: %w", cfg.Durable, cfg.Stream, err)
|
|
}
|
|
return cons, nil
|
|
}
|