From 1b7df2ca26afed20b694e2b1950cc2ab1d6bd8e8 Mon Sep 17 00:00:00 2001 From: Amit Kumar Date: Thu, 18 Jun 2026 11:27:52 +0000 Subject: [PATCH] fix(graphrag): bound SignalStore.LogClusters (TTL + per-tenant cap) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LogClusters was the one unbounded structure on the in-memory hot path: an insert-only map keyed by service×Drain-template-ID, explicitly skipped by Prune. Template IDs shift as Drain generalizes and the key carries the service, so the map grew with every distinct (service, template-id) pair ever seen — unbounded by the shared 50k Drain LRU and by log volume. Under a sustained high-shape log stream (exactly the case when INFO is kept in-memory but not persisted) this leaks RAM. Extend SignalStore.Prune to bound LogClusters the same way it bounds Metrics: - TTL: evict clusters with LastSeen older than the 24h signalRetention cutoff (ages out the template-ID accretion as orphaned clusters go stale). - Cap: maxLogClustersPerTenant=10000 backstop, oldest-LastSeen first. - Evicted clusters' EMITTED_BY / LOGGED_DURING edges are swept (collected by cluster id; removed even when not yet time-stale, e.g. cap-evicted). Prune signature gains maxLogClusters; refresh.go passes the new constant. Tests: rewrote _SweepsStaleLogClusterEdgesOnly (which asserted the old never-pruned behavior) into _DropsStaleLogClustersKeepsFresh; added _LogClusterCapEvictsOldest. Verified: go build/vet, golangci-lint exit 0, go test -race ./internal/graphrag (109), full suite 679, gofmt clean. Closes the last unbounded structure on the in-memory path. CPU stays bounded by the fixed 16-worker pool + drop-when-full event queue; RAM by GOMEMLIMIT (75% budget) + per-tenant caps/TTLs on TraceStore (500k+TTL), SignalStore metrics (2k) + now log clusters (10k), topology LRU (100k), TSDB cardinality, and 24h idle-tenant eviction. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01LDQVJrixs2nJoea67a8pEG --- CLAUDE.md | 4 +- internal/graphrag/refresh.go | 15 +++++-- internal/graphrag/signal_prune_test.go | 62 ++++++++++++++++++++------ internal/graphrag/store.go | 53 +++++++++++++++++----- 4 files changed, 105 insertions(+), 29 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index be0a305..0626c80 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -70,7 +70,7 @@ The `internal/graphrag/` package is the core intelligence layer. It replaces the |-------|-------|-------|-----| | `ServiceStore` | ServiceNode, OperationNode | CALLS, EXPOSES | Permanent | | `TraceStore` | TraceNode, SpanNode | CONTAINS, CHILD_OF | Configurable (default 1h) | -| `SignalStore` | LogClusterNode, MetricNode | EMITTED_BY, MEASURED_BY, LOGGED_DURING | Permanent | +| `SignalStore` | LogClusterNode, MetricNode | EMITTED_BY, MEASURED_BY, LOGGED_DURING | 24h TTL + per-tenant caps | | `AnomalyStore` | AnomalyNode | PRECEDED_BY, TRIGGERED_BY | 24h | ### Node Types (7) @@ -219,7 +219,7 @@ Key settings in `internal/config/config.go`: - `LOG_FTS_ENABLED` — when truthy (`true`/`yes`/`on`/`1`), provisions the SQLite FTS5 `logs_fts` virtual table + sync triggers at startup; when false, log-search uses a 24h-clamped LIKE fallback. **Defaults to `true` when `DB_DRIVER=sqlite`** (BM25 is dramatically faster than LIKE on the kept `search_logs` MCP tool) and `false` otherwise. Toggle off and reclaim the ~30% disk overhead via `POST /api/admin/drop_fts` (refused while the flag is on). The vectordb-backed semantic-search path was removed on 2026-05-24. - `DLQ_MAX_FILES` (1000), `DLQ_MAX_DISK_MB` (500), `DLQ_MAX_RETRIES` (10) - `GRAPHRAG_WORKER_COUNT` (16), `GRAPHRAG_EVENT_QUEUE_SIZE` (100000; **10000 on SQLite**) — sized for 100–200 services; raise further if `otelcontext_graphrag_events_dropped_total` climbs -- `GRAPHRAG_TRACE_TTL` (`1h`; **`30m` on SQLite**), `GRAPHRAG_MAX_SPANS_PER_TENANT` (500000), `GRAPHRAG_TENANT_IDLE_TTL` (`24h`) — in-memory GraphRAG memory bounds. Spans past the per-tenant cap are skipped from the graph only (DB unaffected; metered as `otelcontext_graphrag_events_dropped_total{signal="span_capacity"}`); tenant store slices idle past the TTL are evicted (default tenant immune, self-healing via the 60s rebuild). SignalStore metrics are bounded to 2000/tenant + 24h TTL (constants). +- `GRAPHRAG_TRACE_TTL` (`1h`; **`30m` on SQLite**), `GRAPHRAG_MAX_SPANS_PER_TENANT` (500000), `GRAPHRAG_TENANT_IDLE_TTL` (`24h`) — in-memory GraphRAG memory bounds. Spans past the per-tenant cap are skipped from the graph only (DB unaffected; metered as `otelcontext_graphrag_events_dropped_total{signal="span_capacity"}`); tenant store slices idle past the TTL are evicted (default tenant immune, self-healing via the 60s rebuild). SignalStore is bounded per-tenant by a 24h TTL + a cap: metrics 2000/tenant, log clusters 10000/tenant (constants; the log-cluster cap was added because clusters key on service×Drain-template-ID and would otherwise grow unbounded as template IDs churn). - `PPROF_ADDR` (`127.0.0.1:6060`) — `net/http/pprof` on a dedicated loopback listener (never the public mux); empty disables. Startup also sets a soft `GOMEMLIMIT` (honors the env var, else 75% of the cgroup/host budget via `internal/membudget`). - `INGEST_MIN_SEVERITY` (`INFO`), `STORE_MIN_SEVERITY` (**defaults to `"WARN"` for all drivers**; `""` falls back to same-as-ingest) — two-tier log severity gate. The ingest gate runs at the OTLP receiver and **drops the log entirely** below the threshold (no in-memory enrichment either). The store gate runs at the persist boundary inside the async pipeline (`internal/ingest/pipeline.go:process`) and **only skips the DB row write** — the log still flows through `LogCallback` so GraphRAG Drain template mining and span/trace correlation see it. By default (`STORE_MIN_SEVERITY=WARN`, `INGEST_MIN_SEVERITY=INFO`) INFO logs reach GraphRAG/Drain in-memory but are **not** persisted — the DB only grows with WARN+. To also analyse DEBUG in-memory (not just INFO), set `INGEST_MIN_SEVERITY=DEBUG` (raises in-memory event volume). Setting `STORE_MIN_SEVERITY` ≤ `INGEST_MIN_SEVERITY` is a no-op (logged as a warning at startup). Drops surface via `Pipeline.Stats().StoreFiltered`. - `INGEST_ASYNC_ENABLED` (true), `INGEST_PIPELINE_QUEUE_SIZE` (50000), `INGEST_PIPELINE_WORKERS` (8), `INGEST_PIPELINE_MAX_BYTES` (536870912 = 512 MB; **128 MB on SQLite**) — async ingest pipeline (`internal/ingest/pipeline.go`). Hybrid backpressure: <90% accept all, 90–100% drop healthy batches (errors/slow always pass), 100% return gRPC `RESOURCE_EXHAUSTED`. The byte cap bounds queue memory regardless of item count — at the cap even priority batches get `RESOURCE_EXHAUSTED`/429 (a 429 is recoverable, an OOM kill is not); watch `otelcontext_ingest_pipeline_queue_bytes` and reason `bytes_full`. Set `INGEST_ASYNC_ENABLED=false` to revert to synchronous DB writes inside `Export()`. Drops surface as `otelcontext_ingest_pipeline_dropped_total{signal,reason}`. diff --git a/internal/graphrag/refresh.go b/internal/graphrag/refresh.go index 11ef375..3b759ab 100644 --- a/internal/graphrag/refresh.go +++ b/internal/graphrag/refresh.go @@ -15,6 +15,13 @@ const ( // maxMetricsPerTenant caps each tenant's SignalStore metric map; past // the cap the oldest-LastSeen series are evicted first. maxMetricsPerTenant = 2000 + // maxLogClustersPerTenant caps each tenant's SignalStore LogCluster map. + // Clusters are keyed by service×Drain-template-ID, so without a cap the + // map outgrows the (shared) Drain template LRU as template IDs churn and + // services multiply. The 24h signalRetention TTL is the primary bound; + // this cap is the backstop for high-shape-cardinality streams (~1KB/node + // => ~10MB/tenant worst case). + maxLogClustersPerTenant = 10000 // rebuildRowLimit caps how many span rows a single per-tenant rebuild // pass loads. Hitting it is logged — topology lags until the next tick. rebuildRowLimit = 50000 @@ -45,17 +52,17 @@ func (g *GraphRAG) refreshLoop(ctx context.Context) { case <-ticker.C: g.rebuildAllTenantsFromDB(ctx) pruned := 0 - prunedMetrics := 0 + prunedSignals := 0 signalCutoff := time.Now().Add(-signalRetention) for _, stores := range g.snapshotTenants() { pruned += stores.traces.Prune() - prunedMetrics += stores.signals.Prune(signalCutoff, maxMetricsPerTenant) + prunedSignals += stores.signals.Prune(signalCutoff, maxMetricsPerTenant, maxLogClustersPerTenant) } if pruned > 0 { slog.Debug("GraphRAG pruned expired traces/spans", "count", pruned) } - if prunedMetrics > 0 { - slog.Debug("GraphRAG pruned stale/over-cap metrics", "count", prunedMetrics) + if prunedSignals > 0 { + slog.Debug("GraphRAG pruned stale/over-cap signals (metrics+clusters)", "count", prunedSignals) } g.pruneOldAnomalies() if evicted := g.evictIdleTenants(); evicted > 0 { diff --git a/internal/graphrag/signal_prune_test.go b/internal/graphrag/signal_prune_test.go index 925f07d..98c31ca 100644 --- a/internal/graphrag/signal_prune_test.go +++ b/internal/graphrag/signal_prune_test.go @@ -17,7 +17,7 @@ func TestSignalStore_Prune_DropsStaleMetricsKeepsFresh(t *testing.T) { ss.UpsertMetric("cpu", "old-svc", 1.0, stale) ss.UpsertMetric("cpu", "new-svc", 1.0, now) - if got := ss.Prune(now.Add(-24*time.Hour), 0); got != 1 { + if got := ss.Prune(now.Add(-24*time.Hour), 0, 0); got != 1 { t.Fatalf("Prune removed %d metrics, want 1", got) } if _, ok := ss.Metrics["cpu|old-svc"]; ok { @@ -46,7 +46,7 @@ func TestSignalStore_Prune_CapEvictsOldestFirst(t *testing.T) { ss.UpsertMetric(fmt.Sprintf("m%d", i), "svc", 1.0, now.Add(time.Duration(i)*time.Minute)) } - if got := ss.Prune(now.Add(-24*time.Hour), 2); got != 3 { + if got := ss.Prune(now.Add(-24*time.Hour), 2, 0); got != 3 { t.Fatalf("Prune removed %d metrics, want 3", got) } for i := 0; i < 3; i++ { @@ -65,10 +65,12 @@ func TestSignalStore_Prune_CapEvictsOldestFirst(t *testing.T) { } } -// TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly proves LogClusters -// themselves are untouched (they are Drain-bounded upstream) while their -// stale EMITTED_BY / LOGGED_DURING edges are swept. -func TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly(t *testing.T) { +// TestSignalStore_Prune_DropsStaleLogClustersKeepsFresh proves the TTL bound on +// LogClusters: a cluster idle past the cutoff is removed together with its +// EMITTED_BY / LOGGED_DURING edges, while a fresh cluster and its edge survive. +// LogClusters were previously never pruned (only their stale edges were swept), +// which leaked the map under a sustained high-shape log stream. +func TestSignalStore_Prune_DropsStaleLogClustersKeepsFresh(t *testing.T) { ss := newSignalStore() now := time.Now() stale := now.Add(-48 * time.Hour) @@ -77,19 +79,53 @@ func TestSignalStore_Prune_SweepsStaleLogClusterEdgesOnly(t *testing.T) { ss.AddLoggedDuringEdge("c-old", "span-old", stale) ss.UpsertLogCluster("c-new", "tmpl", "ERROR", "svc2", now) - ss.Prune(now.Add(-24*time.Hour), 0) + ss.Prune(now.Add(-24*time.Hour), 0, 0) - if len(ss.LogClusters) != 2 { - t.Fatalf("LogClusters len = %d, want 2 — Prune must not touch clusters", len(ss.LogClusters)) + if _, ok := ss.LogClusters["c-old"]; ok { + t.Errorf("stale LogCluster survived Prune") + } + if _, ok := ss.LogClusters["c-new"]; !ok { + t.Errorf("fresh LogCluster was pruned") } if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, "c-old", "svc")]; ok { - t.Errorf("stale EMITTED_BY edge survived") + t.Errorf("evicted cluster's EMITTED_BY edge survived") } if _, ok := ss.Edges[edgeKey(EdgeLoggedDuring, "c-old", "span-old")]; ok { - t.Errorf("stale LOGGED_DURING edge survived") + t.Errorf("evicted cluster's LOGGED_DURING edge survived") } if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, "c-new", "svc2")]; !ok { - t.Errorf("fresh EMITTED_BY edge was swept") + t.Errorf("fresh cluster's EMITTED_BY edge was swept") + } +} + +// TestSignalStore_Prune_LogClusterCapEvictsOldest proves the cap backstop: when +// the cluster map still exceeds maxLogClusters after the TTL pass, the +// oldest-LastSeen clusters are evicted first — with their edges — even though +// all of them are fresh. +func TestSignalStore_Prune_LogClusterCapEvictsOldest(t *testing.T) { + ss := newSignalStore() + now := time.Now() + + for i := 0; i < 5; i++ { + id := fmt.Sprintf("c%d", i) + ss.UpsertLogCluster(id, "tmpl", "ERROR", "svc", now.Add(time.Duration(i)*time.Minute)) + } + + ss.Prune(now.Add(-24*time.Hour), 0, 2) + + for i := 0; i < 3; i++ { + id := fmt.Sprintf("c%d", i) + if _, ok := ss.LogClusters[id]; ok { + t.Errorf("oldest cluster %s survived cap eviction", id) + } + if _, ok := ss.Edges[edgeKey(EdgeEmittedBy, id, "svc")]; ok { + t.Errorf("cap-evicted cluster %s left a dangling EMITTED_BY edge", id) + } + } + for i := 3; i < 5; i++ { + if _, ok := ss.LogClusters[fmt.Sprintf("c%d", i)]; !ok { + t.Errorf("newest cluster c%d was evicted — cap must drop oldest first", i) + } } } @@ -110,7 +146,7 @@ func TestSignalStore_UpsertRefreshesEdgeTimestamps(t *testing.T) { ss.AddLoggedDuringEdge("c1", "span-1", stale) ss.AddLoggedDuringEdge("c1", "span-1", now) - ss.Prune(now.Add(-24*time.Hour), 0) + ss.Prune(now.Add(-24*time.Hour), 0, 0) for _, ek := range []string{ edgeKey(EdgeMeasuredBy, "cpu|svc", "svc"), diff --git a/internal/graphrag/store.go b/internal/graphrag/store.go index a229a7c..803cdeb 100644 --- a/internal/graphrag/store.go +++ b/internal/graphrag/store.go @@ -603,15 +603,16 @@ func (ss *SignalStore) LogClustersForService(service string) []*LogClusterNode { return out } -// Prune bounds the SignalStore (modeled on TraceStore.Prune): MetricNodes -// whose LastSeen predates cutoff are removed; if the map still exceeds -// maxMetrics (<=0 = uncapped) the oldest-LastSeen overflow is evicted. Each -// removed metric takes its MEASURED_BY edge with it. Finally, Edges whose -// UpdatedAt predates cutoff are swept — the upsert paths refresh edge -// timestamps, so live correlations survive. LogClusters are NOT pruned -// here: they are bounded upstream by the Drain template LRU; only their -// stale edges go. Returns the number of metrics removed. -func (ss *SignalStore) Prune(cutoff time.Time, maxMetrics int) int { +// Prune bounds the SignalStore (modeled on TraceStore.Prune): MetricNodes and +// LogClusterNodes whose LastSeen predates cutoff are removed; if either map +// still exceeds its cap (maxMetrics / maxLogClusters; <=0 = uncapped) the +// oldest-LastSeen overflow is evicted. Each removed node takes its edges with +// it — metric MEASURED_BY edges inline, evicted-cluster EMITTED_BY / +// LOGGED_DURING edges in the final sweep — and any edge whose UpdatedAt +// predates cutoff is swept too (upsert paths refresh edge timestamps, so live +// correlations survive). Returns the number of nodes (metrics + clusters) +// removed. +func (ss *SignalStore) Prune(cutoff time.Time, maxMetrics, maxLogClusters int) int { ss.mu.Lock() defer ss.mu.Unlock() @@ -640,8 +641,40 @@ func (ss *SignalStore) Prune(cutoff time.Time, maxMetrics int) int { pruned++ } } + // LogClusters: TTL + cap, mirroring the metric bound above. This map was + // previously unbounded — keyed by service×Drain-template-ID with no cap, it + // grew with every distinct (service, template-id) pair ever seen (template + // IDs shift as Drain generalizes), leaking under a sustained high-shape log + // stream. The TTL ages out that accretion; the cap is the backstop. Evicted + // cluster IDs are collected so their EMITTED_BY / LOGGED_DURING edges + // (FromID = cluster id) are swept below even when not yet time-stale. + evictedClusters := make(map[string]struct{}) + for id, lc := range ss.LogClusters { + if lc.LastSeen.Before(cutoff) { + delete(ss.LogClusters, id) + evictedClusters[id] = struct{}{} + pruned++ + } + } + if maxLogClusters > 0 && len(ss.LogClusters) > maxLogClusters { + type clusterAge struct { + id string + seen time.Time + } + byAge := make([]clusterAge, 0, len(ss.LogClusters)) + for id, lc := range ss.LogClusters { + byAge = append(byAge, clusterAge{id, lc.LastSeen}) + } + sort.Slice(byAge, func(i, j int) bool { return byAge[i].seen.Before(byAge[j].seen) }) + for _, e := range byAge[:len(byAge)-maxLogClusters] { + delete(ss.LogClusters, e.id) + evictedClusters[e.id] = struct{}{} + pruned++ + } + } + for ek, e := range ss.Edges { - if e.UpdatedAt.Before(cutoff) { + if _, fromEvicted := evictedClusters[e.FromID]; e.UpdatedAt.Before(cutoff) || fromEvicted { delete(ss.Edges, ek) } }