chanalytics

Documentation

Overview

Package chanalytics is a ClickHouse-backed analytics.Sink: a columnar, append-oriented store for cross-tenant fleet reporting. Three ReplacingMergeTree tables mirror the pganalytics schema; a packed _dedup column carries the version-gate and reprojection ordering. Every read is merge-independent (max/argMax/count-then-DELETE), so behavior matches the pganalytics reference without forcing background merges.

Index

• func LatestFactVersionForTest(ctx context.Context, s *Sink, tenantID, aggregate, aggID string) (int64, error)
• func TruncateForTest(ctx context.Context, s *Sink) error
• type Sink
  • func Open(ctx context.Context, dsn string) (*Sink, error)
  • func (s *Sink) AllWatermarks(ctx context.Context, tenantID string) ([]analytics.Watermark, error)
  • func (s *Sink) AppendEvents(ctx context.Context, events []analytics.Event) error
  • func (s *Sink) Close() error
  • func (s *Sink) LagByTenant(ctx context.Context) (map[string]float64, error)
  • func (s *Sink) MaintainPartitions(_ context.Context, _ time.Duration) (created, dropped int, err error)
  • func (s *Sink) PruneEvents(ctx context.Context, olderThan time.Time) (int64, error)
  • func (s *Sink) PurgeTenant(ctx context.Context, tenantID string) (int64, error)
  • func (s *Sink) ReprojectTenant(ctx context.Context, tenantID, aggregate string, ...) (int64, error)
  • func (s *Sink) SetWatermark(ctx context.Context, ws []analytics.Watermark) error
  • func (s *Sink) UpsertFacts(ctx context.Context, facts []analytics.Fact) error
  • func (s *Sink) Watermark(ctx context.Context, tenantID, aggregate, aggID string) (int64, error)

Functions

func LatestFactVersionForTest

func LatestFactVersionForTest(ctx context.Context, s *Sink, tenantID, aggregate, aggID string) (int64, error)

LatestFactVersionForTest returns the winning fact version for a key (argMax over _dedup). Test-only; proves the read-time version-gate.

func TruncateForTest

func TruncateForTest(ctx context.Context, s *Sink) error

TruncateForTest clears all analytics tables. Test-only.

Types

type Sink

type Sink struct {
	// contains filtered or unexported fields
}

Sink writes the analytics read model into ClickHouse. It ensures its own schema at Open (three CREATE TABLE IF NOT EXISTS), like pganalytics.

func Open

func Open(ctx context.Context, dsn string) (*Sink, error)

Open dials ClickHouse and ensures the schema. Ping proves reachability.

func (*Sink) AllWatermarks

func (s *Sink) AllWatermarks(ctx context.Context, tenantID string) ([]analytics.Watermark, error)

AllWatermarks returns every applied watermark for a tenant in one read.

func (*Sink) AppendEvents

func (s *Sink) AppendEvents(ctx context.Context, events []analytics.Event) error

AppendEvents dedupes on the natural key via the ORDER BY of the events RMT; re-appending the same (tenant, aggregate, agg_id, version) is a no-op. _dedup is 0 at ingest (only reprojection bumps it).

func (*Sink) Close

func (s *Sink) Close() error

Close releases the ClickHouse connection.

func (*Sink) LagByTenant

func (s *Sink) LagByTenant(ctx context.Context) (map[string]float64, error)

LagByTenant reports now() - (that tenant's newest fact commit time), in seconds, per tenant. The winning row per aggregate has the newest at, so max(at) needs no FINAL. An empty table yields no rows -> empty map.

func (*Sink) MaintainPartitions

func (s *Sink) MaintainPartitions(_ context.Context, _ time.Duration) (created, dropped int, err error)

func (*Sink) PruneEvents

func (s *Sink) PruneEvents(ctx context.Context, olderThan time.Time) (int64, error)

PruneEvents deletes history events with at < olderThan across all tenants and returns the number of logical events removed. Count-then-DELETE: the count is taken before the lightweight delete, so a re-run at the same cutoff returns 0.

func (*Sink) PurgeTenant

func (s *Sink) PurgeTenant(ctx context.Context, tenantID string) (int64, error)

PurgeTenant hard-deletes all of one tenant's rows across the three tables and returns the total logical rows removed. ClickHouse has no cross-table transaction, so this deletes table-by-table; it is idempotent and safe to re-run if interrupted.

func (*Sink) ReprojectTenant

func (s *Sink) ReprojectTenant(ctx context.Context, tenantID, aggregate string,
	transform func(payload json.RawMessage) (json.RawMessage, error)) (int64, error)

ReprojectTenant re-writes stored fact and event payloads for a tenant (and optional aggregate) through transform, in place. ClickHouse's ReplacingMergeTree can't UPDATE a row, so instead: read the CURRENT winning row per key via argMax(col, _dedup) (facts: payload, version, at, deleted; events: payload, type, at) plus max(_dedup), transform the payload in Go, and re-INSERT only the rows whose bytes actually changed with _dedup = max(_dedup) + 1, so the rewrite wins the next merge while argMax/max reads see it immediately. For facts (ingested at _dedup = version << dedupShift) the +1 stays inside that version's reprojection band, below the next version; for events (ingested at _dedup = 0, and keyed by version in the ORDER BY) the +1 is simply a reprojection tie-break. The count is computed by byte comparison in Go, so it is exact and idempotent regardless of merge state.

func (*Sink) SetWatermark

func (s *Sink) SetWatermark(ctx context.Context, ws []analytics.Watermark) error

SetWatermark inserts into the applied table (ReplacingMergeTree(version)); a max(version) read is monotonic, so a lower late insert is ignored.

func (*Sink) UpsertFacts

func (s *Sink) UpsertFacts(ctx context.Context, facts []analytics.Fact) error

UpsertFacts version-gates via _dedup: ReplacingMergeTree keeps the row with the max _dedup per (tenant, aggregate, agg_id), so a higher domain version always wins and a stale insert is shadowed. At ingest _dedup = packDedup(version) (version << dedupShift), so the low reprojection bits are 0.

func (*Sink) Watermark

func (s *Sink) Watermark(ctx context.Context, tenantID, aggregate, aggID string) (int64, error)

Watermark reads the highest applied version (0 if unknown). max() over an empty set returns 0 in ClickHouse, so the unknown case needs no special path.