From 29d96aec3373fe9f6ffae130f9306836532bff16 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 19 Apr 2026 17:11:14 +0100 Subject: [PATCH] Reduce metrics store transaction churn Refs #1124 --- .../subsystems/performance-and-scalability.md | 10 ++++ pkg/metrics/store.go | 48 +++++++++++++++++-- pkg/metrics/store_additional_test.go | 31 ++++++++++++ 3 files changed, 84 insertions(+), 5 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 501c4b02f3..d9eca073f9 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -401,6 +401,16 @@ onto the shared backend `resourceType=k8s` transport, but it must preserve the canonical frontend target types for clusters, nodes, pods, and deployments so the workloads and drawer hot paths do not silently drop deployment history or split Kubernetes charts across incompatible cache keys. +That same protected metrics-store boundary also owns write-path churn on local +instances. `pkg/metrics/store.go` must prefer fewer, larger SQLite write +transactions over tiny frequent commits: the default write buffer should stay +large enough to absorb one poll cycle on modest multi-node installs, queued +flush batches should coalesce before commit when the worker is already behind, +and WAL auto-checkpointing should avoid tiny segment rewrite loops that keep +rewriting `metrics.db` on SSD-backed systems. Any future change that reduces +that batching headroom or makes WAL checkpoints more aggressive again must +re-prove the metrics-store hot path with the owned store tests rather than +assuming the earlier vacuum fixes are sufficient. contract instead of inventing an infrastructure-local summary filter branch. For shared line charts on that hot path, the shared sparkline primitive may isolate the selected series inside the existing render budget, but that diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index dab70f9ffe..2fe57bd2d8 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -58,8 +58,11 @@ func DefaultConfig(dataDir string) StoreConfig { } return StoreConfig{ - DBPath: dbPath, - WriteBufferSize: 100, + DBPath: dbPath, + // Large installs can enqueue hundreds of metric points per poll cycle. + // A larger buffer keeps those writes inside a single SQLite transaction + // more often, which materially reduces WAL churn on SSD-backed setups. + WriteBufferSize: 500, FlushInterval: 5 * time.Second, RetentionRaw: 2 * time.Hour, RetentionMinute: 24 * time.Hour, @@ -154,7 +157,9 @@ func NewStore(config StoreConfig) (*Store, error) { "journal_mode(WAL)", "synchronous(NORMAL)", "auto_vacuum(INCREMENTAL)", - "wal_autocheckpoint(500)", + // Checkpoint less aggressively so high-cardinality installs don't + // keep rewriting tiny WAL segments back into the main DB file. + "wal_autocheckpoint(4000)", }, }.Encode() rawDB, err := sql.Open("sqlite", dsn) @@ -674,6 +679,32 @@ func (s *Store) writeBatch(metrics []bufferedMetric) { log.Debug().Int("count", len(metrics)).Msg("Wrote metrics batch") } +// coalesceQueuedBatches drains any already-queued write batches so the worker +// can commit them in a single SQLite transaction. This reduces WAL write +// amplification when the in-memory buffer flushes multiple times during one +// poll cycle. +func (s *Store) coalesceQueuedBatches(initial []bufferedMetric) []bufferedMetric { + if len(initial) == 0 { + return nil + } + + combined := initial + for { + select { + case next, ok := <-s.writeCh: + if !ok { + return combined + } + if len(next) == 0 { + continue + } + combined = append(combined, next...) + default: + return combined + } + } +} + // Query retrieves metrics for a resource within a time range, with optional downsampling func (s *Store) Query(resourceType, resourceID, metricType string, start, end time.Time, stepSecs int64) ([]MetricPoint, error) { resourceType = normalizeMetricResourceType(resourceType) @@ -1279,13 +1310,20 @@ func (s *Store) backgroundWorker() { s.Flush() // Process remaining writes close(s.writeCh) + var remaining []bufferedMetric for batch := range s.writeCh { - s.writeBatch(batch) + if len(batch) == 0 { + continue + } + remaining = append(remaining, batch...) + } + if len(remaining) > 0 { + s.writeBatch(remaining) } return case batch := <-s.writeCh: - s.writeBatch(batch) + s.writeBatch(s.coalesceQueuedBatches(batch)) case maintenance := <-s.maintenanceCh: if maintenance.run != nil { diff --git a/pkg/metrics/store_additional_test.go b/pkg/metrics/store_additional_test.go index fa16c5207a..c70bd929f5 100644 --- a/pkg/metrics/store_additional_test.go +++ b/pkg/metrics/store_additional_test.go @@ -12,6 +12,37 @@ import ( "github.com/rs/zerolog/log" ) +func TestDefaultConfigUsesLargerWriteBuffer(t *testing.T) { + cfg := DefaultConfig("/tmp/pulse") + if cfg.WriteBufferSize != 500 { + t.Fatalf("WriteBufferSize = %d, want 500", cfg.WriteBufferSize) + } +} + +func TestStoreCoalesceQueuedBatches(t *testing.T) { + store := &Store{ + writeCh: make(chan []bufferedMetric, 4), + } + + initial := []bufferedMetric{ + {resourceType: "vm", resourceID: "vm-1", metricType: "cpu", value: 10}, + } + store.writeCh <- []bufferedMetric{ + {resourceType: "vm", resourceID: "vm-1", metricType: "memory", value: 20}, + } + store.writeCh <- []bufferedMetric{ + {resourceType: "vm", resourceID: "vm-2", metricType: "cpu", value: 30}, + } + + combined := store.coalesceQueuedBatches(initial) + if len(combined) != 3 { + t.Fatalf("expected 3 combined metrics, got %d", len(combined)) + } + if len(store.writeCh) != 0 { + t.Fatalf("expected queued batches to be drained, got %d remaining", len(store.writeCh)) + } +} + func TestStoreWriteBatchSync(t *testing.T) { dir := t.TempDir() store, err := NewStore(DefaultConfig(dir))