From e2635c38101ffa26bfd51b6d657c0ea2907b1b7c Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sat, 19 Sep 2026 20:18:01 +0100 Subject: [PATCH] fix(metrics): survive startup write lock and bound allocations Address the remaining verified findings from the pkg/metrics/store.go review (#2079 finding 5 and the lower-impact nits): - writeBatch now retries the whole transaction on a retryable lock error. BEGIN is deferred, so contention is hit at the first INSERT or COMMIT; the old loop retried only Begin() and skipped rows. While the one-time startup maintenance (deferred identity-index rebuild and auto-vacuum VACUUM) is active, an extended retry budget keeps upgrade-time history instead of dropping it after the steady-state budget. - migrateLegacyHostResourceType probes with an indexed EXISTS before opening a write transaction, so an already-migrated store does not dirty the WAL on every boot. - QueryAll caps the per-series MetricPoint preallocation so a fine requested step over a long range cannot reserve hundreds of thousands of slots. Regressions: TestStoreStartupMaintenanceSignalsWriteRetryBudget, TestStoreWriteBatchExtendsRetryDuringStartupMaintenance, TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation. The performance-and-scalability contract records the invariants. Change-source: pulse-maintainer --- .../subsystems/performance-and-scalability.md | 33 ++++ pkg/metrics/store.go | 154 +++++++++++++----- pkg/metrics/store_additional_test.go | 133 +++++++++++++++ 3 files changed, 277 insertions(+), 43 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index a50d82d4ed..b50592f6f5 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -3218,3 +3218,36 @@ and repeat the full-file VACUUM on every restart. The migration now pins a connection, verifies the persisted mode, and leaves the conversion for a later restart if verification fails. `pkg/metrics/store_additional_test.go` pins the persisted mode in `TestStoreAutoVacuumPersistsAcrossRestart`. + +### Metrics-store writes survive the one-time startup write-lock hold + +The deferred identity-index rebuild and the auto-vacuum VACUUM can hold the +SQLite write lock for minutes on a large legacy database. `BEGIN` is deferred, +so the lock is contended at the first `INSERT` or at `COMMIT`, not at `Begin`; +a retry loop around `Begin` alone skipped rows and committed an empty +transaction. `writeBatch` now retries the whole transaction on a retryable lock +error, and while startup maintenance is active it uses an extended budget +(`startupMaintenanceBeginAttempts`) instead of the steady-state +`writeBatchBeginAttempts`, so an upgrade does not silently discard history. The +`startupMaintenanceActive` signal is set for the duration of +`runStartupMaintenance` and cleared on completion; the `stopping` check still +bounds shutdown. `pkg/metrics/store_additional_test.go` pins the signal in +`TestStoreStartupMaintenanceSignalsWriteRetryBudget` and the extended retry in +`TestStoreWriteBatchExtendsRetryDuringStartupMaintenance`. + +### Metrics-store legacy host migration is probed before it writes + +`migrateLegacyHostResourceType` must not open a write transaction on every boot +when there are no legacy rows. It probes with an indexed +`SELECT EXISTS(... WHERE resource_type = 'host' LIMIT 1)` and returns without +writing when the table has no legacy rows, so an already-migrated store does +not dirty the WAL on startup. + +### Metrics-store QueryAll preallocation is capped + +`QueryAll` must not reserve a `MetricPoint` slice sized by the caller's +requested step when the selected tier returns far fewer rows. A 90-day range at +a 5-second step previously reserved over a million slots per series; the +preallocation is now capped at `maxQueryAllSeriesCapacity` and append still +grows the slice for genuinely dense series. `pkg/metrics/store_additional_test.go` +pins the cap in `TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation`. diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index cea4491c93..79e6af59a5 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -109,6 +109,28 @@ const ( defaultRollupInterval = 15 * time.Minute maxRollupChunkWindow = 5 * time.Minute maxRollupChunksPerRun = 128 + + // writeBatchBeginAttempts bounds how many times a metrics write batch + // retries a busy transaction before it is dropped. Five attempts is enough + // for the steady-state WAL writer. + writeBatchBeginAttempts = 5 + // startupMaintenanceBeginAttempts extends that budget while the one-time + // startup maintenance holds the SQLite write lock. The deferred + // identity-index rebuild and the auto-vacuum VACUUM can hold the lock for + // minutes on a large legacy database; the steady-state budget would drop + // every batch enqueued during an upgrade. Each attempt may itself wait + // busy_timeout (30s), so this covers roughly half an hour, matching + // migrateAutoVacuum's context deadline. + startupMaintenanceBeginAttempts = 70 + // maxWriteRetryBackoff caps the sleep between write attempts so a fast + // non-lock failure cannot spin. + maxWriteRetryBackoff = 2 * time.Second + + // maxQueryAllSeriesCapacity caps the per-series preallocation in QueryAll. + // A caller can request a fine step over a long range (e.g. 5s over 90 days) + // while the tier fallback actually yields far fewer rows; without a cap + // every series reserves hundreds of thousands of MetricPoint slots. + maxQueryAllSeriesCapacity = 4096 ) // DefaultConfig returns sensible defaults for metrics storage @@ -272,6 +294,7 @@ type Store struct { stopping atomic.Bool syncWaitWarnNano atomic.Int64 identityMigrationPending atomic.Bool + startupMaintenanceActive atomic.Bool commercialRetentionSeconds atomic.Int64 commercialPurgeEligibleAt atomic.Int64 @@ -724,6 +747,21 @@ func (s *Store) migrateAutoVacuum() { // canonical v6 `resource_type=agent`. This keeps reads/writes agent-only while // preserving historical data. func (s *Store) migrateLegacyHostResourceType() { + // Probe with the resource_type prefix of idx_metrics_query_all before + // opening a write transaction. The migration is a no-op on every store + // that has already run it, and an unconditional write transaction dirties + // the WAL on every boot for nothing. + var legacyHostExists bool + if err := s.db.QueryRow( + `SELECT EXISTS(SELECT 1 FROM metrics WHERE resource_type = 'host' LIMIT 1)`, + ).Scan(&legacyHostExists); err != nil { + log.Warn().Err(err).Msg("Failed to probe legacy host metrics rows") + return + } + if !legacyHostExists { + return + } + tx, err := s.db.Begin() if err != nil { log.Warn().Err(err).Msg("Failed to start legacy host->agent metrics migration") @@ -992,6 +1030,14 @@ func (s *Store) WaitForMaintenance(timeout time.Duration) error { func (s *Store) runStartupMaintenance() { start := time.Now() + + // Signal the ingestion worker that the write lock may be held for a long + // time by the deferred identity-index rebuild and the auto-vacuum VACUUM. + // Without this a large legacy database drops every metric batch enqueued + // during the one-time upgrade (#2079 finding 5). + s.startupMaintenanceActive.Store(true) + defer s.startupMaintenanceActive.Store(false) + if s.startupHook != nil { s.startupHook() } @@ -1196,26 +1242,49 @@ func (s *Store) writeBatch(metrics []bufferedMetric) { Msg("Coalesced duplicate metrics before write") } - var tx *pdb.InstrumentedTx - var err error - - // Retry on SQLITE_BUSY with exponential backoff - for i := 0; i < 5; i++ { - tx, err = s.db.Begin() - if err == nil { - break - } - if i < 4 && isRetryableWriteError(err) { - time.Sleep(time.Duration(100*(i+1)) * time.Millisecond) - continue - } - log.Error().Err(err). - Str("component", "metrics_store"). - Str("action", "begin_write_tx"). - Int("batch_size", len(metrics)). - Msg("Failed to begin metrics transaction") - return + // Retry the whole transaction on SQLITE_BUSY with exponential backoff. + // BEGIN is deferred, so the write lock is actually contended at the first + // INSERT or at COMMIT; retrying only Begin() would skip rows and commit an + // empty transaction. While the one-time startup maintenance holds the + // write lock, extend the budget rather than dropping durable history + // during an upgrade (#2079 finding 5). The stopping check keeps shutdown + // from waiting out the extended budget. + maxAttempts := writeBatchBeginAttempts + if s.startupMaintenanceActive.Load() { + maxAttempts = startupMaintenanceBeginAttempts } + for attempt := 0; attempt < maxAttempts; attempt++ { + retryable, err := s.writeBatchOnce(metrics) + if err == nil { + log.Debug().Int("input_count", inputCount).Int("count", len(metrics)).Msg("Wrote metrics batch") + return + } + if !retryable || attempt == maxAttempts-1 || s.stopping.Load() { + log.Error().Err(err). + Str("component", "metrics_store"). + Str("action", "write_batch"). + Int("batch_size", len(metrics)). + Msg("Failed to write metrics batch") + return + } + backoff := time.Duration(100*(attempt+1)) * time.Millisecond + if backoff > maxWriteRetryBackoff { + backoff = maxWriteRetryBackoff + } + time.Sleep(backoff) + } +} + +// writeBatchOnce commits one metrics transaction. It reports retryable=true +// when the failure was a transient lock so the caller can retry the whole +// batch instead of dropping it. A non-retryable per-row insert error is logged +// and skipped, preserving the batch's other rows. +func (s *Store) writeBatchOnce(metrics []bufferedMetric) (bool, error) { + tx, err := s.db.Begin() + if err != nil { + return isRetryableWriteError(err), fmt.Errorf("begin metrics transaction: %w", err) + } + defer func() { _ = tx.Rollback() }() stmt, err := tx.Prepare(` INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) @@ -1227,40 +1296,32 @@ func (s *Store) writeBatch(metrics []bufferedMetric) { max_value = COALESCE(excluded.max_value, max_value) `) if err != nil { - _ = tx.Rollback() - log.Error().Err(err). - Str("component", "metrics_store"). - Str("action", "prepare_write_stmt"). - Int("batch_size", len(metrics)). - Msg("Failed to prepare metrics insert") - return + return false, fmt.Errorf("prepare metrics insert: %w", err) } defer stmt.Close() for _, m := range metrics { _, err := stmt.Exec(m.resourceType, m.resourceID, m.metricType, m.value, m.timestamp.Unix(), string(m.tier)) - if err != nil { - log.Warn().Err(err). - Str("component", "metrics_store"). - Str("action", "insert_metric"). - Str("resource_type", m.resourceType). - Str("resource_id", m.resourceID). - Str("metric_type", m.metricType). - Str("tier", string(m.tier)). - Msg("Failed to insert metric") + if err == nil { + continue } + if isRetryableWriteError(err) { + return true, fmt.Errorf("insert metric: %w", err) + } + log.Warn().Err(err). + Str("component", "metrics_store"). + Str("action", "insert_metric"). + Str("resource_type", m.resourceType). + Str("resource_id", m.resourceID). + Str("metric_type", m.metricType). + Str("tier", string(m.tier)). + Msg("Failed to insert metric") } if err := tx.Commit(); err != nil { - log.Error().Err(err). - Str("component", "metrics_store"). - Str("action", "commit_write_tx"). - Int("batch_size", len(metrics)). - Msg("Failed to commit metrics batch") - return + return isRetryableWriteError(err), fmt.Errorf("commit metrics batch: %w", err) } - - log.Debug().Int("input_count", inputCount).Int("count", len(metrics)).Msg("Wrote metrics batch") + return false, nil } func coalesceMetricBatch(metrics []bufferedMetric) []bufferedMetric { @@ -1846,6 +1907,13 @@ func estimateQueryAllBatchSeriesCapacity(start, end time.Time, stepSecs int64) i if buckets < 1 { return 1 } + // The selected tier's actual resolution is usually far coarser than the + // requested step, so the row count is much smaller than the bucket count. + // Cap the preallocation; append still grows the slice for genuinely dense + // series. + if buckets > maxQueryAllSeriesCapacity { + return maxQueryAllSeriesCapacity + } return buckets } diff --git a/pkg/metrics/store_additional_test.go b/pkg/metrics/store_additional_test.go index 44b981d8ee..b98bc220b6 100644 --- a/pkg/metrics/store_additional_test.go +++ b/pkg/metrics/store_additional_test.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "syscall" "testing" "time" @@ -1719,3 +1720,135 @@ func TestStoreAutoVacuumPersistsAcrossRestart(t *testing.T) { } } } + +// TestStoreStartupMaintenanceSignalsWriteRetryBudget verifies the ingestion +// worker can tell when the one-time startup maintenance holds the write lock, +// and that the signal is cleared once maintenance completes. +func TestStoreStartupMaintenanceSignalsWriteRetryBudget(t *testing.T) { + previousHook := startupMaintenanceHook + defer func() { startupMaintenanceHook = previousHook }() + + // The automatic startup run may begin before NewStore returns, so the hook + // reads the store through an atomic pointer rather than a captured local. + var storePtr atomic.Pointer[Store] + var observedDuringHook atomic.Bool + startupMaintenanceHook = func() { + if s := storePtr.Load(); s != nil { + observedDuringHook.Store(s.startupMaintenanceActive.Load()) + } + } + + dir := t.TempDir() + cfg := DefaultConfig(dir) + cfg.DBPath = filepath.Join(dir, "startup-signal.db") + cfg.FlushInterval = time.Hour + + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("NewStore returned error: %v", err) + } + defer store.Close() + storePtr.Store(store) + + if err := store.WaitForMaintenance(10 * time.Second); err != nil { + t.Fatalf("maintenance did not complete: %v", err) + } + + // Re-run synchronously so the hook is guaranteed to observe the flag while + // it is active. + observedDuringHook.Store(false) + store.runStartupMaintenance() + + if !observedDuringHook.Load() { + t.Fatalf("startup maintenance did not signal an active write-retry budget") + } + if store.startupMaintenanceActive.Load() { + t.Fatalf("startup maintenance left the extended write-retry budget active") + } +} + +// TestStoreWriteBatchExtendsRetryDuringStartupMaintenance proves that a batch +// enqueued while the startup maintenance holds the write lock survives past +// the steady-state retry budget. Without the extension it is dropped after +// five fast attempts (#2079 finding 5). +func TestStoreWriteBatchExtendsRetryDuringStartupMaintenance(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "retry.db") + + holder, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(0)") + if err != nil { + t.Fatalf("open holder: %v", err) + } + defer holder.Close() + holder.SetMaxOpenConns(1) + schema := ` + CREATE TABLE metrics ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + resource_type TEXT NOT NULL, + resource_id TEXT NOT NULL, + metric_type TEXT NOT NULL, + value REAL NOT NULL, + min_value REAL, + max_value REAL, + timestamp INTEGER NOT NULL, + tier TEXT NOT NULL DEFAULT 'raw' + ); + CREATE UNIQUE INDEX idx_metrics_lookup + ON metrics(resource_type, resource_id, metric_type, tier, timestamp); + ` + if _, err := holder.Exec(schema); err != nil { + t.Fatalf("create schema: %v", err) + } + if _, err := holder.Exec("BEGIN IMMEDIATE"); err != nil { + t.Fatalf("begin immediate: %v", err) + } + + rawDB, err := sql.Open("sqlite", "file:"+path+"?_pragma=busy_timeout(0)") + if err != nil { + t.Fatalf("open writer: %v", err) + } + defer rawDB.Close() + store := &Store{db: pdb.Wrap(rawDB, "metrics")} + store.startupMaintenanceActive.Store(true) + + // Hold the write lock for longer than the steady-state budget + // (5 attempts, ~1s of backoff) but within the extended budget. + released := make(chan struct{}) + go func() { + time.Sleep(2 * time.Second) + _, _ = holder.Exec("ROLLBACK") + close(released) + }() + + store.writeBatch([]bufferedMetric{{ + resourceType: "vm", + resourceID: "vm-1", + metricType: "cpu", + value: 1, + timestamp: time.Now().UTC().Truncate(time.Second), + tier: TierRaw, + }}) + <-released + + var count int + if err := store.db.QueryRow("SELECT COUNT(*) FROM metrics").Scan(&count); err != nil { + t.Fatalf("count metrics: %v", err) + } + if count != 1 { + t.Fatalf("expected the batch to survive the startup lock window, got %d rows", count) + } +} + +// TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation verifies a fine +// requested step over a long range cannot reserve hundreds of thousands of +// MetricPoint slots per series when the tier fallback returns far fewer rows. +func TestEstimateQueryAllBatchSeriesCapacityCapsPreallocation(t *testing.T) { + start := time.Unix(0, 0) + end := start.Add(90 * 24 * time.Hour) + if got := estimateQueryAllBatchSeriesCapacity(start, end, 5); got != maxQueryAllSeriesCapacity { + t.Fatalf("90 days at 5s step: capacity = %d, want cap %d", got, maxQueryAllSeriesCapacity) + } + if got := estimateQueryAllBatchSeriesCapacity(start, start.Add(10*time.Minute), 5); got != 122 { + t.Fatalf("10 minutes at 5s step: capacity = %d, want 122", got) + } +}