From 6b79aa99721133f531cb57695b12027ff8f29dd8 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Tue, 11 Aug 2026 09:58:29 +0100 Subject: [PATCH] Bound synchronous metrics writes so a slow disk cannot stall monitoring WriteBatchSync waited unboundedly for the ingestion worker's commit. The monitoring pipeline calls it inline from state broadcast, agent ingest, and poll publish, so a metrics disk slow enough to back up the write queue froze the monitor after its first cycle: polls stopped being scheduled, PollStatus.LastSuccess never advanced, and healthy API sources degraded to stale/agent-only while SQLite ground through retention maintenance (107s cleanup, multi-second commits on the reporter's instance). enqueueAndWait now shares a single 2s budget across enqueue and commit. A queue that cannot accept the batch within the budget drops it with a warning, matching enqueueWrite's saturation behavior. A batch that enqueues but has not committed stays queued and is not lost; the caller moves on and a rate-limited warning records the backlog. Healthy disks keep read-your-writes semantics. Refs #1437 Contract-Neutral: behavioral fix: bound metrics store sync write wait (#1437), no public contract delta --- pkg/metrics/store.go | 59 +++++++++++++- pkg/metrics/store_write_backpressure_test.go | 86 ++++++++++++++++++++ 2 files changed, 143 insertions(+), 2 deletions(-) create mode 100644 pkg/metrics/store_write_backpressure_test.go diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index 7e544e549..6f3dc713f 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -205,6 +205,7 @@ type Store struct { maintenanceDoneCh chan struct{} stopOnce sync.Once stopping atomic.Bool + syncWaitWarnNano atomic.Int64 identityMigrationPending atomic.Bool commercialRetentionSeconds atomic.Int64 commercialPurgeEligibleAt atomic.Int64 @@ -916,13 +917,67 @@ func (s *Store) enqueueWrite(req writeRequest) { } } +// syncWriteWaitTimeout bounds how long a synchronous batch write blocks on the +// ingestion worker. The monitoring pipeline calls WriteBatchSync inline (state +// broadcast, agent ingest, poll publish), so an unbounded wait lets a slow +// metrics disk starve polling entirely: on #1437's instance SQLite commits ran +// for seconds to minutes and the monitor froze after its first cycle while +// history writes queued behind retention maintenance. Within the budget the +// call keeps read-your-writes; past it the caller moves on and history lands +// whenever the worker catches up. +const syncWriteWaitTimeout = 2 * time.Second + +// enqueueAndWait hands the batch to the ingestion worker and waits for its +// commit, but never longer than syncWriteWaitTimeout in total. If the queue +// cannot even accept the batch within the budget the batch is dropped, matching +// enqueueWrite's saturation behavior. If only the commit is outstanding the +// batch stays queued and is not lost. func (s *Store) enqueueAndWait(req writeRequest) { if req.done == nil { req.done = make(chan struct{}) } - s.writeCh <- req - <-req.done + timer := time.NewTimer(syncWriteWaitTimeout) + defer timer.Stop() + + select { + case s.writeCh <- req: + case <-s.stopCh: + return + case <-timer.C: + log.Warn(). + Str("component", "metrics_store"). + Str("action", "drop_sync_write_batch"). + Int("batch_size", len(req.metrics)). + Int("write_queue_depth", len(s.writeCh)). + Int("write_queue_capacity", cap(s.writeCh)). + Msg("Metrics write queue saturated, dropping synchronous batch to keep monitoring live") + return + } + + select { + case <-req.done: + case <-s.stopCh: + case <-timer.C: + s.warnSyncWriteBacklog(len(req.metrics)) + } +} + +// warnSyncWriteBacklog reports a lagging ingestion worker at most once per +// 30-second window. Unlike a dropped batch this is not data loss, so the +// per-call signal is redundant while the condition persists. +func (s *Store) warnSyncWriteBacklog(batchSize int) { + now := time.Now().UnixNano() + last := s.syncWaitWarnNano.Load() + if now-last < int64(30*time.Second) || !s.syncWaitWarnNano.CompareAndSwap(last, now) { + return + } + log.Warn(). + Str("component", "metrics_store"). + Str("action", "sync_write_backlogged"). + Int("batch_size", batchSize). + Int("write_queue_depth", len(s.writeCh)). + Msg("Metrics write worker lagging, batch left queued and monitoring continues") } func (s *Store) drainBuffer() []bufferedMetric { diff --git a/pkg/metrics/store_write_backpressure_test.go b/pkg/metrics/store_write_backpressure_test.go new file mode 100644 index 000000000..f387e6b4c --- /dev/null +++ b/pkg/metrics/store_write_backpressure_test.go @@ -0,0 +1,86 @@ +package metrics + +import ( + "testing" + "time" +) + +// Regression coverage for #1437: WriteBatchSync sits on the monitoring +// pipeline (state broadcast, agent ingest, poll publish). When the ingestion +// worker cannot keep up with the disk, the call must return within its wait +// budget instead of stalling the monitor until polling stops. + +// newUnservicedStore builds a bare store whose ingestion worker never runs, +// modelling a writer wedged behind slow SQLite maintenance. +func newUnservicedStore(queueCap int) *Store { + return &Store{ + writeCh: make(chan writeRequest, queueCap), + stopCh: make(chan struct{}), + } +} + +func backpressureProbeMetric() WriteMetric { + return WriteMetric{ + ResourceType: "vm", + ResourceID: "backpressure-probe", + MetricType: "cpu", + Value: 1, + Timestamp: time.Unix(1_700_000_000, 0), + Tier: TierRaw, + } +} + +func requireReturnsWithinWaitBudget(t *testing.T, s *Store, label string) { + t.Helper() + + done := make(chan struct{}) + go func() { + s.WriteBatchSync([]WriteMetric{backpressureProbeMetric()}) + close(done) + }() + + select { + case <-done: + case <-time.After(syncWriteWaitTimeout + 3*time.Second): + t.Fatalf("WriteBatchSync stalled past its wait budget (%s)", label) + } +} + +func TestWriteBatchSyncReturnsWhenQueueSaturated(t *testing.T) { + s := newUnservicedStore(1) + s.writeCh <- writeRequest{metrics: []bufferedMetric{{}}} + + requireReturnsWithinWaitBudget(t, s, "saturated queue") + + if got := len(s.writeCh); got != 1 { + t.Fatalf("saturated queue depth = %d, want the pre-existing batch only", got) + } +} + +func TestWriteBatchSyncReturnsWhenCommitLags(t *testing.T) { + s := newUnservicedStore(4) + + requireReturnsWithinWaitBudget(t, s, "lagging commit") + + if got := len(s.writeCh); got != 1 { + t.Fatalf("queued batches = %d, want 1 (batch must stay queued, not be dropped)", got) + } +} + +func TestWriteBatchSyncReturnsOnClosedStore(t *testing.T) { + s := newUnservicedStore(1) + s.writeCh <- writeRequest{metrics: []bufferedMetric{{}}} + close(s.stopCh) + + done := make(chan struct{}) + go func() { + s.WriteBatchSync([]WriteMetric{backpressureProbeMetric()}) + close(done) + }() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("WriteBatchSync did not observe store shutdown") + } +}