acache

README

internal/acache — Traces filter autocomplete cache

Microsecond service / operation / attribute-value autocomplete for the /traces (and /explore) filter pickers, served straight out of Redis instead of round-tripping ClickHouse. Redis runs as a pure cache (save "", appendonly no, maxmemory-policy allkeys-lru) — data loss is acceptable because every key is rebuildable from the span stream and reads fall back to CH on a miss.

Two halves: an ingestion side that folds spans into Redis sorted sets, and a read API that serves the pickers.

---

Why local pre-aggregation (not a ZINCRBY per span)

At 1B spans/day (~11.5k/s) a ZINCRBY +1 per span per facet would be tens of thousands of Redis round-trips per second. Instead ObserveSpan only updates an in-memory delta map under a brief lock (no I/O), and a background flusher drains the whole window into Redis with one pipeline every FlushEvery (default 2s). Collapsing a window of a (service, op, value) into a single ZINCRBY <delta> is mathematically identical to N×+1 but costs one network op. This is the production-correct shape of "batch all writes with a pipeline."

---

Redis key layout

Key	Type	Holds
coremetry:services	SET	service names
coremetry:services:rank	ZSET	service → frequency
coremetry:ops:<svc>	SET	operation names for a service
coremetry:ops:rank:<svc>	ZSET	operation → frequency
coremetry:attr:keys	SET	known attribute keys
coremetry:attr:<key>	ZSET	value → frequency (low/med cardinality)
coremetry:attr:card:<key>	HLL	approximate distinct count (high cardinality)

Cardinality policy decides, per attribute key, whether values are kept as a ranked ZSET (CardTrack), counted with a HyperLogLog (CardHLL), or ignored (CardSkip). It is swappable at runtime (SetPolicy) so a new attribute is a config change, never a code change.

Staleness — two strategies:

• default: sliding EXPIRE <key> 86400 re-applied on every flush that touches the key (a service unseen for 24h drops out wholesale);
• WindowMode: per-member time windows via a companion <key>:ts ZSET (member → lastSeenUnix) + a background sweeper (ZREMRANGEBYSCORE) that evicts individual stale members.

---

A) Ingestion — wiring the write path

ObserveSpan is fire-and-forget and never blocks on Redis, so hook it into the existing async ingest side-effect path (the consumer flusher runs on context.Background()).

// main.go — construct once, share the pooled client with internal/cache.
opts, _ := redis.ParseURL(cfg.Redis.URL)   // one parse
rdb := redis.NewClient(opts)               // one pool, shared below

acStore := acache.NewStore(rdb, acache.Options{
    Policy:     buildPolicy(cfg),  // from env CSV or system_settings (see §C)
    FlushEvery: 2 * time.Second,
    TTL:        24 * time.Hour,
    TopN:       1000,
})
acStore.Start(ctx)                          // launches the flusher goroutine

// internal/otlp ingester — call per span on the side-effect path.
func (ing *Ingester) addSpan(sp *chstore.Span) bool {
    ing.acache.ObserveSpan(sp)   // non-blocking: in-memory delta only
    return ing.Spans.Add(sp)
}

Disabled-store contract: acache.NewStore(nil, …) (or a NewStoreFromURL that failed to connect) returns a no-op store — every write is dropped, every read misses — so single-instance / Redis-down installs degrade exactly like cache.NewNoop.

// Standalone (own pool) if you don't want to thread the shared client:
acStore, err := acache.NewStoreFromURL(cfg.Redis.URL, acache.Options{})
if err != nil { log.Printf("[acache] %v — running disabled", err) } // still usable

---

B) Read API — autocomplete endpoints

Each getter returns a hit bool; on a miss the handler falls back to the existing CH-backed picker (DB fallback is optional and the caller decides). Response shapes stay drop-in compatible with the current pickers ({names,total,hasMore}, {scope,key,count}[], {value,count}[]).

// GET /api/service-names — try the cache, fall back to CH on a miss.
func (s *Server) getServiceNames(w http.ResponseWriter, r *http.Request) {
    q := r.URL.Query().Get("q")
    limit := atoiDefault(r.URL.Query().Get("limit"), 200)

    if names, total, hit := s.acache.GetServices(r.Context(), q, limit); hit {
        writeJSON(w, namesResp{Names: names, Total: total, HasMore: total > len(names)})
        return
    }
    // miss → existing serveCached + ClickHouse path
    s.serveServiceNamesFromCH(w, r, q, limit)
}

// GET /api/operation-names?service=…
names, total, hit := s.acache.GetOperations(r.Context(), svc, q, limit)

// GET /api/attribute-keys
keys, hit := s.acache.GetAttributeKeys(r.Context())   // []string

// GET /api/attribute-values?key=…
vals, approx, freeText, hit := s.acache.GetAttributeValues(r.Context(), key, q, limit)
if hit && freeText {
    // high-cardinality key: render a free-text input, not a dropdown.
    writeJSON(w, attrValuesResp{FreeText: true, ApproxDistinct: approx})
} else if hit {
    writeJSON(w, attrValuesResp{Values: vals}) // []{value,count}, ranked
}

---

C) Config — allowlist/denylist without code changes

The cardinality policy is an interface; build a StaticPolicy from whatever config source fits. Two options (the codebase supports both patterns):

Env CSV (boot-time, via the existing splitCSV helper):

func buildPolicy(cfg config.Config) acache.Policy {
    low  := splitCSV(os.Getenv("COREMETRY_ACACHE_LOWCARD_KEYS"))
    high := splitCSV(os.Getenv("COREMETRY_ACACHE_HIGHCARD_KEYS"))
    if len(low) == 0 && len(high) == 0 {
        return acache.DefaultPolicy()
    }
    return acache.NewStaticPolicy(low, high, acache.CardHLL)
}

Admin-editable system_settings (runtime, no restart — recommended since new attributes appear at runtime). Store an {low:[…], high:[…], default} JSON blob under a acache_policy key (mirror branding/pipeline_rules), load at boot, and on PUT /api/acache-policy (admin-gated + s.audit(...)) call acStore.SetPolicy(newPolicy) to swap it atomically:

acStore.SetPolicy(acache.NewStaticPolicy(blob.Low, blob.High, acache.CardHLL))

DefaultPolicy() ships a reasonable OTel-shaped allowlist (http/db/rpc/cloud low-card facets) and denylist (ids, urls, statements); unknown keys default to CardHLL so an unclassified key is counted but never stores values — bounding memory against a cardinality explosion.

---

Operational notes

• Connection pool: never opens a connection per call — NewStore takes an already-pooled *redis.Client; NewStoreFromURL dials one pool up front. Prefer sharing the internal/cache client so the binary keeps a single pool.
• Bounded memory: per-flush in-memory distinct values per key are capped (MaxDistinctPerKey, default 50k); the persistent value ZSET is trimmed to TopN (default 1000) on every flush; oversized values (MaxValLen, default 256B) are dropped before they reach the aggregator.
• Lock contention: ObserveSpan holds a single mutex only for map writes (microseconds). If a profile shows contention above ~50k spans/s, shard the aggregator by hash(serviceName) and merge at flush.
• See redis-acache.conf.example for the pure-cache Redis config.

Documentation

Overview

Package acache is the Traces filter autocomplete cache.

Goal: serve service names, operation names and common attribute *values* for the /traces (and /explore) filter pickers in microseconds, straight out of Redis, without round-tripping ClickHouse. Data loss is acceptable — Redis runs as a pure cache (save "", appendonly no, maxmemory-policy allkeys-lru). On a cold cache or a Redis blip every read reports a miss and the caller falls back to the existing CH-backed picker endpoints.

Two halves:

Ingestion (write): ObserveSpan(sp) is called per span on the ingest
  side-effect path. It does NOT touch Redis — it folds the span into an
  in-memory delta aggregator under a brief lock. A background flusher
  drains the accumulated deltas into Redis with ONE pipeline every
  FlushEvery. At ~11.5k spans/s (1B/day) a per-span ZINCRBY would be
  ~tens of thousands of Redis ops/s; local pre-aggregation collapses an
  entire flush window of a (service, op, attr-value) into a single
  ZINCRBY <delta>, which is mathematically identical to N×(+1) but
  costs one network op. This is the production-correct shape of "batch
  all writes with a pipeline".

Read (autocomplete): GetServices / GetOperations / GetAttributeKeys /
  GetAttributeValues read the sorted sets directly, ordered by frequency.
  Each returns a `hit bool`; a miss (cold key or Redis error) lets the
  HTTP handler fall back to ClickHouse — DB fallback is optional and the
  caller decides.

Redis key layout (all under the coremetry: namespace):

coremetry:services            SET   service names
coremetry:services:rank       ZSET  service -> frequency
coremetry:ops:<svc>           SET   operation names for a service
coremetry:ops:rank:<svc>      ZSET  operation -> frequency
coremetry:attr:keys           SET   known attribute keys
coremetry:attr:<key>          ZSET  value -> frequency   (low/med cardinality)
coremetry:attr:card:<key>     HLL   approximate distinct count (high cardinality)

Cardinality policy decides, per attribute key, whether values are kept as a ranked ZSET (Track), counted approximately with a HyperLogLog (HLL), or ignored (Skip). The policy is swappable at runtime (SetPolicy) so adding a new attribute never requires a code change — wire it from an env CSV or from the admin-editable system_settings blob.

Index

• type Card
• type Options
• type Policy
• type PolicyFunc
  • func (f PolicyFunc) Classify(k string) Card
• type StaticPolicy
  • func DefaultPolicy() *StaticPolicy
  • func NewStaticPolicy(lowCard, highCard []string, def Card) *StaticPolicy
  • func (p *StaticPolicy) Classify(k string) Card
• type Store
  • func NewStore(cli *redis.Client, opt Options) *Store
  • func NewStoreFromURL(url string, opt Options) (*Store, error)
  • func (s *Store) Enabled() bool
  • func (s *Store) Flush(ctx context.Context) error
  • func (s *Store) GetAttributeKeys(ctx context.Context) (keys []string, hit bool)
  • func (s *Store) GetAttributeValues(ctx context.Context, key, prefix string, limit int) (vals []ValueCount, approxCount int64, freeText bool, hit bool)
  • func (s *Store) GetOperations(ctx context.Context, svc, prefix string, limit int) (names []string, total int, hit bool)
  • func (s *Store) GetServices(ctx context.Context, prefix string, limit int) (names []string, total int, hit bool)
  • func (s *Store) ObserveSpan(sp *chstore.Span)
  • func (s *Store) SetPolicy(p Policy)
  • func (s *Store) Start(ctx context.Context)
  • func (s *Store) Sweep(ctx context.Context) error
• type ValueCount

Types

type Card

type Card uint8

Card is how a single attribute key's values are cached.

const (
	// CardTrack keeps the top-N values as a frequency-ranked ZSET. Use for
	// low/medium-cardinality keys whose values are useful in a dropdown
	// (http.route, db.system, cloud.region…).
	CardTrack Card = iota
	// CardHLL keeps only a HyperLogLog approximate distinct count — no values.
	// Use for high-cardinality keys (trace_id, k8s.pod.name, account ids):
	// the picker shows "free-text (~N distinct)" instead of a value list.
	CardHLL
	// CardSkip records the key under coremetry:attr:keys but stores nothing
	// about its values.
	CardSkip
)

type Options

type Options struct {
	Policy            Policy        // attribute-key classifier (default DefaultPolicy())
	FlushEvery        time.Duration // pipeline flush cadence (default 2s)
	TTL               time.Duration // sliding EXPIRE on every touched key (default 24h; 0 disables)
	TopN              int           // values kept per CardTrack key (default 1000)
	MaxValLen         int           // attribute values longer than this are ignored (default 256)
	MaxDistinctPerKey int           // per-flush in-memory distinct cap per attr key (default 50000)

	// WindowMode swaps the sliding-EXPIRE staleness strategy for per-member
	// time windows: every flush stamps member->now into a companion ZSET, and
	// a background sweeper drops members not seen within WindowSize. More
	// precise (a single stale operation is evicted without dropping the whole
	// service), at the cost of a parallel ZSET per ranked key. Default off.
	WindowMode bool
	WindowSize time.Duration // default 24h
	SweepEvery time.Duration // sweeper cadence (default 5m)
}

Options tune the cache. The zero value is invalid — use defaults via NewStore, which fills any unset field.

type Policy

type Policy interface {
	Classify(attrKey string) Card
}

Policy classifies an attribute key into a Card. Implementations must be safe for concurrent use (Classify is called on the hot ingest path).

type PolicyFunc

type PolicyFunc func(string) Card

PolicyFunc adapts a plain function to Policy.

func (PolicyFunc) Classify

func (f PolicyFunc) Classify(k string) Card

type StaticPolicy

type StaticPolicy struct {
	Default Card
	// contains filtered or unexported fields
}

StaticPolicy resolves keys against an allowlist (Track) and a denylist (HLL); everything else falls to Default. Build it from config (splitCSV env vars) or from a system_settings JSON blob so new attributes are a config change, not a code change.

func DefaultPolicy

func DefaultPolicy() *StaticPolicy

DefaultPolicy is a sensible starting allowlist/denylist for an OTel-shaped span stream. Unknown keys default to CardHLL — we count them (so the picker can say "free-text") but never store their values, which bounds memory against a cardinality explosion from an un-classified key.

func NewStaticPolicy

func NewStaticPolicy(lowCard, highCard []string, def Card) *StaticPolicy

NewStaticPolicy builds a policy from a low/med-cardinality allowlist and a high-cardinality denylist. def is applied to keys in neither list.

func (*StaticPolicy) Classify

func (p *StaticPolicy) Classify(k string) Card

type Store

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

Store is the autocomplete cache. A nil *redis.Client (or a NewStoreFromURL that failed to connect) yields a disabled store: every write is a no-op and every read reports a miss, so callers degrade to ClickHouse cleanly — the same graceful-degradation contract as cache.NewNoop.

func NewStore

func NewStore(cli *redis.Client, opt Options) *Store

NewStore wraps an already-constructed, pooled *redis.Client. This is the preferred constructor inside the binary: build the client once and share it with internal/cache so both layers ride one connection pool. Pass cli == nil to get a disabled (no-op) store.

func NewStoreFromURL

func NewStoreFromURL(url string, opt Options) (*Store, error)

NewStoreFromURL is the standalone constructor: it parses a redis:// URL and dials its own pooled client, mirroring internal/cache.New (parse → NewClient → 3s PING). On parse/ping failure it returns a *disabled* store plus the error, so a caller that ignores the error still gets a working no-op store.

Inside Coremetry prefer NewStore(sharedClient, …) to avoid a second pool.

func (*Store) Enabled

func (s *Store) Enabled() bool

Enabled reports whether the store talks to a live Redis.

func (*Store) Flush

func (s *Store) Flush(ctx context.Context) error

Flush drains the current delta window into Redis with a single pipeline. Exported so callers can force a flush (tests, graceful shutdown). Safe to call concurrently with ObserveSpan.

func (*Store) GetAttributeKeys

func (s *Store) GetAttributeKeys(ctx context.Context) (keys []string, hit bool)

GetAttributeKeys returns the known attribute keys (sorted). hit is false on a cold cache or error.

func (*Store) GetAttributeValues

func (s *Store) GetAttributeValues(ctx context.Context, key, prefix string, limit int) (vals []ValueCount, approxCount int64, freeText bool, hit bool)

GetAttributeValues returns cached values for an attribute key.

• CardTrack key: a frequency-ranked value list (prefix-filtered), freeText false, approxCount 0.
• CardHLL key: no values; approxCount is the HyperLogLog distinct estimate and freeText is true, signalling the picker to render a free-text input ("~N distinct values") instead of a dropdown.
• CardSkip key (or cold): hit false → caller falls back to ClickHouse.

func (*Store) GetOperations

func (s *Store) GetOperations(ctx context.Context, svc, prefix string, limit int) (names []string, total int, hit bool)

GetOperations returns operation names for a service, ordered by frequency.

func (*Store) GetServices

func (s *Store) GetServices(ctx context.Context, prefix string, limit int) (names []string, total int, hit bool)

GetServices returns service names ordered by frequency. If prefix is set it is matched case-insensitively (substring, or glob when it contains * / ?). total is the number of matches; hit is false on a cold cache or Redis error so the caller can fall back to ClickHouse.

func (*Store) ObserveSpan

func (s *Store) ObserveSpan(sp *chstore.Span)

ObserveSpan folds a span into the delta aggregator. It is non-blocking (a brief mutex around map writes only — no Redis I/O) and safe to call from the fire-and-forget ingest side-effect path with context.Background().

func (*Store) SetPolicy

func (s *Store) SetPolicy(p Policy)

SetPolicy swaps the cardinality policy atomically. Use it to apply an admin-edited allowlist/denylist at runtime without restarting.

func (*Store) Start

func (s *Store) Start(ctx context.Context)

Start launches the background flusher (and, in WindowMode, the staleness sweeper). Both stop when ctx is cancelled, after a final drain. No-op on a disabled store.

func (*Store) Sweep

func (s *Store) Sweep(ctx context.Context) error

Sweep evicts members not seen within WindowSize. For each registered ranked key it reads the stale members from the companion :ts ZSET, ZREMs them from the rank ZSET, SREMs them from the companion set, then prunes the :ts ZSET. No-op unless WindowMode is on.

type ValueCount

type ValueCount struct {
	Value string `json:"value"`
	Count int64  `json:"count"`
}

ValueCount is a single attribute value plus its observed frequency.