cache

Documentation

Overview

Package cache provides a thin abstraction over Redis (or none) for two scaling primitives the rest of Coremetry depends on:

• Cache: short-TTL hot read cache, used by API handlers in front of ClickHouse for endpoints that get polled at high frequency.
• Lock: distributed lock with token-based release, used by background workers (evaluator, anomaly detector) so multiple Coremetry replicas don't all run the same scheduled work.

Both have a Noop fallback so the binary keeps running unchanged when Redis is not configured (single-instance dev / hobby deployments).

Index

• func LeaderTTL(interval time.Duration) time.Duration
• func New(url string) (Cache, Lock, error)
• func NewNoop() (Cache, Lock)
• func StartRedisReprobe(ctx context.Context, interval time.Duration, ...)
• type Cache
• type LeaderHolder
  • func NewLeaderHolder(lock Lock, key string, ttl time.Duration) *LeaderHolder
  • func (h *LeaderHolder) IsLeader() bool
  • func (h *LeaderHolder) Start(ctx context.Context)
• type Lock
• type RedisStats
• type SwitchableCache
  • func NewSwitchableCache(c Cache) *SwitchableCache
  • func (s *SwitchableCache) Current() Cache
  • func (s *SwitchableCache) Del(ctx context.Context, key string) error
  • func (s *SwitchableCache) DelPrefix(ctx context.Context, prefix string) error
  • func (s *SwitchableCache) Get(ctx context.Context, key string) ([]byte, bool, error)
  • func (s *SwitchableCache) MGet(ctx context.Context, keys []string) ([][]byte, error)
  • func (s *SwitchableCache) Ping(ctx context.Context) error
  • func (s *SwitchableCache) Publish(ctx context.Context, channel string, msg []byte) error
  • func (s *SwitchableCache) ScanPrefix(ctx context.Context, prefix string) ([][]byte, error)
  • func (s *SwitchableCache) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error
  • func (s *SwitchableCache) SetNX(ctx context.Context, key string, value []byte, ttl time.Duration) (bool, error)
  • func (s *SwitchableCache) Stats(ctx context.Context) (RedisStats, error)
  • func (s *SwitchableCache) Subscribe(ctx context.Context, channel string) (<-chan []byte, error)
  • func (s *SwitchableCache) Swap(c Cache)
• type SwitchableLock
  • func NewSwitchableLock(l Lock) *SwitchableLock
  • func (s *SwitchableLock) Current() Lock
  • func (s *SwitchableLock) Refresh(ctx context.Context, key string, ttl time.Duration) (bool, error)
  • func (s *SwitchableLock) Release(ctx context.Context, key string) error
  • func (s *SwitchableLock) Swap(l Lock)
  • func (s *SwitchableLock) TryAcquire(ctx context.Context, key string, ttl time.Duration) (bool, error)

Functions

func LeaderTTL

func LeaderTTL(interval time.Duration) time.Duration

LeaderTTL picks a sensible lease TTL given a worker's tick interval. Bounded so very short ticks (10s) still get a 30s floor (avoids thrashing on Redis blips) and very long ticks (hourly retention sweep) don't get an hour-long lease (failover must stay bounded even when the worker itself runs rarely).

Rule: TTL = clamp(3×interval, 30s, 10min). Refresh fires at TTL/3, so a 30s TTL refreshes every 10s; a 10min TTL refreshes every ~3min.

func New

func New(url string) (Cache, Lock, error)

New parses a Redis URL (redis://host:port/db) and returns a working Cache+Lock pair. On URL parse error or initial PING failure it falls back to the Noop implementation and returns the error so the caller can log it — Coremetry should not crash just because Redis is unhealthy.

func NewNoop

func NewNoop() (Cache, Lock)

NewNoop returns a Cache+Lock pair that does nothing for cache and always grants the lock. Used when Redis is not configured.

func StartRedisReprobe

func StartRedisReprobe(ctx context.Context, interval time.Duration, connect func() (Cache, Lock, error), cs *SwitchableCache, ls *SwitchableLock, onRecovered func())

StartRedisReprobe launches the boot-failure recovery goroutine — v0.8.341 (H4). Started from main.go ONLY when Redis was configured but the boot ping failed (the lockDegraded state, v0.8.212): retries `connect` every `interval` (jittered ±20% so a fleet that lost Redis together doesn't stampede it on recovery); on the first success it swaps the real impls into cs/ls, fires onRecovered (clears the /admin/stats lockDegraded flag + re-kicks pub/sub subscribers), and STOPS — the probe exists to end the degraded boot state once, not to health-check Redis forever (steady-state failures surface per-call as before).

connect is injected (rather than calling New(url) directly) so the swap-once semantics are unit-testable without a Redis.

Types

type Cache

type Cache interface {
	Get(ctx context.Context, key string) ([]byte, bool, error)
	Set(ctx context.Context, key string, value []byte, ttl time.Duration) error
	// SetNX sets key=value only when the key doesn't already exist
	// (Redis SET NX semantics) and reports whether the claim was won.
	// A lightweight cross-pod dedup primitive (v0.8.350 — the
	// dependencies-cache warmer claims one warm cycle per fleet window),
	// NOT a lock: no token, no release, the TTL is the sole lifetime.
	// For contended long-lived leadership use Lock/LeaderHolder instead.
	// Noop returns true — a single-instance deployment always "wins".
	SetNX(ctx context.Context, key string, value []byte, ttl time.Duration) (bool, error)
	Del(ctx context.Context, key string) error
	// ScanPrefix returns every value whose key matches the given
	// prefix (Redis SCAN MATCH + MGET). Returned values are the
	// stored bytes; expired / missing keys silently drop out.
	// Used by the cluster membership service (v0.5.253) to list
	// every live pod's heartbeat without exposing the raw Redis
	// client through the abstraction. Noop returns (nil, nil) so
	// single-instance dev falls back to a 1-member view.
	ScanPrefix(ctx context.Context, prefix string) ([][]byte, error)
	// MGet batch-reads the given keys in ONE round trip (Redis MGET).
	// The result is positionally aligned with keys: missing / expired
	// keys yield a nil slot, never an error. Added v0.8.403 for the
	// user-presence read (/api/users enriches every row with an
	// online/lastSeenAt stamp) — a per-key Get loop would turn one
	// admin page load into N sequential round trips. Noop returns
	// all-nil slots so presence degrades to "unknown" without Redis.
	MGet(ctx context.Context, keys []string) ([][]byte, error)
	// Ping reports liveness of the underlying cache. Noop returns nil
	// (treats cache-disabled mode as healthy — there's no remote to be
	// down). Used by the status page.
	Ping(ctx context.Context) error
	// Stats returns Redis INFO + DBSIZE for the System page admin
	// view. Noop returns a zero-valued struct so callers can render
	// "Redis not configured" without error handling.
	Stats(ctx context.Context) (RedisStats, error)
	// Publish broadcasts msg on a Redis pub/sub channel. Used by
	// the cross-pod L1 cache invalidation flow (v0.5.337):
	// a mutating endpoint DELs L2 + publishes the key, every
	// pod's Subscribe loop receives the key and evicts L1. Noop
	// returns nil — single-instance pods have no peers to notify.
	Publish(ctx context.Context, channel string, msg []byte) error
	// Subscribe returns a channel of incoming pub/sub messages
	// for the given channel name. The returned chan closes when
	// ctx is cancelled. Noop returns a chan that never delivers
	// (and closes on ctx cancellation), so the caller's
	// invalidation goroutine sits idle without busy-looping.
	Subscribe(ctx context.Context, channel string) (<-chan []byte, error)
	// DelPrefix evicts every key whose name starts with prefix
	// (v0.6.11 — bug-fix). Used by the L2 invalidation path
	// when an operator mutation affects many keys at once
	// (e.g. "topology-edges:*" — one mute change invalidates
	// every time-window-keyed topology view). Pre-v0.6.11 the
	// L2 invalidator did a SCAN and discarded the keys, so L2
	// was never actually drained — operator-reported staleness
	// after exclude/mute changes was the symptom. Noop returns
	// nil (single-instance pods have no L2). Redis impl uses
	// cursor-paginated SCAN + batched UNLINK so a runaway
	// prefix doesn't pin the client.
	DelPrefix(ctx context.Context, prefix string) error
}

Cache is a read-through cache. Get returns ok=true on a hit; ok=false on a miss is NOT an error.

type LeaderHolder

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

LeaderHolder — v0.5.426. Designates ONE pod as the leader for a given background worker via a heartbeat-refreshed Redis lock. Replaces the prior per-tick TryAcquire/Release pattern, which at N pods caused each worker to alternate execution across all N (correct but log-noisy + N× CH load on workers that don't actually need parallel execution).

Lifecycle:

1. NewLeaderHolder(lock, key, ttl) returns a non-running holder. ttl is the lease TTL; refresh runs at ttl/3.

2. Start(ctx) launches the heartbeat goroutine. The goroutine immediately attempts TryAcquire, then enters the refresh loop. ctx.Done() cleanly releases the lock and exits.

3. IsLeader() reports the current state. Background workers check this at the top of each tick and skip when false:

   if !leader.IsLeader() { return } // ... do work ...

4. The held state flips back to false when Refresh reports the lease is definitively lost (ok=false) OR — v0.8.341 — when Refresh ERRORS have gone on longer than the lease TTL (Redis partition / AUTH flap: we can no longer prove we hold it, and the lease HAS expired server-side, so a peer may legitimately be leader). The workers stop running; the holder falls back into the acquire loop and leadership picks back up when Redis returns.

Bounded behaviour at N pods: exactly one pod is leader at any moment (subject to lease TTL crossover during failover — same guarantee Kubernetes leader election provides). When the leader pod dies, another acquires within ttl of the next poll.

Per-worker key: each background worker (errors-inbox, anomaly recorder, topology aggregator, …) holds its OWN LeaderHolder with a unique key. Different pods can lead different workers; this matches the prior per-worker lockKey structure + keeps failover granular.

func NewLeaderHolder

func NewLeaderHolder(lock Lock, key string, ttl time.Duration) *LeaderHolder

NewLeaderHolder returns a holder for the given lock + key. ttl is the Redis key TTL while held — the holder refreshes at ttl/3. Pick ttl long enough that a pod restart doesn't thrash leadership (30-60s is typical) but short enough that a crashed pod doesn't block leadership for too long. Use LeaderTTL(interval) for a sensible per-worker default.

func (*LeaderHolder) IsLeader

func (h *LeaderHolder) IsLeader() bool

IsLeader returns true when this pod currently holds the lock. Cheap (atomic load) — safe to call in tick hot paths.

func (*LeaderHolder) Start

func (h *LeaderHolder) Start(ctx context.Context)

Start launches the heartbeat goroutine. Safe to call once; repeated calls are a no-op (sync.Once). Caller is expected to hold the goroutine open via ctx — typically the same ctx driving the rest of the background workers.

type Lock

type Lock interface {
	TryAcquire(ctx context.Context, key string, ttl time.Duration) (ok bool, err error)
	Release(ctx context.Context, key string) error
	// v0.5.426 — Refresh extends the TTL on a key WE already hold.
	// Returns false when our token no longer owns the key (i.e.
	// we lost leadership — lease expired and someone else acquired).
	// Used by LeaderHolder to keep long-lived leadership without
	// the release+reacquire race window of the prior per-tick
	// pattern.
	Refresh(ctx context.Context, key string, ttl time.Duration) (ok bool, err error)
}

Lock is a best-effort distributed lock. TryAcquire returns ok=false (without error) when the lock is already held — callers should skip their work, not retry. Release is safe to call from any code path (defer-friendly).

type RedisStats

type RedisStats struct {
	Version             string  `json:"version"`
	Mode                string  `json:"mode"` // standalone|sentinel|cluster
	Uptime              int64   `json:"uptimeSec"`
	ConnectedClients    int     `json:"connectedClients"`
	Keys                int64   `json:"keys"`
	UsedMemoryBytes     int64   `json:"usedMemoryBytes"`
	UsedMemoryPeakBytes int64   `json:"usedMemoryPeakBytes"`
	MaxMemoryBytes      int64   `json:"maxMemoryBytes"`
	HitRate             float64 `json:"hitRate"`   // keyspace_hits / (hits+misses), 0..1
	OpsPerSec           float64 `json:"opsPerSec"` // instantaneous_ops_per_sec
	NetInputKBps        float64 `json:"netInputKbps"`
	NetOutputKBps       float64 `json:"netOutputKbps"`
	EvictedKeys         int64   `json:"evictedKeys"`
	ExpiredKeys         int64   `json:"expiredKeys"`
}

RedisStats is the slice of INFO + DBSIZE the System page renders. Kept small so one trip per Stats() call covers the panel without streaming the full INFO blob (which is hundreds of fields). All fields are zeroed on parse failure — UI shows "—" for those rows rather than crashing the panel.

type SwitchableCache

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

SwitchableCache / SwitchableLock wrap the live Cache/Lock behind an atomic.Pointer so main.go can swap the Noop boot-fallback for the real Redis impl at runtime without a pod restart — v0.8.341 (H4). Mirrors logstore.Switchable (internal/logstore/switchable.go), which does the same for the logs read backend.

Why: pre-v0.8.341, cache.New pinged Redis ONCE at boot (3s); on failure main wired the Noop cache + Noop always-leader lock for the pod's LIFETIME. After Redis recovered, N pods all stayed "leader" forever — duplicate evaluators / notifications / retention sweeps — and the L2 cache + cross-pod SSE bridge stayed dead. Every consumer (LeaderHolders, api.Server, sse bridge, cluster membership) captures its Cache/Lock at construction, so the fix is a stable wrapper they can all hold while the inner impl is hot-swapped by the background re-probe (StartRedisReprobe).

Wired ALWAYS — healthy boots wrap the real impl from the start, so the only cost on the happy path is one atomic pointer load per call.

func NewSwitchableCache

func NewSwitchableCache(c Cache) *SwitchableCache

func (*SwitchableCache) Current

func (s *SwitchableCache) Current() Cache

Current returns the live inner cache. Callers must not hold it across requests — that would defeat the swap.

func (*SwitchableCache) Del

func (s *SwitchableCache) Del(ctx context.Context, key string) error

func (*SwitchableCache) DelPrefix

func (s *SwitchableCache) DelPrefix(ctx context.Context, prefix string) error

func (*SwitchableCache) Get

func (s *SwitchableCache) Get(ctx context.Context, key string) ([]byte, bool, error)

func (*SwitchableCache) MGet

func (s *SwitchableCache) MGet(ctx context.Context, keys []string) ([][]byte, error)

func (*SwitchableCache) Ping

func (s *SwitchableCache) Ping(ctx context.Context) error

func (*SwitchableCache) Publish

func (s *SwitchableCache) Publish(ctx context.Context, channel string, msg []byte) error

func (*SwitchableCache) ScanPrefix

func (s *SwitchableCache) ScanPrefix(ctx context.Context, prefix string) ([][]byte, error)

func (*SwitchableCache) Set

func (s *SwitchableCache) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error

func (*SwitchableCache) SetNX

func (s *SwitchableCache) SetNX(ctx context.Context, key string, value []byte, ttl time.Duration) (bool, error)

func (*SwitchableCache) Stats

func (s *SwitchableCache) Stats(ctx context.Context) (RedisStats, error)

func (*SwitchableCache) Subscribe

func (s *SwitchableCache) Subscribe(ctx context.Context, channel string) (<-chan []byte, error)

Subscribe binds to the inner impl live at CALL time — a subscription opened while the inner is Noop stays on the Noop channel after a swap (the Noop chan only closes on ctx cancel). main.go's recovery callback therefore re-kicks the two boot-time subscribers (SSE bridge + L1 invalidation) so they re-subscribe against the real Redis.

func (*SwitchableCache) Swap

func (s *SwitchableCache) Swap(c Cache)

Swap atomically replaces the inner cache. In-flight calls finish on the impl they started with; nil is ignored (a failed probe must never leave consumers with a nil cache — same rule as logstore.Switchable.Swap).

type SwitchableLock

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

SwitchableLock — see SwitchableCache. LeaderHolder binds its Lock at construction (NewLeaderHolder captures it), so holders wired over a SwitchableLock transparently start hitting real Redis on their next heartbeat after a Swap — no reconstruction, no restart.

func NewSwitchableLock

func NewSwitchableLock(l Lock) *SwitchableLock

func (*SwitchableLock) Current

func (s *SwitchableLock) Current() Lock

Current returns the live inner lock.

func (*SwitchableLock) Refresh

func (s *SwitchableLock) Refresh(ctx context.Context, key string, ttl time.Duration) (bool, error)

func (*SwitchableLock) Release

func (s *SwitchableLock) Release(ctx context.Context, key string) error

func (*SwitchableLock) Swap

func (s *SwitchableLock) Swap(l Lock)

Swap atomically replaces the inner lock; nil is ignored.

Convergence window — v0.8.341 (H4): while degraded, every pod's Noop lock said "always leader" (unchanged single-instance semantics). After the swap, each holder's next heartbeat calls Refresh on the REAL redisLock, which has no token registered for the key → returns (false, nil) → the holder drops held and races TryAcquire (SetNX) against its peers. Exactly one pod wins; the fleet converges from N leaders to 1 within one refresh cadence (ttl/3, ≤ ~3min worst case for the 10min TTL cap).

func (*SwitchableLock) TryAcquire

func (s *SwitchableLock) TryAcquire(ctx context.Context, key string, ttl time.Duration) (bool, error)