// Package index writes and queries the ClickHouse archive index — one row per // stored S3 object — so operators can answer "which objects hold vault logs // from host X between Y and Z" without scanning S3. The concrete store is // behind the Index interface so the archiver and CLI test against a fake. package index import ( "context" "crypto/tls" "fmt" "time" "github.com/ClickHouse/clickhouse-go/v2" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" ) // Row is one archive-index record. type Row struct { ObjectKey string Bucket string Subject string Hosts []string MinTS time.Time MaxTS time.Time EventCount uint64 RawBytes uint64 StoredBytes uint64 Compression string Cipher string ContainerFormat string KeyName string KeyFingerprint string } // Result is one row returned by Search (a subset relevant to retrieval). type Result struct { ObjectKey string Bucket string Subject string Hosts []string MinTS time.Time MaxTS time.Time EventCount uint64 RawBytes uint64 StoredBytes uint64 KeyName string KeyFingerprint string } // Index is the archive-index surface. type Index interface { Insert(ctx context.Context, row Row) error Search(ctx context.Context, q SearchQuery) ([]Result, error) InitSchema(ctx context.Context) error Ping(ctx context.Context) error Close() error } // Config configures the ClickHouse client. type Config struct { Address string // host:port (native protocol, 9000) Database string Table string Username string Password string TLS bool } // ClickHouse is the ClickHouse-backed Index. type ClickHouse struct { conn driver.Conn database string table string } // NewClickHouse connects to ClickHouse. func NewClickHouse(ctx context.Context, cfg Config) (*ClickHouse, error) { opts := &clickhouse.Options{ Addr: []string{cfg.Address}, Auth: clickhouse.Auth{ Database: cfg.Database, Username: cfg.Username, Password: cfg.Password, }, } if cfg.TLS { opts.TLS = &tls.Config{MinVersion: tls.VersionTLS12} } conn, err := clickhouse.Open(opts) if err != nil { return nil, fmt.Errorf("open clickhouse: %w", err) } ch := &ClickHouse{conn: conn, database: cfg.Database, table: cfg.Table} return ch, nil } // Ping verifies connectivity. func (c *ClickHouse) Ping(ctx context.Context) error { return c.conn.Ping(ctx) } // Close closes the connection. func (c *ClickHouse) Close() error { return c.conn.Close() } // InitSchema creates the database and table if absent (idempotent). func (c *ClickHouse) InitSchema(ctx context.Context) error { if err := c.conn.Exec(ctx, CreateDatabaseSQL(c.database)); err != nil { return fmt.Errorf("create database: %w", err) } if err := c.conn.Exec(ctx, CreateTableSQL(c.database, c.table)); err != nil { return fmt.Errorf("create table: %w", err) } return nil } // Insert writes one row. func (c *ClickHouse) Insert(ctx context.Context, row Row) error { sql := fmt.Sprintf( "INSERT INTO %s.%s (object_key, bucket, subject, hosts, min_ts, max_ts, event_count, raw_bytes, stored_bytes, compression, cipher, container_format, key_name, key_fingerprint) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", c.database, c.table) err := c.conn.Exec(ctx, sql, row.ObjectKey, row.Bucket, row.Subject, row.Hosts, row.MinTS, row.MaxTS, row.EventCount, row.RawBytes, row.StoredBytes, row.Compression, row.Cipher, row.ContainerFormat, row.KeyName, row.KeyFingerprint, ) if err != nil { return fmt.Errorf("insert index row: %w", err) } return nil } // Search runs the parameterized query built from q. func (c *ClickHouse) Search(ctx context.Context, q SearchQuery) ([]Result, error) { sql, args := buildSearchSQL(c.database, c.table, q) rows, err := c.conn.Query(ctx, sql, args...) if err != nil { return nil, fmt.Errorf("search index: %w", err) } defer func() { _ = rows.Close() }() var out []Result for rows.Next() { var r Result if err := rows.Scan( &r.ObjectKey, &r.Bucket, &r.Subject, &r.Hosts, &r.MinTS, &r.MaxTS, &r.EventCount, &r.RawBytes, &r.StoredBytes, &r.KeyName, &r.KeyFingerprint, ); err != nil { return nil, fmt.Errorf("scan result: %w", err) } out = append(out, r) } if err := rows.Err(); err != nil { return nil, fmt.Errorf("iterate results: %w", err) } return out, nil }