From 11a88b4eb6de3aec09dc586e01e44cd11b203b8c Mon Sep 17 00:00:00 2001 From: Vishal Rana Date: Thu, 15 Oct 2026 12:50:21 -0700 Subject: [PATCH] fix: keep detector findings and writer byte accounting correct under stalls An audit of detector and writer behaviour under saturation reproduced these defects, and a review found one more: - Log patterns filtered and took extrema on the original OTLP timestamp, so logs with only an observed time disappeared. They now use canonical native event time. - A stalled service watermark froze log patterns. Anomaly evaluation is memoised per completed minute; patterns run on their own cadence and the snapshot is rebuilt. - A failed pattern query was published as a healthy snapshot and never retried. The error now propagates and the query is retried. - A pending annotation batch replaced before persistence lost a short incident. Displaced findings now merge into the replacement with their evaluated intervals. - Anomaly findings evaluated before a failed pattern query were dropped once the watermark advanced. They now carry over, bounded, to the next successful publication. - The writer released a request's byte reservation when its caller gave up while the request was still queued or committing, bypassing the decoded-byte budget. The reservation now moves with the request and is released exactly once when its payload is no longer held. Claude-Session: https://claude.ai/code/session_01XA52nWyQ3R5f5kK8kjpAQW --- internal/intelligence/annotations_test.go | 58 +++++ internal/intelligence/detector.go | 105 +++++--- .../intelligence/detector_recovery_test.go | 149 ++++++++++- internal/intelligence/parquet_test.go | 5 +- .../intelligence/pattern_recovery_test.go | 125 +++++++++ internal/intelligence/rollup_detector_test.go | 17 +- internal/telemetry/store/admission_test.go | 4 +- internal/telemetry/store/writer.go | 123 ++++++--- .../telemetry/store/writer_accounting_test.go | 243 ++++++++++++++++++ internal/telemetry/store/writer_test.go | 6 +- 10 files changed, 741 insertions(+), 94 deletions(-) create mode 100644 internal/intelligence/pattern_recovery_test.go create mode 100644 internal/telemetry/store/writer_accounting_test.go diff --git a/internal/intelligence/annotations_test.go b/internal/intelligence/annotations_test.go index bf4bb08a..f62603be 100644 --- a/internal/intelligence/annotations_test.go +++ b/internal/intelligence/annotations_test.go @@ -171,3 +171,61 @@ func TestDetectorWritesFailedFindingsWithTheNextBatch(t *testing.T) { t.Fatal("second write did not happen") } } + +func TestDetectorKeepsDisplacedPendingIncidentAndItsInterval(t *testing.T) { + d := NewDetector(&query.Duck{}, DefaultDetectorConfig()) + ctx, cancel := context.WithCancel(t.Context()) + entered := make(chan annotationBatch, 3) + release := make(chan struct{}) + done := make(chan struct{}) + go func() { + defer close(done) + d.persistAnnotations(ctx, func(ctx context.Context, findings []annotations.Anomaly, at time.Time) error { + entered <- annotationBatch{append([]annotations.Anomaly(nil), findings...), at} + select { + case <-release: + return nil + case <-ctx.Done(): + return ctx.Err() + } + }) + }() + defer func() { cancel(); <-done }() + now := time.Now().UTC() + d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now}) + select { + case <-entered: + case <-time.After(time.Second): + t.Fatal("persistence did not start") + } + incident := Anomaly{ServiceName: "api", Type: AnomalyLatencyDegradation, ZScore: 4, WindowStart: now.Add(-time.Minute), WindowEnd: now} + d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(time.Minute), Anomalies: []Anomaly{incident}}) + // A repeated live pattern refresh must not duplicate this finding. + d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(2 * time.Minute), Anomalies: []Anomaly{incident}}) + d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now.Add(3 * time.Minute)}) + close(release) + select { + case batch := <-entered: + if len(batch.findings) != 1 || !batch.findings[0].From.Equal(incident.WindowStart) || !batch.findings[0].To.Equal(incident.WindowEnd) || !batch.at.Equal(now.Add(3*time.Minute)) { + t.Fatalf("displaced incident was lost, duplicated or extended: %+v", batch) + } + case <-time.After(time.Second): + t.Fatal("pending incident was not persisted") + } +} + +func TestDetectorBoundsPendingFindings(t *testing.T) { + d := NewDetector(&query.Duck{}, DefaultDetectorConfig()) + now := time.Now().UTC() + for _, count := range []int{maxCarriedAnomalies + 1, 1} { + findings := make([]Anomaly, count) + for i := range findings { + findings[i] = Anomaly{Type: AnomalyLatencyDegradation, WindowStart: now.Add(time.Duration(i) * time.Minute), WindowEnd: now.Add(time.Duration(i+1) * time.Minute)} + } + d.publishSnapshot(IntelligenceSnapshot{GeneratedAt: now, Anomalies: findings}) + } + batch := <-d.annotationBatches + if len(batch.findings) != maxCarriedAnomalies || !batch.findings[len(batch.findings)-1].To.Equal(now.Add(time.Minute)) { + t.Fatalf("pending findings not bounded with newest retained: %d", len(batch.findings)) + } +} diff --git a/internal/intelligence/detector.go b/internal/intelligence/detector.go index 57f8e367..f7abd054 100644 --- a/internal/intelligence/detector.go +++ b/internal/intelligence/detector.go @@ -14,6 +14,7 @@ import ( "log/slog" "math" "runtime/debug" + "slices" "strings" "sync" "time" @@ -22,19 +23,22 @@ import ( "github.com/labstack/fanout/internal/query" ) -// isNoDataError returns true if the error indicates telemetry tables are empty/unavailable. -func isNoDataError(errMsg string) bool { - return strings.Contains(errMsg, "Table with name") -} - // Detector runs intelligence detection on observability data type Detector struct { - duck *query.Duck - config DetectorConfig - mu sync.RWMutex - snapshot *IntelligenceSnapshot - annotationMu sync.Mutex - annotationBatches chan annotationBatch + duck *query.Duck + config DetectorConfig + mu sync.RWMutex + snapshot *IntelligenceSnapshot + evaluationMu sync.Mutex + evaluation *anomalyEvaluation + annotationMu sync.Mutex + unpublishedFindings []annotations.Anomaly + annotationBatches chan annotationBatch +} + +type anomalyEvaluation struct { + through time.Time + findings []Anomaly } type annotationBatch struct { @@ -98,9 +102,6 @@ func (d *Detector) runCheck(ctx context.Context) { slog.Error("generate detector snapshot failed", "error", err) return } - if previous := d.LatestSnapshot(); previous != nil && snapshot.GeneratedAt.Equal(previous.GeneratedAt) { - return - } d.publishSnapshot(snapshot) anomalyCount := len(snapshot.Anomalies) @@ -119,22 +120,44 @@ func (d *Detector) publishSnapshot(snapshot IntelligenceSnapshot) { // while a newer cycle replaces the single pending batch. d.annotationMu.Lock() defer d.annotationMu.Unlock() + batch.findings = mergeAnnotationFindings(d.unpublishedFindings, batch.findings) + d.unpublishedFindings = nil select { case d.annotationBatches <- batch: return default: } select { - case <-d.annotationBatches: + case displaced := <-d.annotationBatches: + batch.findings = mergeAnnotationFindings(displaced.findings, batch.findings) default: } d.annotationBatches <- batch } -// maxCarriedAnomalies bounds findings kept across failed writes; it matches the -// anomaly history the store retains. +// maxCarriedAnomalies bounds unpublished findings, pending batches, and failed +// writes; it matches the anomaly history the store retains. const maxCarriedAnomalies = 10000 +// Preserve each evaluated interval, deduplicate repeated evaluations, and keep +// the newest findings when unpublished, pending, or failed writes reach the history limit. +func mergeAnnotationFindings(older, newer []annotations.Anomaly) []annotations.Anomaly { + capacity := min(maxCarriedAnomalies, len(older)+len(newer)) + merged := make([]annotations.Anomaly, 0, capacity) + seen := make(map[annotations.Anomaly]bool, capacity) + for _, findings := range [][]annotations.Anomaly{newer, older} { + for i := len(findings) - 1; i >= 0 && len(merged) < maxCarriedAnomalies; i-- { + finding := findings[i] + if !seen[finding] { + seen[finding] = true + merged = append(merged, finding) + } + } + } + slices.Reverse(merged) + return merged +} + func (d *Detector) persistAnnotations(ctx context.Context, record func(context.Context, []annotations.Anomaly, time.Time) error) { // Keep failed findings with their evaluated intervals, even when the // watermark stalls and no new batch arrives to trigger another write. @@ -156,7 +179,7 @@ func (d *Detector) persistAnnotations(ctx context.Context, record func(context.C case <-ctx.Done(): return case batch = <-d.annotationBatches: - batch.findings = append(carried.findings, batch.findings...) + batch.findings = mergeAnnotationFindings(carried.findings, batch.findings) case <-retryC: batch = carried } @@ -166,9 +189,6 @@ func (d *Detector) persistAnnotations(ctx context.Context, record func(context.C retryC = nil if err := record(ctx, batch.findings, batch.at); err != nil && ctx.Err() == nil { slog.Error("persist detector annotations", "error", err, "carried", len(batch.findings)) - if len(batch.findings) > maxCarriedAnomalies { - batch.findings = batch.findings[len(batch.findings)-maxCarriedAnomalies:] - } carried = batch if retry == nil { retry = time.NewTimer(delay) @@ -193,24 +213,36 @@ func (d *Detector) LatestSnapshot() *IntelligenceSnapshot { // GenerateSnapshot generates a complete intelligence snapshot func (d *Detector) GenerateSnapshot(ctx context.Context) (IntelligenceSnapshot, error) { + d.evaluationMu.Lock() + defer d.evaluationMu.Unlock() completed, err := d.duck.CompletedServiceRollupMinute(ctx) if err != nil { return IntelligenceSnapshot{}, fmt.Errorf("read detector service watermark: %w", err) } - if previous := d.LatestSnapshot(); previous != nil && !completed.After(previous.evaluatedThrough) { - return *previous, nil - } now := time.Now() lookback := now.Add(-d.config.LookbackWindow) - // Detect anomalies - anomalies, err := d.detectAnomalies(ctx, completed.Add(-d.config.LookbackWindow), completed) - if err != nil { - return IntelligenceSnapshot{}, err + // Only completed anomaly evaluations are memoized. Live patterns run on + // every wall-time check, even before the first rollup or while it stalls. + if d.evaluation == nil || !completed.Equal(d.evaluation.through) { + anomalies, err := d.detectAnomalies(ctx, completed.Add(-d.config.LookbackWindow), completed) + if err != nil { + return IntelligenceSnapshot{}, err + } + d.evaluation = &anomalyEvaluation{through: completed, findings: anomalies} } + anomalies := d.evaluation.findings // Detect log patterns - patterns := d.detectLogPatterns(ctx, lookback, now) + patterns, err := d.detectLogPatterns(ctx, lookback, now) + if err != nil { + // A later watermark may replace this evaluation before patterns recover. + // Keep its original intervals until a successful annotation publication. + d.annotationMu.Lock() + d.unpublishedFindings = mergeAnnotationFindings(d.unpublishedFindings, detectorAnnotations(IntelligenceSnapshot{Anomalies: anomalies}, d.duck.DefaultNamespace())) + d.annotationMu.Unlock() + return IntelligenceSnapshot{}, err + } // Calculate health score healthScore := d.calculateHealthScore(anomalies, patterns) @@ -585,7 +617,7 @@ func (d *Detector) readServiceAnomalies(ctx context.Context, statement string) ( } // detectLogPatterns groups repetitive log messages -func (d *Detector) detectLogPatterns(ctx context.Context, start, end time.Time) []Pattern { +func (d *Detector) detectLogPatterns(ctx context.Context, start, end time.Time) ([]Pattern, error) { startNano := start.UnixNano() endNano := end.UnixNano() namespace := d.duck.DefaultNamespace() @@ -598,10 +630,10 @@ func (d *Detector) detectLogPatterns(ctx context.Context, start, end time.Time) severity, service as service_name, COUNT(*) AS occurrence_count, - MIN(time_unix_nano) AS first_seen_nano, - MAX(time_unix_nano) AS last_seen_nano + epoch_ns(MIN(time)::TIMESTAMP_NS) AS first_seen_nano, + epoch_ns(MAX(time)::TIMESTAMP_NS) AS last_seen_nano FROM logs - WHERE time_unix_nano >= %d AND time_unix_nano < %d + WHERE time >= make_timestamp_ns(%d)::TIMESTAMPTZ_NS AND time < make_timestamp_ns(%d)::TIMESTAMPTZ_NS AND severity IN ('WARN', 'ERROR', 'FATAL') %s GROUP BY COALESCE(body_template, SUBSTRING(body, 1, 100)), severity, service @@ -612,10 +644,7 @@ func (d *Detector) detectLogPatterns(ctx context.Context, start, end time.Time) resp := d.duck.ExecuteSQL(ctx, query.SQLRequest{Query: sql}) if resp.Error != "" { - if !isNoDataError(resp.Error) { - slog.Error("pattern detection failed", "error", resp.Error) - } - return nil + return nil, fmt.Errorf("pattern detection: %s", resp.Error) } var patterns []Pattern @@ -637,7 +666,7 @@ func (d *Detector) detectLogPatterns(ctx context.Context, start, end time.Time) }) } - return patterns + return patterns, nil } // calculateHealthScore calculates overall system health (0-100, lower is worse) diff --git a/internal/intelligence/detector_recovery_test.go b/internal/intelligence/detector_recovery_test.go index f905fbff..c7b25ed3 100644 --- a/internal/intelligence/detector_recovery_test.go +++ b/internal/intelligence/detector_recovery_test.go @@ -79,6 +79,7 @@ func TestDetectorRetriesFailedQueriesAtUnchangedWatermark(t *testing.T) { if err := query.CreateCacheTables(db); err != nil { t.Fatal(err) } + createDetectorLogs(t, db) end := time.Now().UTC().Truncate(time.Minute).Add(-2 * time.Hour) seedDetectorRecoveryRollup(t, db, end) d := NewDetector(&query.Duck{DB: db}, DefaultDetectorConfig()) @@ -112,8 +113,8 @@ func TestDetectorRetriesFailedQueriesAtUnchangedWatermark(t *testing.T) { } } d.runCheck(t.Context()) - if d.LatestSnapshot() != got || len(d.annotationBatches) != 0 { - t.Fatal("successful minute was evaluated again") + if latest := d.LatestSnapshot(); latest == got || latest.Anomalies[0] != got.Anomalies[0] { + t.Fatal("live refresh changed the successful anomaly evaluation") } }) } @@ -180,9 +181,10 @@ func TestDetectorRetriesPersistenceAtUnchangedWatermark(t *testing.T) { } } d.runCheck(t.Context()) - if d.LatestSnapshot() != previous || len(d.annotationBatches) != 0 { - t.Fatal("stalled watermark queued another batch") + if latest := d.LatestSnapshot(); latest == previous || latest.Anomalies[0] != previous.Anomalies[0] { + t.Fatal("live refresh changed the stalled anomaly evaluation") } + refreshed := d.LatestSnapshot() unlock() unlock = nil if err := waitAttempt(); err != nil { @@ -197,7 +199,144 @@ func TestDetectorRetriesPersistenceAtUnchangedWatermark(t *testing.T) { t.Fatalf("retry changed evaluated interval: %+v", a) } } - if d.LatestSnapshot() != previous { + if d.LatestSnapshot() != refreshed { t.Fatal("persistence retry changed snapshot") } } + +func TestDetectorRetriesPatternFailureWithoutPublishingHealthySnapshot(t *testing.T) { + for _, withPrevious := range []bool{false, true} { + name := "startup" + if withPrevious { + name = "previous" + } + t.Run(name, func(t *testing.T) { + native, err := duckdb.NewConnector("", nil) + if err != nil { + t.Fatal(err) + } + defer native.Close() + connector := &detectorFailureConnector{Connector: native} + db := sql.OpenDB(connector) + defer db.Close() + if err := query.CreateCacheTables(db); err != nil { + t.Fatal(err) + } + createDetectorLogs(t, db) + end := time.Now().UTC().Truncate(time.Minute).Add(-time.Hour) + seedDetectorRecoveryRollup(t, db, end) + d := NewDetector(&query.Duck{DB: db}, DefaultDetectorConfig()) + if withPrevious { + d.runCheck(t.Context()) + <-d.annotationBatches + } + previous := d.LatestSnapshot() + at := time.Now().UTC().Add(-time.Minute) + if _, err := db.Exec(`INSERT INTO logs SELECT ?::TIMESTAMP_NS::TIMESTAMPTZ_NS, ?, 'failure', 'failure', 'ERROR', 'api', 'default' FROM range(10)`, at, at.UnixNano()); err != nil { + t.Fatal(err) + } + connector.fail = "AS occurrence_count" + if _, err := d.GenerateSnapshot(t.Context()); err == nil { + t.Fatal("pattern query failure was treated as an empty successful result") + } + d.runCheck(t.Context()) + if d.LatestSnapshot() != previous || len(d.annotationBatches) != 0 { + t.Fatal("pattern failure published a snapshot or annotations") + } + // The anomaly evaluation completed even if patterns failed. Recover + // patterns while refusing to re-run that completed anomaly query. + connector.fail = "AS current_rate" + d.runCheck(t.Context()) + got := d.LatestSnapshot() + if got == nil || got == previous || len(got.Patterns) != 1 || len(got.Anomalies) != 3 || got.HealthScore >= 100 { + t.Fatalf("pattern query was not retried at the same watermark: %+v", got) + } + }) + } +} + +func TestDetectorKeepsUnpublishedFindingsWhenWatermarkAdvances(t *testing.T) { + native, err := duckdb.NewConnector("", nil) + if err != nil { + t.Fatal(err) + } + defer native.Close() + connector := &detectorFailureConnector{Connector: native, fail: "AS occurrence_count"} + db := sql.OpenDB(connector) + defer db.Close() + if err := query.CreateCacheTables(db); err != nil { + t.Fatal(err) + } + createDetectorLogs(t, db) + end := time.Now().UTC().Truncate(time.Minute).Add(-time.Hour) + seedDetectorRecoveryRollup(t, db, end) + d := NewDetector(&query.Duck{DB: db}, DefaultDetectorConfig()) + for range 2 { + d.runCheck(t.Context()) + } + if d.LatestSnapshot() != nil || len(d.annotationBatches) != 0 { + t.Fatal("pattern failure published a snapshot or annotations") + } + if d.evaluation == nil || len(d.evaluation.findings) != 3 { + t.Fatalf("missing successful anomaly evaluation: %+v", d.evaluation) + } + want := detectorAnnotations(IntelligenceSnapshot{Anomalies: d.evaluation.findings}, d.duck.DefaultNamespace()) + + // Replace the failed-publication evaluation with a healthy, later window + // while patterns still fail, then recover at that new watermark. + advanced := end.Add(time.Minute) + if _, err := db.Exec(`UPDATE service_rollup SET error_rate=0,p95_ms=10,spans=100,served_spans=100`); err != nil { + t.Fatal(err) + } + if _, err := db.Exec(`UPDATE rollup_state SET last_ingested_unix_nano=? WHERE cache_key='service_rollup_v3'`, advanced.Add(30*time.Second).UnixNano()); err != nil { + t.Fatal(err) + } + d.runCheck(t.Context()) + if d.evaluation == nil || !d.evaluation.through.Equal(advanced) || len(d.evaluation.findings) != 0 || d.LatestSnapshot() != nil || len(d.annotationBatches) != 0 { + t.Fatalf("healthy evaluation did not replace the unpublished evaluation: %+v", d.evaluation) + } + connector.fail = "" + d.runCheck(t.Context()) + got := d.LatestSnapshot() + if got == nil || !got.evaluatedThrough.Equal(advanced) || len(got.Anomalies) != 0 || got.HealthScore != 100 { + t.Fatalf("recovered snapshot should reflect the healthy window: %+v", got) + } + batch := <-d.annotationBatches + if len(batch.findings) != len(want) { + t.Fatalf("unpublished incident lost or duplicated: %+v", batch.findings) + } + for i, finding := range batch.findings { + if finding != want[i] { + t.Fatalf("unpublished finding changed: got %+v, want %+v", finding, want[i]) + } + } + + // Persist the recovered batch through the real annotation storage path. + cfg := config.Config{DataDir: t.TempDir(), DuckDBMemory: "256MB", DuckDBThreads: 2, DuckDBMaxConns: 2} + repo, err := telemetrystore.Open(cfg.TelemetryDir()) + if err != nil { + t.Fatal(err) + } + defer repo.Close() + duck, err := query.NewDuck(t.Context(), cfg, repo) + if err != nil { + t.Fatal(err) + } + defer duck.Close() + if err := duck.RecordAnomalies(t.Context(), batch.findings, batch.at); err != nil { + t.Fatal(err) + } + persisted, err := annotations.New(duck).Read(t.Context(), annotations.Request{From: end.Add(-time.Hour), To: advanced.Add(time.Hour)}) + if err != nil || len(persisted.Anomalies) != len(want) { + t.Fatalf("persisted findings: %+v, %v", persisted, err) + } + for _, finding := range persisted.Anomalies { + if !finding.From.Equal(end.Add(-15*time.Minute)) || !finding.To.Equal(end) { + t.Fatalf("unpublished incident interval changed: %+v", finding) + } + } + d.runCheck(t.Context()) + if batch := <-d.annotationBatches; len(batch.findings) != 0 { + t.Fatalf("published findings were carried again: %+v", batch.findings) + } +} diff --git a/internal/intelligence/parquet_test.go b/internal/intelligence/parquet_test.go index 8c15dfd4..cac52263 100644 --- a/internal/intelligence/parquet_test.go +++ b/internal/intelligence/parquet_test.go @@ -99,7 +99,10 @@ func TestDetectorsReadNativeParquetTimestamps(t *testing.T) { } }) } - patterns := detector.detectLogPatterns(ctx, start, end) + patterns, err := detector.detectLogPatterns(ctx, start, end) + if err != nil { + t.Fatal(err) + } if len(patterns) != 1 || patterns[0].Count != 3 || patterns[0].FirstSeen.UnixNano() != start.UnixNano() { t.Fatalf("log patterns: %#v", patterns) } diff --git a/internal/intelligence/pattern_recovery_test.go b/internal/intelligence/pattern_recovery_test.go new file mode 100644 index 00000000..8bd66139 --- /dev/null +++ b/internal/intelligence/pattern_recovery_test.go @@ -0,0 +1,125 @@ +package intelligence + +import ( + "database/sql" + "testing" + "time" + + "github.com/labstack/fanout/internal/config" + "github.com/labstack/fanout/internal/query" + "github.com/labstack/fanout/internal/telemetry" + telemetrystore "github.com/labstack/fanout/internal/telemetry/store" +) + +func createDetectorLogs(t *testing.T, db *sql.DB) { + t.Helper() + db.SetMaxOpenConns(1) + if _, err := db.ExecContext(t.Context(), `SET TimeZone='UTC'`); err != nil { + t.Fatal(err) + } + if _, err := db.ExecContext(t.Context(), `CREATE TABLE logs ( + time TIMESTAMPTZ_NS, time_unix_nano BIGINT, body_template VARCHAR, + body VARCHAR, severity VARCHAR, service VARCHAR, namespace VARCHAR)`); err != nil { + t.Fatal(err) + } +} + +func TestLogPatternsIncludeObservedTimeAndExactEventBounds(t *testing.T) { + cfg := config.Config{DataDir: t.TempDir(), DuckDBMemory: "256MB", DuckDBThreads: 2, DuckDBMaxConns: 1} + repo, err := telemetrystore.Open(cfg.TelemetryDir()) + if err != nil { + t.Fatal(err) + } + defer repo.Close() + duck, err := query.NewDuck(t.Context(), cfg, repo) + if err != nil { + t.Fatal(err) + } + defer duck.Close() + start := time.Date(2026, 10, 1, 12, 0, 0, 123456789, time.UTC) + end := start.Add(100 * time.Nanosecond) + batch := telemetrystore.Batch{ID: "pattern-event-time"} + for _, observedOnly := range []bool{false, true} { + for _, at := range []time.Time{start.Add(-time.Nanosecond), start, start.Add(time.Nanosecond), end.Add(-time.Nanosecond), end} { + log := telemetry.Log{Namespace: "default", ServiceName: "api", Severity: "ERROR", Body: "original", BodyTemplate: "original", TimeUnixNanos: at.UnixNano()} + if observedOnly { + log.Body, log.BodyTemplate = "observed", "observed" + log.TimeUnixNanos, log.ObservedTimeNanos = 0, at.UnixNano() + } + batch.Logs = append(batch.Logs, log) + } + } + if err := repo.Commit(t.Context(), batch); err != nil { + t.Fatal(err) + } + d := NewDetector(duck, DefaultDetectorConfig()) + patterns, err := d.detectLogPatterns(t.Context(), start, end) + if err != nil { + t.Fatal(err) + } + if len(patterns) != 2 { + t.Fatalf("patterns=%+v, want original and observed timestamps", patterns) + } + for _, pattern := range patterns { + if pattern.Count != 3 || !pattern.FirstSeen.Equal(start) || !pattern.LastSeen.Equal(end.Add(-time.Nanosecond)) { + t.Fatalf("event-time bounds lost precision: %+v", pattern) + } + } +} + +func TestDetectorRefreshesPatternsAtStalledAndZeroWatermarks(t *testing.T) { + for _, completed := range []bool{false, true} { + name := "zero" + if completed { + name = "stalled" + } + t.Run(name, func(t *testing.T) { + db, err := sql.Open("duckdb", "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + if err := query.CreateCacheTables(db); err != nil { + t.Fatal(err) + } + createDetectorLogs(t, db) + end := time.Now().UTC().Truncate(time.Minute).Add(-time.Hour) + if completed { + seedDetectorRecoveryRollup(t, db, end) + } + d := NewDetector(&query.Duck{DB: db}, DefaultDetectorConfig()) + d.runCheck(t.Context()) + previous := d.LatestSnapshot() + if previous == nil || len(previous.Patterns) != 0 { + t.Fatalf("initial snapshot: %+v", previous) + } + at := time.Now().UTC().Add(-time.Minute) + if _, err := db.Exec(`INSERT INTO logs SELECT ?::TIMESTAMP_NS::TIMESTAMPTZ_NS, ?, 'failure', 'failure', 'ERROR', 'api', 'default' FROM range(10)`, at, at.UnixNano()); err != nil { + t.Fatal(err) + } + // Changing rollup data must not change the cached anomaly evaluation. + if _, err := db.Exec(`UPDATE service_rollup SET error_rate=0,p95_ms=0,spans=1`); err != nil { + t.Fatal(err) + } + d.runCheck(t.Context()) + got := d.LatestSnapshot() + if got == previous || len(got.Patterns) != 1 || got.Patterns[0].Count != 10 || got.HealthScore != previous.HealthScore-5 { + t.Fatalf("live logs did not refresh: %+v", got) + } + if len(got.Anomalies) != len(previous.Anomalies) { + t.Fatal("stalled anomaly evaluation changed") + } + for i, a := range got.Anomalies { + if a != previous.Anomalies[i] || !a.WindowEnd.Equal(end) { + t.Fatalf("anomaly interval or values changed: %+v", a) + } + } + // Move the live window beyond the logs without changing the watermark. + d.config.LookbackWindow = 30 * time.Second + d.runCheck(t.Context()) + if expired := d.LatestSnapshot(); len(expired.Patterns) != 0 || expired.HealthScore != previous.HealthScore { + t.Fatalf("expired patterns retained: %+v", expired) + } + }) + } +} diff --git a/internal/intelligence/rollup_detector_test.go b/internal/intelligence/rollup_detector_test.go index febfb5c2..8bed02d8 100644 --- a/internal/intelligence/rollup_detector_test.go +++ b/internal/intelligence/rollup_detector_test.go @@ -82,6 +82,7 @@ func TestDetectorsStopBeforeTheServiceWatermarkMinute(t *testing.T) { if err := query.CreateCacheTables(db); err != nil { t.Fatal(err) } + createDetectorLogs(t, db) base := time.Date(2026, 10, 1, 12, 0, 0, 0, time.UTC) for i, rate := range []float64{0, .5, 1} { if _, err := db.Exec(`INSERT INTO service_rollup(namespace,bucket,service,spans,served_spans,p95_ms,error_rate) VALUES ('prod',?,'api',100,100,10,?)`, base.Add(time.Duration(i)*5*time.Minute), rate); err != nil { @@ -111,7 +112,7 @@ func TestDetectorsStopBeforeTheServiceWatermarkMinute(t *testing.T) { } } -func TestDetectorPersistsEvaluatedWindowAndSkipsStalledMinute(t *testing.T) { +func TestDetectorPersistsEvaluatedWindowWhilePatternsRefresh(t *testing.T) { cfg := config.Config{DataDir: t.TempDir(), DuckDBMemory: "256MB", DuckDBThreads: 2, DuckDBMaxConns: 2} repo, err := telemetrystore.Open(cfg.TelemetryDir()) if err != nil { @@ -165,26 +166,30 @@ func TestDetectorPersistsEvaluatedWindowAndSkipsStalledMinute(t *testing.T) { t.Fatalf("historical interval: %+v", persisted) } // A recheck would see different values; an unchanged completed minute must - // retain the exact prior snapshot and enqueue no annotation write. + // retain the anomaly evaluation while rebuilding the live snapshot. if _, err := duck.DB.Exec(`UPDATE service_rollup SET error_rate=.9 WHERE bucket >= ?`, start); err != nil { t.Fatal(err) } for range 2 { detector.runCheck(t.Context()) } - if detector.LatestSnapshot() != previous || len(detector.annotationBatches) != 0 { - t.Fatal("stalled watermark re-evaluated or queued annotations") + if latest := detector.LatestSnapshot(); latest == previous || latest.Anomalies[0] != a { + t.Fatal("stalled watermark changed the anomaly evaluation") + } + refreshed := <-detector.annotationBatches + if len(refreshed.findings) != 1 || !refreshed.findings[0].From.Equal(start) || !refreshed.findings[0].To.Equal(end) { + t.Fatalf("live refresh changed persisted interval: %+v", refreshed) } after := read() if len(after) != 1 || !after[0].From.Equal(persisted[0].From) || !after[0].To.Equal(persisted[0].To) { t.Fatalf("stalled interval changed: %+v", after) } - // Crossing only seconds in the same minute still skips evaluation. + // Crossing only seconds in the same minute still skips anomaly evaluation. if _, err := duck.DB.Exec(`UPDATE rollup_state SET last_ingested_unix_nano=? WHERE cache_key='service_rollup_v3'`, end.Add(59*time.Second).UnixNano()); err != nil { t.Fatal(err) } detector.runCheck(t.Context()) - if detector.LatestSnapshot() != previous || len(detector.annotationBatches) != 0 { + if detector.LatestSnapshot().Anomalies[0] != a { t.Fatal("incomplete minute triggered evaluation") } if _, err := duck.DB.Exec(`UPDATE rollup_state SET last_ingested_unix_nano=? WHERE cache_key='service_rollup_v3'`, end.Add(time.Minute).UnixNano()); err != nil { diff --git a/internal/telemetry/store/admission_test.go b/internal/telemetry/store/admission_test.go index c63613a2..ea49ece3 100644 --- a/internal/telemetry/store/admission_test.go +++ b/internal/telemetry/store/admission_test.go @@ -17,8 +17,8 @@ import ( // still an out-of-memory kill however cheap each batch became. // // Submit is where that is countable -- both transports funnel through it, and -// it already blocks until durable acknowledgement, so charging it on entry and -// releasing it on return measures exactly the bytes the process is holding. +// it transfers its reservation to the queue until the writer finishes with the +// payload, even when the caller stops waiting for durable acknowledgement. func TestSubmitRefusesWorkBeyondTheInFlightBudget(t *testing.T) { w := NewWriter(nil, 1000) w.inFlightBudget = 4096 diff --git a/internal/telemetry/store/writer.go b/internal/telemetry/store/writer.go index 296809fd..d3be2cfc 100644 --- a/internal/telemetry/store/writer.go +++ b/internal/telemetry/store/writer.go @@ -51,6 +51,8 @@ type Writer struct { groupWindow time.Duration shutdownGrace time.Duration done chan struct{} + stopping chan struct{} + submissionMu sync.RWMutex submissions chan submission } @@ -66,6 +68,7 @@ type submission struct { type commitJob struct { batches []Batch acks []chan error + bytes int } // ErrIngestOverBudget is returned when accepting a request would take the @@ -109,7 +112,7 @@ func (w *Writer) release(bytes int) { func NewWriter(repository *Repository, batchSize int) *Writer { return &Writer{ repository: repository, batchSize: batchSize, groupWindow: groupAdmissionWindow, - done: make(chan struct{}), submissions: make(chan submission, submissionQueueDepth), + done: make(chan struct{}), stopping: make(chan struct{}), submissions: make(chan submission, submissionQueueDepth), } } @@ -127,22 +130,16 @@ func (w *Writer) Submit(ctx context.Context, batch Batch) error { if batchRows(batch) == 0 { return nil } - // Charged here rather than at the transport because both transports funnel - // through this call, and it already blocks until durable acknowledgement -- - // so the ledger measures exactly the bytes the process is holding. + // The caller owns the reservation until enqueue succeeds. The queue and + // then the commit job retain it even if the caller stops waiting for its ack. bytes := batchBytes(batch) if err := w.reserve(bytes); err != nil { return err } - defer w.release(bytes) request := submission{batch: batch, bytes: bytes, ack: make(chan error, 1)} - select { - case w.submissions <- request: - metrics.UpdateQueueDepth("batch", len(w.submissions)) - case <-w.done: - return errors.New("telemetry writer is stopped") - case <-ctx.Done(): - return ctx.Err() + if err := w.enqueueSubmission(ctx, request); err != nil { + w.release(bytes) + return err } select { case err := <-request.ack: @@ -154,9 +151,45 @@ func (w *Writer) Submit(ctx context.Context, batch Batch) error { } } +func (w *Writer) enqueueSubmission(ctx context.Context, request submission) error { + w.submissionMu.RLock() + defer w.submissionMu.RUnlock() + // Once shutdown starts, never enqueue into an already drained queue. + select { + case <-w.stopping: + return errors.New("telemetry writer is stopped") + default: + } + select { + case w.submissions <- request: + metrics.UpdateQueueDepth("batch", len(w.submissions)) + return nil + case <-w.stopping: + return errors.New("telemetry writer is stopped") + case <-ctx.Done(): + return ctx.Err() + } +} + func (w *Writer) Run(ctx context.Context) error { - defer close(w.done) - defer metrics.UpdateQueueDepth("batch", 0) + defer func() { + close(w.stopping) + // Wake blocked enqueues before excluding producers and draining. A send + // racing the stop can succeed, but must finish before this final drain. + w.submissionMu.Lock() + defer w.submissionMu.Unlock() + for { + select { + case request := <-w.submissions: + w.release(request.bytes) + request.ack <- errors.New("telemetry writer stopped before durable acknowledgement") + default: + metrics.UpdateQueueDepth("batch", 0) + close(w.done) + return + } + } + }() w.compaction, _ = w.repository.(compactionPressure) jobs := make(chan commitJob, commitQueueDepth) workerCtx, cancelWorkers := context.WithCancel(context.Background()) @@ -214,9 +247,17 @@ func (w *Writer) Run(ctx context.Context) error { } } -func (w *Writer) enqueueSubmissions(ctx context.Context, request submission, out chan<- commitJob) error { +func (w *Writer) enqueueSubmissions(ctx context.Context, request submission, out chan<- commitJob) (err error) { requests := make([]submission, 1, maxAdmissionRequests) requests[0] = request + defer func() { + if err != nil { + for _, pending := range requests { + w.release(pending.bytes) + pending.ack <- err + } + } + }() rows := batchRows(request.batch) bytes := request.bytes limit := w.batchLimit() @@ -281,7 +322,7 @@ drained: direct := requests[0] requests = requests[1:] direct.batch.ID = uuid.NewString() - if err := enqueueJob(ctx, out, commitJob{batches: []Batch{direct.batch}, acks: []chan error{direct.ack}}); err != nil { + if err := w.enqueueJob(ctx, out, commitJob{batches: []Batch{direct.batch}, acks: []chan error{direct.ack}, bytes: direct.bytes}); err != nil { return err } continue @@ -314,47 +355,47 @@ drained: for i := range group { acks[i] = group[i].ack } - if err := enqueueJob(ctx, out, commitJob{batches: []Batch{batch}, acks: acks}); err != nil { + if err := w.enqueueJob(ctx, out, commitJob{batches: []Batch{batch}, acks: acks, bytes: groupBytes}); err != nil { return err } } return nil } -func enqueueJob(ctx context.Context, out chan<- commitJob, job commitJob) error { +func (w *Writer) enqueueJob(ctx context.Context, out chan<- commitJob, job commitJob) error { select { case out <- job: return nil case <-ctx.Done(): + w.finishJob(job, ctx.Err()) return ctx.Err() } } func (w *Writer) commitWorker(ctx context.Context, jobs <-chan commitJob) { - for { - select { - case <-ctx.Done(): - return - case job, ok := <-jobs: - if !ok { - return - } - if err := w.commitJob(ctx, job); err != nil { - for _, ack := range job.acks { - ack <- err - } - slog.Error("telemetry batch commit exhausted retries", "error", err) - continue - } - for _, batch := range job.batches { - metrics.RecordIngest("spans", len(batch.Spans)) - metrics.RecordIngest("logs", len(batch.Logs)) - metrics.RecordIngest("metrics", len(batch.Metrics)) - } - for _, ack := range job.acks { - ack <- nil - } + for job := range jobs { + if err := ctx.Err(); err != nil { + w.finishJob(job, err) + continue + } + if err := w.commitJob(ctx, job); err != nil { + w.finishJob(job, err) + slog.Error("telemetry batch commit exhausted retries", "error", err) + continue + } + for _, batch := range job.batches { + metrics.RecordIngest("spans", len(batch.Spans)) + metrics.RecordIngest("logs", len(batch.Logs)) + metrics.RecordIngest("metrics", len(batch.Metrics)) } + w.finishJob(job, nil) + } +} + +func (w *Writer) finishJob(job commitJob, err error) { + w.release(job.bytes) + for _, ack := range job.acks { + ack <- err } } diff --git a/internal/telemetry/store/writer_accounting_test.go b/internal/telemetry/store/writer_accounting_test.go new file mode 100644 index 00000000..075fac25 --- /dev/null +++ b/internal/telemetry/store/writer_accounting_test.go @@ -0,0 +1,243 @@ +package store + +import ( + "context" + "errors" + "runtime" + "strings" + "testing" + "testing/synctest" + "time" + + "github.com/labstack/fanout/internal/telemetry" +) + +func accountingBatch() Batch { + return Batch{Logs: []telemetry.Log{{Body: strings.Repeat("x", 4096)}}} +} + +func TestWriterRetainsQueuedReservationAfterCallerTimeout(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + w := testWriter(&recordingCommitter{}, 1) + w.SetInFlightBudget(8192) + batch := accountingBatch() + for range 8 { + ctx, cancel := context.WithTimeout(t.Context(), time.Millisecond) + err := w.Submit(ctx, batch) + cancel() + if err != context.DeadlineExceeded && !errors.Is(err, ErrIngestOverBudget) { + t.Fatalf("submit: %v", err) + } + } + if got := w.inFlightBytes.Load(); got != int64(len(w.submissions)*batchBytes(batch)) || got > 8192 || got == 0 { + t.Fatalf("retained queue=%d bytes=%d, budget=8192", len(w.submissions), got) + } + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := w.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + if got := w.inFlightBytes.Load(); got != 0 || len(w.submissions) != 0 { + t.Fatalf("shutdown retained queue=%d bytes=%d", len(w.submissions), got) + } + }) +} + +func TestWriterRetainsCommittingReservationAfterCallerCancellation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + c := &blockingCommitter{entered: make(chan struct{}), release: make(chan struct{})} + w := testWriter(c, 1) + w.SetInFlightBudget(1) // One oversized request is still admitted. + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + defer func() { + select { + case <-c.release: + default: + close(c.release) + } + cancel() + if err := <-done; err != nil { + t.Error(err) + } + }() + submitCtx, stop := context.WithCancel(t.Context()) + result := make(chan error, 1) + batch := accountingBatch() + go func() { result <- w.Submit(submitCtx, batch) }() + <-c.entered + stop() + if err := <-result; err != context.Canceled { + t.Fatalf("submit=%v", err) + } + if got := w.inFlightBytes.Load(); got != int64(batchBytes(batch)) { + t.Fatalf("commit reservation released early: %d", got) + } + if err := w.Submit(t.Context(), batch); !errors.Is(err, ErrIngestOverBudget) { + t.Fatalf("retry bypassed retained commit: %v", err) + } + close(c.release) + synctest.Wait() + if got := w.inFlightBytes.Load(); got != 0 { + t.Fatalf("completed commit retained %d bytes", got) + } + }) +} + +func TestWriterReleasesUnenqueuedAndDiscardedJobsExactlyOnce(t *testing.T) { + for _, mode := range []string{"admission", "direct", "grouped"} { + t.Run(mode, func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + w := testWriter(&recordingCommitter{}, 1) + w.SetInFlightBudget(1 << 20) + w.groupWindow = time.Hour + if mode == "admission" { + w.batchSize = 100 + } + if mode == "grouped" { + w.batchSize = 3 + } + requests := make([]submission, 4) + for i := range requests { + batch := accountingBatch() + requests[i] = submission{batch: batch, bytes: batchBytes(batch), ack: make(chan error, 1)} + if err := w.reserve(requests[i].bytes); err != nil { + t.Fatal(err) + } + if i > 0 { + w.submissions <- requests[i] + } + } + capacity := 0 + if mode == "direct" { + capacity = 1 // Transfer one job, then fail the next enqueue. + } + out := make(chan commitJob, capacity) + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- w.enqueueSubmissions(ctx, requests[0], out) }() + synctest.Wait() + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("enqueue=%v", err) + } + want := int64(len(out) * requests[0].bytes) + if got := w.inFlightBytes.Load(); got != want { + t.Fatalf("enqueue error retained %d bytes, want queued job bytes %d", got, want) + } + close(out) + // Canceled workers must discard and release jobs already queued. + w.commitWorker(ctx, out) + if got := w.inFlightBytes.Load(); got != 0 { + t.Fatalf("discard retained %d bytes", got) + } + for _, request := range requests { + select { + case err := <-request.ack: + if err == nil { + t.Fatal("discard acknowledged as durable") + } + default: + t.Fatal("discarded request was not acknowledged") + } + } + }) + }) + } +} + +func TestWriterReleasesReservationOnCommitFailure(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + w := testWriter(&recordingCommitter{failures: commitRetryLimit}, 1) + w.SetInFlightBudget(8192) + w.retryDelay = func(int) time.Duration { return 0 } + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + if err := w.Submit(t.Context(), accountingBatch()); err == nil { + t.Fatal("failed commit acknowledged as durable") + } + if got := w.inFlightBytes.Load(); got != 0 { + t.Fatalf("failed commit retained %d bytes", got) + } + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + }) +} + +func TestWriterReleasesReservationsDuringConcurrentShutdown(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + w := testWriter(&recordingCommitter{}, 1) + w.SetInFlightBudget(1 << 20) + w.submissions = make(chan submission, 1) + results := make(chan error, 16) + for range cap(results) { + go func() { results <- w.Submit(t.Context(), accountingBatch()) }() + } + synctest.Wait() // One queued request and fifteen blocked enqueues. + ctx, cancel := context.WithCancel(t.Context()) + cancel() + if err := w.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + t.Fatal(err) + } + for range cap(results) { + <-results + } + if err := w.Submit(t.Context(), accountingBatch()); err == nil { + t.Fatal("stopped writer accepted work") + } + if got := w.inFlightBytes.Load(); got != 0 || len(w.submissions) != 0 { + t.Fatalf("shutdown retained queue=%d bytes=%d", len(w.submissions), got) + } + }) +} + +type shutdownCommitter struct { + entered chan struct{} +} + +func (c *shutdownCommitter) Commit(ctx context.Context, _ Batch) error { + c.entered <- struct{}{} + <-ctx.Done() + return ctx.Err() +} + +func TestWriterReleasesActiveAndQueuedJobsAfterShutdownGrace(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + workers := min(maxCommitWorkers, max(1, runtime.GOMAXPROCS(0))) + c := &shutdownCommitter{entered: make(chan struct{}, 2*workers)} + w := testWriter(c, 1) + w.shutdownGrace = time.Millisecond + w.SetInFlightBudget(1 << 20) + results := make(chan error, 2*workers) + for range cap(results) { + go func() { results <- w.Submit(t.Context(), accountingBatch()) }() + } + synctest.Wait() + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error, 1) + go func() { done <- w.Run(ctx) }() + for range workers { + <-c.entered + } + synctest.Wait() // All workers committing, the remaining jobs queued. + if got := w.inFlightBytes.Load(); got != int64(cap(results)*batchBytes(accountingBatch())) { + t.Fatalf("active and queued bytes=%d", got) + } + cancel() + if err := <-done; err != nil { + t.Fatal(err) + } + for range cap(results) { + if err := <-results; err == nil { + t.Fatal("shutdown discard acknowledged as durable") + } + } + if got := w.inFlightBytes.Load(); got != 0 { + t.Fatalf("shutdown grace retained %d bytes", got) + } + }) +} diff --git a/internal/telemetry/store/writer_test.go b/internal/telemetry/store/writer_test.go index 54bede99..e5d268bf 100644 --- a/internal/telemetry/store/writer_test.go +++ b/internal/telemetry/store/writer_test.go @@ -34,7 +34,7 @@ func (c *recordingCommitter) Commit(_ context.Context, batch Batch) error { func testWriter(committer batchCommitter, batchSize int) *Writer { return &Writer{ - repository: committer, batchSize: batchSize, groupWindow: groupAdmissionWindow, done: make(chan struct{}), + repository: committer, batchSize: batchSize, groupWindow: groupAdmissionWindow, done: make(chan struct{}), stopping: make(chan struct{}), submissions: make(chan submission, submissionQueueDepth), } } @@ -84,6 +84,7 @@ func TestWriterAcknowledgesOnlyAfterAtomicCommit(t *testing.T) { func TestWriterGroupsQueuedSubmissions(t *testing.T) { committer := &recordingCommitter{} w := testWriter(committer, maxGroupBatchRows) + w.SetInFlightBudget(8192) results := make(chan error, 4) for i := range 4 { go func() { @@ -99,6 +100,9 @@ func TestWriterGroupsQueuedSubmissions(t *testing.T) { t.Fatal(err) } } + if got := w.inFlightBytes.Load(); got != 0 { + t.Fatalf("grouped commit retained %d bytes", got) + } cancel() if err := <-runDone; err != nil { t.Fatal(err)