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
111 lines
2.9 KiB
Go
111 lines
2.9 KiB
Go
package index
|
|
|
|
import (
|
|
"regexp"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestSubjectToRegex(t *testing.T) {
|
|
cases := []struct {
|
|
glob string
|
|
match []string
|
|
nomatch []string
|
|
}{
|
|
{
|
|
glob: "logs.vm.*",
|
|
match: []string{"logs.vm.db-1", "logs.vm.web"},
|
|
nomatch: []string{"logs.vm", "logs.vm.db.1", "logs.k8s.x"},
|
|
},
|
|
{
|
|
glob: "logs.k8s.vault.>",
|
|
match: []string{"logs.k8s.vault.audit", "logs.k8s.vault.a.b", "logs.k8s.vault"},
|
|
nomatch: []string{"logs.k8s.shop.web", "logs.vm.x"},
|
|
},
|
|
{
|
|
glob: "logs.vm.db-1",
|
|
match: []string{"logs.vm.db-1"},
|
|
nomatch: []string{"logs.vm.db-2", "logs.vm.db-1.x"},
|
|
},
|
|
}
|
|
for _, c := range cases {
|
|
re := regexp.MustCompile(subjectToRegex(c.glob))
|
|
for _, s := range c.match {
|
|
if !re.MatchString(s) {
|
|
t.Errorf("%q -> %q should match %q", c.glob, re.String(), s)
|
|
}
|
|
}
|
|
for _, s := range c.nomatch {
|
|
if re.MatchString(s) {
|
|
t.Errorf("%q -> %q should NOT match %q", c.glob, re.String(), s)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestBuildSearchSQLFull(t *testing.T) {
|
|
from := time.Date(2026, 7, 26, 0, 0, 0, 0, time.UTC)
|
|
to := time.Date(2026, 7, 27, 0, 0, 0, 0, time.UTC)
|
|
sql, args := buildSearchSQL("logs", "archive_index", SearchQuery{
|
|
Subject: "logs.k8s.vault.>",
|
|
Host: "node-1",
|
|
From: from,
|
|
To: to,
|
|
Limit: 50,
|
|
})
|
|
if !strings.Contains(sql, "FROM logs.archive_index") {
|
|
t.Errorf("missing table: %s", sql)
|
|
}
|
|
for _, want := range []string{"match(subject, ?)", "has(hosts, ?)", "max_ts >= ?", "min_ts <= ?", "ORDER BY min_ts", "LIMIT ?"} {
|
|
if !strings.Contains(sql, want) {
|
|
t.Errorf("sql missing %q: %s", want, sql)
|
|
}
|
|
}
|
|
if len(args) != 5 {
|
|
t.Fatalf("args = %d, want 5: %v", len(args), args)
|
|
}
|
|
if args[1] != "node-1" {
|
|
t.Errorf("host arg = %v", args[1])
|
|
}
|
|
if args[4] != 50 {
|
|
t.Errorf("limit arg = %v", args[4])
|
|
}
|
|
}
|
|
|
|
func TestBuildSearchSQLEmpty(t *testing.T) {
|
|
sql, args := buildSearchSQL("logs", "archive_index", SearchQuery{})
|
|
if strings.Contains(sql, "WHERE") {
|
|
t.Errorf("empty query should have no WHERE: %s", sql)
|
|
}
|
|
if len(args) != 0 {
|
|
t.Errorf("args = %v, want none", args)
|
|
}
|
|
}
|
|
|
|
func TestBuildSearchSQLHostGlob(t *testing.T) {
|
|
sql, args := buildSearchSQL("logs", "archive_index", SearchQuery{Host: "db-*"})
|
|
if !strings.Contains(sql, "arrayExists(h -> match(h, ?), hosts)") {
|
|
t.Errorf("host glob should use arrayExists/match: %s", sql)
|
|
}
|
|
if len(args) != 1 {
|
|
t.Fatalf("args = %v", args)
|
|
}
|
|
re := regexp.MustCompile(args[0].(string))
|
|
if !re.MatchString("db-1") || re.MatchString("web-1") {
|
|
t.Errorf("host glob regex wrong: %q", args[0])
|
|
}
|
|
}
|
|
|
|
func TestDDLContainsKeyColumns(t *testing.T) {
|
|
ddl := CreateTableSQL("logs", "archive_index")
|
|
for _, col := range []string{"object_key", "subject", "hosts", "min_ts", "max_ts", "event_count", "key_fingerprint", "bloom_filter"} {
|
|
if !strings.Contains(ddl, col) {
|
|
t.Errorf("DDL missing %q", col)
|
|
}
|
|
}
|
|
if !strings.Contains(ddl, "IF NOT EXISTS") {
|
|
t.Errorf("DDL should be idempotent")
|
|
}
|
|
}
|