diff --git a/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-metrics-write-amplification-2026-05-03.md b/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-metrics-write-amplification-2026-05-03.md index f4c4da69de..e8cfb06c99 100644 --- a/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-metrics-write-amplification-2026-05-03.md +++ b/docs/release-control/v6/internal/records/known-rc-issue-closure-for-ga-metrics-write-amplification-2026-05-03.md @@ -60,3 +60,49 @@ The metrics write-amplification path reported in `#1124` has a v6 runtime fix, operator-facing controls, documentation, and subsystem-contract coverage. The `known-rc-issue-closure-for-ga` gate remains satisfied for the current v6 release candidate. + +## 2026-07-24 Physical-Write Revalidation + +The earlier proof established rollup controls and bounded retention, but it did +not attribute physical writes by SQLite object or quantify WAL/checkpoint +amplification. Issue `#1124` therefore remained open after a v6.1.1 operator +reported continued writes. A deterministic 2,197-resource estate +(Docker containers, Proxmox guests, agents, Kubernetes pods, storage targets, +and physical disks) now persists 393,630 samples over 30 simulated 10-second +polls and reports logical payload, per-object `dbstat`, page-cache, WAL, +checkpoint, process-write, restart, integrity, and retention counters. + +The v6.1.1 and pre-fix `main` store implementations were byte-identical. Their +four-index schema wrote 279,063 WAL frames (1,149,739,592 bytes) for +21,214,200 logical payload bytes. The retained main database was 106,991,616 +bytes, of which the overlapping `idx_metrics_lookup` and +`idx_metrics_unique` trees each occupied 24,014,848 bytes. This identifies the +amplification source directly: each metric insert maintained the table, +`sqlite_sequence`, and four indexes before checkpointing the dirty pages back +to the main file. + +The revalidated schema makes the lookup index unique and orders its columns as +the metric identity, eliminating the second overlapping tree without changing +query order or durability settings. On the same workload it wrote 187,616 WAL +frames (772,977,952 bytes), a 32.8% frame/byte reduction, and retained +82,948,096 bytes. Linux process I/O accounting measured WAL-phase writes +falling from 1,155,710,976 to 774,930,432 bytes and explicit checkpoint writes +from 106,983,424 to 82,939,904 bytes (32.1% fewer total bytes). With the +production 4,000-page auto-checkpoint enabled, a traced 157,452-sample run +reduced total process write bytes from 392,445,952 to 257,208,320 and `fsync` +calls from 42 to 32. + +The migration keeps the legacy unique index authoritative until deferred +startup maintenance transactionally swaps the indexes. Tests cover legacy +duplicates, close/restart, a process exit between index drop and replacement, +backup/restore from both schema generations, writes racing the migration, and +reads remaining available on the four-connection WAL pool. Query-plan checks +still require indexed searches. Concurrent read/write p95 remained 0.88 ms +against a 5 ms SLO, and the 500-node/10-reader dashboard p95 remained 2.37 ms +against a 30 ms SLO. Retention removed all 393,630 expired rows and incremental +vacuum reduced the 82,948,096-byte database to 36,864 bytes with zero freelist +pages in 2.57 seconds. + +The durable store remains in WAL mode with `synchronous=NORMAL`; no +authoritative history moved to volatile memory. The fix is on `main` for the +next release and is not part of v6.1.1. diff --git a/docs/release-control/v6/internal/status.json b/docs/release-control/v6/internal/status.json index 128f81f1c6..6eadd284f4 100644 --- a/docs/release-control/v6/internal/status.json +++ b/docs/release-control/v6/internal/status.json @@ -9206,21 +9206,7 @@ ] } ], - "work_claims": [ - { - "id": "issue-1124-metrics-write-amplification-lane-l10", - "agent_id": "issue-1124-metrics-write-amplification", - "summary": "Measure and reduce metrics.db physical write amplification for issue #1124 without weakening durability or bounded runtime behavior.", - "target_id": "v6-product-lane-expansion", - "claimed_at": "2026-07-24T08:05:22Z", - "heartbeat_at": "2026-07-24T08:05:22Z", - "expires_at": "2026-07-24T16:05:22Z", - "work_item": { - "kind": "lane", - "id": "L10" - } - } - ], + "work_claims": [], "open_decisions": [], "source_of_truth_file": "docs/release-control/v6/internal/SOURCE_OF_TRUTH.md", "resolved_decisions": [ 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 d929221728..fbb84903a1 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -1201,6 +1201,23 @@ WAL checkpoints more aggressive again, reopens duplicate-write failures, removes the bounded rollup-window checkpointing, or removes the metrics DB path/cadence controls must re-prove the metrics-store hot path with the owned store tests rather than assuming the earlier vacuum fixes are sufficient. +The metrics identity and single-series lookup must also share one unique +`idx_metrics_lookup(resource_type, resource_id, metric_type, tier, timestamp)` +B-tree. Reintroducing a second full-width identity index is a write-amplification +regression even when retained `metrics.db` size looks small: every sample would +dirty both trees in the WAL and every checkpoint would copy both sets of pages +back into the main file. Legacy databases may keep their existing unique index +authoritative while the replacement index is built, but that O(rows) migration +must run on deferred startup maintenance and swap indexes in one SQLite +transaction so constructor latency, concurrent reads, uniqueness, and crash +recovery remain intact. The deterministic issue-1124 profile must continue to +attribute writes by logical payload, table/index `dbstat` pages, page-cache +writes/spills, WAL frames, checkpoint frames, process write bytes where the +host exposes them, and traced sync calls rather than treating final file size +as a proxy. Its checked invariant persists 157,452 samples across 2,197 mixed +provider resources and caps the no-auto-checkpoint workload at 40,000 WAL +frames; the former four-index schema produces 50,516 frames while the +consolidated schema produces 35,030. Retention must also return freed SQLite pages to the OS proportionally to the current freelist, bounded per cycle, and on every retention cycle rather than only when that cycle deleted rows: a fixed small diff --git a/pkg/metrics/issue1124_write_amplification_test.go b/pkg/metrics/issue1124_write_amplification_test.go new file mode 100644 index 0000000000..cd2d6981a9 --- /dev/null +++ b/pkg/metrics/issue1124_write_amplification_test.go @@ -0,0 +1,1109 @@ +package metrics + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + sqlite3 "modernc.org/sqlite" +) + +const issue1124ProfileEnv = "PULSE_METRICS_WRITE_PROFILE" + +type issue1124DBStat struct { + Name string `json:"name"` + Pages int64 `json:"pages"` + PageBytes int64 `json:"page_bytes"` + PayloadBytes int64 `json:"payload_bytes"` +} + +type issue1124ProfileReport struct { + SchemaMode string `json:"schema_mode"` + AutoCheckpoint bool `json:"auto_checkpoint"` + Ticks int `json:"ticks"` + Resources int `json:"resources"` + Samples int `json:"samples"` + WriteCalls int `json:"write_calls"` + LogicalPayloadBytes int64 `json:"logical_payload_bytes"` + Elapsed time.Duration `json:"elapsed"` + WALBytes int64 `json:"wal_bytes"` + WALFrames int64 `json:"wal_frames"` + CacheWritesBefore int64 `json:"cache_writes_before_checkpoint"` + CacheWritesAfter int64 `json:"cache_writes_after_checkpoint"` + CacheSpills int64 `json:"cache_spills"` + PhysicalWriteBytes int64 `json:"physical_write_bytes"` + CheckpointWriteBytes int64 `json:"checkpoint_write_bytes"` + CheckpointedFrames int64 `json:"checkpointed_frames"` + MainBytesBefore int64 `json:"main_bytes_before_checkpoint"` + MainBytesAfter int64 `json:"main_bytes_after_checkpoint"` + PageSize int64 `json:"page_size"` + PageCount int64 `json:"page_count"` + FreelistCount int64 `json:"freelist_count"` + ReadCount int64 `json:"concurrent_read_count"` + MaxReadLatency time.Duration `json:"max_read_latency"` + RestartElapsed time.Duration `json:"restart_elapsed"` + RestartMainBytes int64 `json:"restart_main_bytes"` + RestartPageCount int64 `json:"restart_page_count"` + RestartFreelist int64 `json:"restart_freelist_count"` + RetentionElapsed time.Duration `json:"retention_elapsed"` + RetentionRows int `json:"retention_rows"` + RetentionMainBytes int64 `json:"retention_main_bytes"` + RetentionPageCount int64 `json:"retention_page_count"` + RetentionFreelist int64 `json:"retention_freelist_count"` + DBStats []issue1124DBStat `json:"db_stats"` +} + +type issue1124Series struct { + resourceType string + resourceID string + metricTypes []string +} + +func TestStoreConsolidatesLegacyMetricsIndexes(t *testing.T) { + suppressTestLogs(t) + dbPath := filepath.Join(t.TempDir(), "metrics.db") + issue1124CreateLegacyStore(t, dbPath, 2_000, false) + + cfg := DefaultConfig(filepath.Dir(dbPath)) + cfg.DBPath = dbPath + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("migrate legacy store: %v", err) + } + if err := store.WaitForMaintenance(30 * time.Second); err != nil { + _ = store.Close() + t.Fatalf("wait for post-migration maintenance: %v", err) + } + issue1124AssertConsolidatedIndexes(t, store) + + var rows int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&rows); err != nil { + _ = store.Close() + t.Fatalf("count migrated rows: %v", err) + } + if rows != 2_000 { + _ = store.Close() + t.Fatalf("migrated rows=%d, want 2000", rows) + } + + ts := time.Unix(1_700_001_999, 0) + store.WriteBatchSync([]WriteMetric{{ + ResourceType: "vm", + ResourceID: "legacy-1999", + MetricType: "cpu", + Value: 99, + Timestamp: ts, + Tier: TierRaw, + }}) + var value float64 + if err := store.db.QueryRow(` + SELECT value FROM metrics + WHERE resource_type = 'vm' AND resource_id = 'legacy-1999' + AND metric_type = 'cpu' AND tier = 'raw' AND timestamp = ? + `, ts.Unix()).Scan(&value); err != nil { + _ = store.Close() + t.Fatalf("query migrated upsert: %v", err) + } + if value != 99 { + _ = store.Close() + t.Fatalf("migrated upsert value=%v, want 99", value) + } + if err := store.Close(); err != nil { + t.Fatalf("close migrated store: %v", err) + } + + restarted, err := NewStore(cfg) + if err != nil { + t.Fatalf("restart migrated store: %v", err) + } + issue1124AssertConsolidatedIndexes(t, restarted) + if err := restarted.Close(); err != nil { + t.Fatalf("close restarted store: %v", err) + } +} + +func TestStoreIdentityMigrationDeduplicatesLegacyRows(t *testing.T) { + suppressTestLogs(t) + dbPath := filepath.Join(t.TempDir(), "metrics.db") + issue1124CreateLegacyStore(t, dbPath, 2, true) + + cfg := DefaultConfig(filepath.Dir(dbPath)) + cfg.DBPath = dbPath + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("migrate duplicate legacy store: %v", err) + } + defer store.Close() + issue1124AssertConsolidatedIndexes(t, store) + + var count int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&count); err != nil { + t.Fatalf("count deduplicated rows: %v", err) + } + if count != 1 { + t.Fatalf("deduplicated row count=%d, want 1", count) + } + + ts := time.Unix(1_700_000_000, 0) + store.WriteBatchSync([]WriteMetric{{ + ResourceType: "vm", + ResourceID: "legacy-0000", + MetricType: "cpu", + Value: 42, + Timestamp: ts, + Tier: TierRaw, + }}) + var value float64 + if err := store.db.QueryRow(`SELECT value FROM metrics`).Scan(&value); err != nil { + t.Fatalf("query deduplicated value: %v", err) + } + if value != 42 { + t.Fatalf("upsert after dedupe value=%v, want 42", value) + } +} + +func TestStoreIdentityMigrationRecoversAfterCrash(t *testing.T) { + if os.Getenv("PULSE_METRICS_IDENTITY_CRASH_CHILD") == "1" { + metricsIdentityMigrationHook = func() { + os.Exit(88) + } + dbPath := os.Getenv("PULSE_METRICS_IDENTITY_CRASH_DB") + cfg := DefaultConfig(filepath.Dir(dbPath)) + cfg.DBPath = dbPath + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + store, err := NewStore(cfg) + if err != nil { + os.Exit(90) + } + _ = store.WaitForMaintenance(30 * time.Second) + os.Exit(89) + } + + suppressTestLogs(t) + dbPath := filepath.Join(t.TempDir(), "metrics.db") + issue1124CreateLegacyStore(t, dbPath, 10_000, false) + + cmd := exec.Command(os.Args[0], "-test.run=^TestStoreIdentityMigrationRecoversAfterCrash$") + cmd.Env = append( + os.Environ(), + "PULSE_METRICS_IDENTITY_CRASH_CHILD=1", + "PULSE_METRICS_IDENTITY_CRASH_DB="+dbPath, + ) + err := cmd.Run() + exitErr, ok := err.(*exec.ExitError) + if !ok || exitErr.ExitCode() != 88 { + t.Fatalf("migration crash child error=%v, want exit status 88", err) + } + + cfg := DefaultConfig(filepath.Dir(dbPath)) + cfg.DBPath = dbPath + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("recover store after interrupted migration: %v", err) + } + defer store.Close() + if err := store.WaitForMaintenance(30 * time.Second); err != nil { + t.Fatalf("wait for crash-recovery migration: %v", err) + } + issue1124AssertConsolidatedIndexes(t, store) + + var rows int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&rows); err != nil { + t.Fatalf("count rows after crash recovery: %v", err) + } + if rows != 10_000 { + t.Fatalf("rows after crash recovery=%d, want 10000", rows) + } + var integrity string + if err := store.db.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil { + t.Fatalf("integrity check after crash recovery: %v", err) + } + if integrity != "ok" { + t.Fatalf("integrity check after crash recovery=%q, want ok", integrity) + } +} + +func TestStoreIdentityMigrationBackupRestore(t *testing.T) { + suppressTestLogs(t) + dir := t.TempDir() + legacyPath := filepath.Join(dir, "legacy.db") + issue1124CreateLegacyStore(t, legacyPath, 5_000, false) + + legacyBackup := filepath.Join(dir, "legacy-backup.db") + issue1124CopyFile(t, legacyPath, legacyBackup) + cfg := DefaultConfig(dir) + cfg.DBPath = legacyBackup + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + restored, err := NewStore(cfg) + if err != nil { + t.Fatalf("restore legacy backup: %v", err) + } + if err := restored.WaitForMaintenance(30 * time.Second); err != nil { + _ = restored.Close() + t.Fatalf("legacy backup maintenance: %v", err) + } + issue1124AssertConsolidatedIndexes(t, restored) + var rows int + if err := restored.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&rows); err != nil { + _ = restored.Close() + t.Fatalf("count restored legacy backup: %v", err) + } + if rows != 5_000 { + _ = restored.Close() + t.Fatalf("restored legacy rows=%d, want 5000", rows) + } + var integrity string + if err := restored.db.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil { + _ = restored.Close() + t.Fatalf("legacy backup integrity check: %v", err) + } + if integrity != "ok" { + _ = restored.Close() + t.Fatalf("legacy backup integrity=%q, want ok", integrity) + } + if err := restored.Close(); err != nil { + t.Fatalf("close restored legacy backup: %v", err) + } + + consolidatedBackup := filepath.Join(dir, "consolidated-backup.db") + issue1124CopyFile(t, legacyBackup, consolidatedBackup) + cfg.DBPath = consolidatedBackup + reopened, err := NewStore(cfg) + if err != nil { + t.Fatalf("restore consolidated backup: %v", err) + } + defer reopened.Close() + issue1124AssertConsolidatedIndexes(t, reopened) + if err := reopened.db.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil { + t.Fatalf("consolidated backup integrity check: %v", err) + } + if integrity != "ok" { + t.Fatalf("consolidated backup integrity=%q, want ok", integrity) + } +} + +func TestStoreIdentityMigrationKeepsReadersAvailable(t *testing.T) { + suppressTestLogs(t) + dbPath := filepath.Join(t.TempDir(), "metrics.db") + issue1124CreateLegacyStore(t, dbPath, 10_000, false) + + migrationEntered := make(chan struct{}) + releaseMigration := make(chan struct{}) + var hookOnce sync.Once + metricsIdentityMigrationHook = func() { + hookOnce.Do(func() { close(migrationEntered) }) + <-releaseMigration + } + t.Cleanup(func() { + metricsIdentityMigrationHook = nil + }) + + cfg := DefaultConfig(filepath.Dir(dbPath)) + cfg.DBPath = dbPath + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("open legacy store: %v", err) + } + defer store.Close() + + select { + case <-migrationEntered: + case <-time.After(5 * time.Second): + t.Fatal("deferred identity migration did not start") + } + + queryDone := make(chan error, 1) + go func() { + _, err := store.Query( + "vm", + "legacy-0001", + "cpu", + time.Unix(1_699_999_999, 0), + time.Unix(1_700_000_002, 0), + 0, + ) + queryDone <- err + }() + select { + case err := <-queryDone: + if err != nil { + t.Fatalf("query during identity migration: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("query blocked behind deferred identity migration") + } + + writeDone := make(chan struct{}) + go func() { + store.WriteBatchSync([]WriteMetric{{ + ResourceType: "vm", + ResourceID: "during-migration", + MetricType: "cpu", + Value: 42, + Timestamp: time.Unix(1_700_020_000, 0), + Tier: TierRaw, + }}) + close(writeDone) + }() + close(releaseMigration) + + select { + case <-writeDone: + case <-time.After(35 * time.Second): + t.Fatal("write did not complete after identity migration released the WAL writer") + } + if err := store.WaitForMaintenance(35 * time.Second); err != nil { + t.Fatalf("wait for identity migration: %v", err) + } + issue1124AssertConsolidatedIndexes(t, store) + + var value float64 + if err := store.db.QueryRow(` + SELECT value FROM metrics + WHERE resource_type = 'vm' AND resource_id = 'during-migration' + AND metric_type = 'cpu' AND tier = 'raw' AND timestamp = 1700020000 + `).Scan(&value); err != nil { + t.Fatalf("query write completed during migration: %v", err) + } + if value != 42 { + t.Fatalf("write completed during migration value=%v, want 42", value) + } +} + +// TestMetricsWriteAmplificationInvariant persists 157,452 deterministic +// multi-provider samples, enough to force repeated B-tree splits and expose an +// accidentally restored identity index. The frame ceiling includes 14% margin +// over the consolidated schema's measured 35,030 frames; the v6.1.1 schema +// produces 50,516 frames and fails this invariant. +func TestMetricsWriteAmplificationInvariant(t *testing.T) { + suppressTestLogs(t) + dir := t.TempDir() + dbPath := filepath.Join(dir, "metrics.db") + cfg := DefaultConfig(dir) + cfg.DBPath = dbPath + cfg.FlushInterval = time.Hour + cfg.RollupInterval = time.Hour + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + defer store.Close() + if err := store.WaitForMaintenance(30 * time.Second); err != nil { + t.Fatalf("startup maintenance: %v", err) + } + issue1124AssertConsolidatedIndexes(t, store) + + store.db.SetMaxOpenConns(1) + store.db.SetMaxIdleConns(1) + if _, err := store.db.Exec(`PRAGMA wal_autocheckpoint=0`); err != nil { + t.Fatalf("disable auto-checkpoint: %v", err) + } + if _, err := store.db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil { + t.Fatalf("reset WAL: %v", err) + } + + const ticks = 12 + series := issue1124Estate() + batches := issue1124ProviderBatches(series) + base := time.Unix(1_700_000_000, 0).UTC() + samples := 0 + for tick := 0; tick < ticks; tick++ { + ts := base.Add(time.Duration(tick) * 10 * time.Second) + for _, template := range batches { + batch := make([]WriteMetric, len(template)) + for i, metric := range template { + metric.Timestamp = ts + metric.Value = float64((tick+i)%10_000) / 100 + batch[i] = metric + } + store.WriteBatchSync(batch) + samples += len(batch) + } + } + + if samples != 157_452 { + t.Fatalf("profile samples=%d, want 157452", samples) + } + var persisted int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&persisted); err != nil { + t.Fatalf("count persisted metrics: %v", err) + } + if persisted != samples { + t.Fatalf("persisted rows=%d, want %d", persisted, samples) + } + + pageSize := issue1124PragmaInt(t, store, "page_size") + walBytes := issue1124FileSize(t, dbPath+"-wal") + walFrames := (walBytes - 32) / (pageSize + 24) + const maxWALFrames = 40_000 + if walFrames > maxWALFrames { + t.Fatalf( + "WAL frames=%d for %d samples, limit=%d; redundant indexes or transaction churn regressed write amplification", + walFrames, + samples, + maxWALFrames, + ) + } + + var busy, logFrames, checkpointed int64 + if err := store.db.QueryRow(`PRAGMA wal_checkpoint(PASSIVE)`).Scan(&busy, &logFrames, &checkpointed); err != nil { + t.Fatalf("checkpoint: %v", err) + } + if busy != 0 || logFrames != walFrames || checkpointed != walFrames { + t.Fatalf( + "checkpoint counters busy=%d log=%d checkpointed=%d, want busy=0 log=checkpointed=%d", + busy, + logFrames, + checkpointed, + walFrames, + ) + } +} + +// TestIssue1124WriteAmplificationProfile is an opt-in measurement harness for +// comparing released schemas and candidate changes under the same deterministic +// multi-provider estate. It reports WAL frames and per-B-tree dbstat pages +// separately, so retained database size is never used as a write-volume proxy. +// +// Run with PULSE_METRICS_WRITE_PROFILE=full (v6.1.1 layout) or consolidated +// and optionally PULSE_METRICS_WRITE_PROFILE_TICKS=N. The default 30 ticks +// persist 393,630 samples across 2,197 resources. Set +// PULSE_METRICS_WRITE_PROFILE_AUTOCHECKPOINT=1 for production checkpoint/fsync +// tracing, PULSE_METRICS_WRITE_PROFILE_READERS=1 for a paced concurrent reader, +// or PULSE_METRICS_WRITE_PROFILE_RETENTION=1 for retention/compaction counters. +func TestIssue1124WriteAmplificationProfile(t *testing.T) { + mode := os.Getenv(issue1124ProfileEnv) + if mode == "" { + t.Skip("set " + issue1124ProfileEnv + " to run the write-amplification profile") + } + if mode != "full" && mode != "consolidated" && mode != "full-batched" && + mode != "consolidated-batched" && mode != "without-rowid" { + t.Fatalf("%s must be full, consolidated, full-batched, consolidated-batched, or without-rowid, got %q", issue1124ProfileEnv, mode) + } + + ticks := 30 + if raw := os.Getenv("PULSE_METRICS_WRITE_PROFILE_TICKS"); raw != "" { + parsed, err := strconv.Atoi(raw) + if err != nil || parsed < 1 || parsed > 720 { + t.Fatalf("PULSE_METRICS_WRITE_PROFILE_TICKS must be between 1 and 720, got %q", raw) + } + ticks = parsed + } + + suppressTestLogs(t) + dir := t.TempDir() + dbPath := filepath.Join(dir, "metrics.db") + cfg := DefaultConfig(dir) + cfg.DBPath = dbPath + cfg.FlushInterval = time.Hour + cfg.RollupInterval = time.Hour + cfg.RetentionRaw = 10 * 365 * 24 * time.Hour + + store, err := NewStore(cfg) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + if err := store.WaitForMaintenance(30 * time.Second); err != nil { + _ = store.Close() + t.Fatalf("startup maintenance: %v", err) + } + + // Pin profiling to one configured connection so wal_autocheckpoint is + // deterministic. Production's four-connection concurrency has separate SLO + // coverage; this profile isolates bytes/pages written per logical sample. + store.db.SetMaxOpenConns(1) + store.db.SetMaxIdleConns(1) + autoCheckpoint := os.Getenv("PULSE_METRICS_WRITE_PROFILE_AUTOCHECKPOINT") == "1" + if !autoCheckpoint { + if _, err := store.db.Exec(`PRAGMA wal_autocheckpoint=0`); err != nil { + _ = store.Close() + t.Fatalf("disable auto-checkpoint: %v", err) + } + } + + if mode == "full" || mode == "full-batched" { + // Reconstruct the v6.1.1 schema so released and consolidated layouts + // remain comparable after the production migration lands. + if _, err := store.db.Exec(` + DROP INDEX idx_metrics_lookup; + CREATE INDEX idx_metrics_lookup + ON metrics(resource_type, resource_id, metric_type, tier, timestamp); + CREATE UNIQUE INDEX idx_metrics_unique + ON metrics(resource_type, resource_id, metric_type, timestamp, tier); + `); err != nil { + _ = store.Close() + t.Fatalf("install v6.1.1 index layout: %v", err) + } + } + if mode == "without-rowid" { + if _, err := store.db.Exec(` + DROP TABLE metrics; + CREATE TABLE metrics ( + 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', + PRIMARY KEY(resource_type, resource_id, metric_type, tier, timestamp) + ) WITHOUT ROWID; + CREATE INDEX idx_metrics_tier_time ON metrics(tier, timestamp); + CREATE INDEX idx_metrics_query_all + ON metrics(resource_type, resource_id, tier, timestamp, metric_type); + `); err != nil { + _ = store.Close() + t.Fatalf("install WITHOUT ROWID schema: %v", err) + } + } + + if _, err := store.db.Exec(`PRAGMA wal_checkpoint(TRUNCATE)`); err != nil { + _ = store.Close() + t.Fatalf("reset WAL: %v", err) + } + _ = issue1124DBStatus(t, store, sqlite3.DBStatusCacheWrite, true) + _ = issue1124DBStatus(t, store, sqlite3.DBStatusCacheSpill, true) + processWritesBefore := issue1124ProcessWriteBytes() + + series := issue1124Estate() + batches := issue1124ProviderBatches(series) + samplesPerTick := 0 + for _, batch := range batches { + samplesPerTick += len(batch) + } + + var readCount atomic.Int64 + var maxReadNanos atomic.Int64 + var readStop chan struct{} + var readDone chan struct{} + var readErr atomic.Value + base := time.Unix(1_700_000_000, 0).UTC() + if os.Getenv("PULSE_METRICS_WRITE_PROFILE_READERS") == "1" { + readStop = make(chan struct{}) + readDone = make(chan struct{}) + go func() { + defer close(readDone) + for { + select { + case <-readStop: + return + default: + } + started := time.Now() + _, queryErr := store.Query( + "dockerContainer", + "docker-container-0000", + "cpu", + base.Add(-time.Minute), + base.Add(time.Duration(ticks+1)*10*time.Second), + 0, + ) + if queryErr != nil { + readErr.Store(queryErr) + return + } + readCount.Add(1) + elapsed := time.Since(started).Nanoseconds() + for { + previous := maxReadNanos.Load() + if elapsed <= previous || maxReadNanos.CompareAndSwap(previous, elapsed) { + break + } + } + time.Sleep(5 * time.Millisecond) + } + }() + } + + logicalBytes := int64(0) + started := time.Now() + for tick := 0; tick < ticks; tick++ { + ts := base.Add(time.Duration(tick) * 10 * time.Second) + var tickBatch []WriteMetric + for _, template := range batches { + batch := make([]WriteMetric, len(template)) + for i, metric := range template { + metric.Timestamp = ts + metric.Value = float64((tick+i)%10_000) / 100 + batch[i] = metric + logicalBytes += issue1124LogicalBytes(metric) + } + if mode == "full-batched" || mode == "consolidated-batched" { + tickBatch = append(tickBatch, batch...) + } else { + store.WriteBatchSync(batch) + } + } + if len(tickBatch) > 0 { + store.WriteBatchSync(tickBatch) + } + } + writeElapsed := time.Since(started) + if readStop != nil { + close(readStop) + <-readDone + } + if value := readErr.Load(); value != nil { + _ = store.Close() + t.Fatalf("concurrent query: %v", value) + } + + pageSize := issue1124PragmaInt(t, store, "page_size") + pageCount := issue1124PragmaInt(t, store, "page_count") + freelistCount := issue1124PragmaInt(t, store, "freelist_count") + walBytes := issue1124FileSize(t, dbPath+"-wal") + mainBefore := issue1124FileSize(t, dbPath) + cacheWritesBefore := issue1124DBStatus(t, store, sqlite3.DBStatusCacheWrite, false) + cacheSpills := issue1124DBStatus(t, store, sqlite3.DBStatusCacheSpill, false) + processWritesAfterWAL := issue1124ProcessWriteBytes() + walFrames := int64(0) + if walBytes >= 32 { + walFrames = (walBytes - 32) / (pageSize + 24) + } + + var busy, logFrames, checkpointed int64 + if err := store.db.QueryRow(`PRAGMA wal_checkpoint(PASSIVE)`).Scan(&busy, &logFrames, &checkpointed); err != nil { + _ = store.Close() + t.Fatalf("checkpoint: %v", err) + } + if busy != 0 { + _ = store.Close() + t.Fatalf("checkpoint remained busy: busy=%d log=%d checkpointed=%d", busy, logFrames, checkpointed) + } + mainAfter := issue1124FileSize(t, dbPath) + cacheWritesAfter := issue1124DBStatus(t, store, sqlite3.DBStatusCacheWrite, false) + processWritesAfterCheckpoint := issue1124ProcessWriteBytes() + dbStats := issue1124ReadDBStats(t, store) + + expectedSamples := ticks * samplesPerTick + var rowCount int + if err := store.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&rowCount); err != nil { + _ = store.Close() + t.Fatalf("count persisted metrics: %v", err) + } + if rowCount != expectedSamples { + _ = store.Close() + t.Fatalf("persisted rows=%d, expected=%d", rowCount, expectedSamples) + } + + if err := store.Close(); err != nil { + t.Fatalf("close store: %v", err) + } + + restartStarted := time.Now() + restarted, err := NewStore(cfg) + if err != nil { + t.Fatalf("restart store: %v", err) + } + if err := restarted.WaitForMaintenance(30 * time.Second); err != nil { + _ = restarted.Close() + t.Fatalf("restart maintenance: %v", err) + } + var integrity string + if err := restarted.db.QueryRow(`PRAGMA integrity_check`).Scan(&integrity); err != nil { + _ = restarted.Close() + t.Fatalf("restart integrity check: %v", err) + } + if integrity != "ok" { + _ = restarted.Close() + t.Fatalf("restart integrity check returned %q", integrity) + } + var restartedRows int + if err := restarted.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&restartedRows); err != nil { + _ = restarted.Close() + t.Fatalf("restart count: %v", err) + } + if restartedRows != expectedSamples { + _ = restarted.Close() + t.Fatalf("restart rows=%d, expected=%d", restartedRows, expectedSamples) + } + restartElapsed := time.Since(restartStarted) + restartMainBytes := issue1124FileSize(t, dbPath) + restartPageCount := issue1124PragmaInt(t, restarted, "page_count") + restartFreelist := issue1124PragmaInt(t, restarted, "freelist_count") + var ( + retentionElapsed time.Duration + retentionRows = restartedRows + retentionMainBytes = restartMainBytes + retentionPageCount = restartPageCount + retentionFreelist = restartFreelist + ) + if os.Getenv("PULSE_METRICS_WRITE_PROFILE_RETENTION") == "1" { + restarted.config.RetentionRaw = time.Nanosecond + restarted.config.RetentionMinute = time.Nanosecond + restarted.config.RetentionHourly = time.Nanosecond + restarted.config.RetentionDaily = time.Nanosecond + retentionStarted := time.Now() + restarted.runRetention() + retentionElapsed = time.Since(retentionStarted) + if err := restarted.db.QueryRow(`SELECT COUNT(*) FROM metrics`).Scan(&retentionRows); err != nil { + _ = restarted.Close() + t.Fatalf("count rows after retention: %v", err) + } + retentionMainBytes = issue1124FileSize(t, dbPath) + retentionPageCount = issue1124PragmaInt(t, restarted, "page_count") + retentionFreelist = issue1124PragmaInt(t, restarted, "freelist_count") + } + if err := restarted.Close(); err != nil { + t.Fatalf("close restarted store: %v", err) + } + + report := issue1124ProfileReport{ + SchemaMode: mode, + AutoCheckpoint: autoCheckpoint, + Ticks: ticks, + Resources: len(series), + Samples: expectedSamples, + WriteCalls: ticks * len(batches), + LogicalPayloadBytes: logicalBytes, + Elapsed: writeElapsed, + WALBytes: walBytes, + WALFrames: walFrames, + CacheWritesBefore: cacheWritesBefore, + CacheWritesAfter: cacheWritesAfter, + CacheSpills: cacheSpills, + PhysicalWriteBytes: issue1124CounterDelta(processWritesBefore, processWritesAfterWAL), + CheckpointWriteBytes: issue1124CounterDelta(processWritesAfterWAL, processWritesAfterCheckpoint), + CheckpointedFrames: checkpointed, + MainBytesBefore: mainBefore, + MainBytesAfter: mainAfter, + PageSize: pageSize, + PageCount: pageCount, + FreelistCount: freelistCount, + ReadCount: readCount.Load(), + MaxReadLatency: time.Duration(maxReadNanos.Load()), + RestartElapsed: restartElapsed, + RestartMainBytes: restartMainBytes, + RestartPageCount: restartPageCount, + RestartFreelist: restartFreelist, + RetentionElapsed: retentionElapsed, + RetentionRows: retentionRows, + RetentionMainBytes: retentionMainBytes, + RetentionPageCount: retentionPageCount, + RetentionFreelist: retentionFreelist, + DBStats: dbStats, + } + if mode == "full-batched" || mode == "consolidated-batched" { + report.WriteCalls = ticks + } + encoded, err := json.Marshal(report) + if err != nil { + t.Fatalf("marshal profile: %v", err) + } + t.Logf("ISSUE1124_PROFILE %s", encoded) +} + +func issue1124Estate() []issue1124Series { + const ( + dockerContainers = 975 + proxmoxGuests = 400 + agentHosts = 120 + kubernetesPods = 300 + storageTargets = 250 + physicalDisks = 152 + ) + + series := make([]issue1124Series, 0, dockerContainers+proxmoxGuests+agentHosts+kubernetesPods+storageTargets+physicalDisks) + appendSeries := func(resourceType, prefix string, count int, metricTypes ...string) { + for i := 0; i < count; i++ { + series = append(series, issue1124Series{ + resourceType: resourceType, + resourceID: fmt.Sprintf("%s-%04d", prefix, i), + metricTypes: metricTypes, + }) + } + } + appendSeries("dockerContainer", "docker-container", dockerContainers, "cpu", "memory", "disk", "netin", "netout", "diskread", "diskwrite") + appendSeries("vm", "proxmox-guest", proxmoxGuests, "cpu", "memory", "disk", "netin", "netout", "diskread", "diskwrite") + appendSeries("agent", "agent-host", agentHosts, "cpu", "memory", "disk", "netin", "netout", "diskread", "diskwrite") + appendSeries("pod", "kubernetes-pod", kubernetesPods, "cpu", "memory", "netin", "netout") + appendSeries("storage", "storage-target", storageTargets, "usage", "used", "total", "avail") + appendSeries("disk", "physical-disk", physicalDisks, "temperature", "wearout", "power_on_hours") + return series +} + +func issue1124ProviderBatches(series []issue1124Series) [][]WriteMetric { + byProvider := make(map[string][]WriteMetric) + order := []string{"dockerContainer", "vm", "agent", "pod", "storage", "disk"} + for _, resource := range series { + for _, metricType := range resource.metricTypes { + byProvider[resource.resourceType] = append(byProvider[resource.resourceType], WriteMetric{ + ResourceType: resource.resourceType, + ResourceID: resource.resourceID, + MetricType: metricType, + Tier: TierRaw, + }) + } + } + batches := make([][]WriteMetric, 0, len(order)) + for _, provider := range order { + batches = append(batches, byProvider[provider]) + } + return batches +} + +func issue1124LogicalBytes(metric WriteMetric) int64 { + // Application payload only: encoded strings plus value and timestamp. + // SQLite page headers, record headers, rowids, indexes, WAL headers, and + // checkpoint rewrites are intentionally excluded from this denominator. + return int64(len(metric.ResourceType)+len(metric.ResourceID)+len(metric.MetricType)+len(metric.Tier)) + 16 +} + +func issue1124PragmaInt(t *testing.T, store *Store, name string) int64 { + t.Helper() + var value int64 + if err := store.db.QueryRow("PRAGMA " + name).Scan(&value); err != nil { + t.Fatalf("PRAGMA %s: %v", name, err) + } + return value +} + +func issue1124DBStatus(t *testing.T, store *Store, op sqlite3.DBStatusOp, reset bool) int64 { + t.Helper() + conn, err := store.db.DB.Conn(context.Background()) + if err != nil { + t.Fatalf("open SQLite status connection: %v", err) + } + defer conn.Close() + + var current int + if err := conn.Raw(func(driverConn any) error { + status, ok := driverConn.(sqlite3.DBStatus) + if !ok { + return fmt.Errorf("SQLite driver does not expose DBStatus") + } + value, _, err := status.Status(op, reset) + current = value + return err + }); err != nil { + t.Fatalf("read SQLite DB status %d: %v", op, err) + } + return int64(current) +} + +func issue1124FileSize(t *testing.T, path string) int64 { + t.Helper() + info, err := os.Stat(path) + if err != nil { + t.Fatalf("stat %s: %v", path, err) + } + return info.Size() +} + +func issue1124ProcessWriteBytes() int64 { + data, err := os.ReadFile("/proc/self/io") + if err != nil { + return -1 + } + for _, line := range strings.Split(string(data), "\n") { + key, value, ok := strings.Cut(line, ":") + if !ok || strings.TrimSpace(key) != "write_bytes" { + continue + } + parsed, err := strconv.ParseInt(strings.TrimSpace(value), 10, 64) + if err == nil { + return parsed + } + } + return -1 +} + +func issue1124CounterDelta(before, after int64) int64 { + if before < 0 || after < before { + return -1 + } + return after - before +} + +func issue1124CopyFile(t *testing.T, source, destination string) { + t.Helper() + data, err := os.ReadFile(source) + if err != nil { + t.Fatalf("read backup source %s: %v", source, err) + } + if err := os.WriteFile(destination, data, 0o600); err != nil { + t.Fatalf("write backup destination %s: %v", destination, err) + } +} + +func issue1124ReadDBStats(t *testing.T, store *Store) []issue1124DBStat { + t.Helper() + rows, err := store.db.Query(` + SELECT name, COUNT(*), SUM(pgsize), SUM(payload) + FROM dbstat + WHERE name IN ( + 'metrics', + 'sqlite_sequence', + 'idx_metrics_lookup', + 'idx_metrics_identity', + 'idx_metrics_tier_time', + 'idx_metrics_query_all', + 'idx_metrics_unique' + ) + GROUP BY name + ORDER BY name + `) + if err != nil { + t.Fatalf("query dbstat: %v", err) + } + defer rows.Close() + + stats := make([]issue1124DBStat, 0, 7) + for rows.Next() { + var stat issue1124DBStat + if err := rows.Scan(&stat.Name, &stat.Pages, &stat.PageBytes, &stat.PayloadBytes); err != nil { + t.Fatalf("scan dbstat: %v", err) + } + stats = append(stats, stat) + } + if err := rows.Err(); err != nil { + t.Fatalf("dbstat rows: %v", err) + } + return stats +} + +func issue1124CreateLegacyStore(t *testing.T, dbPath string, rows int, duplicates bool) { + t.Helper() + db, err := sql.Open("sqlite", dbPath) + if err != nil { + t.Fatalf("open legacy store: %v", err) + } + defer db.Close() + + if _, err := db.Exec(` + 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 INDEX idx_metrics_lookup + ON metrics(resource_type, resource_id, metric_type, tier, timestamp); + CREATE INDEX idx_metrics_tier_time ON metrics(tier, timestamp); + CREATE INDEX idx_metrics_query_all + ON metrics(resource_type, resource_id, tier, timestamp, metric_type); + CREATE TABLE metrics_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ); + `); err != nil { + t.Fatalf("create legacy schema: %v", err) + } + if !duplicates { + if _, err := db.Exec(` + CREATE UNIQUE INDEX idx_metrics_unique + ON metrics(resource_type, resource_id, metric_type, timestamp, tier) + `); err != nil { + t.Fatalf("create legacy unique index: %v", err) + } + } + + tx, err := db.Begin() + if err != nil { + t.Fatalf("begin legacy seed: %v", err) + } + stmt, err := tx.Prepare(` + INSERT INTO metrics(resource_type, resource_id, metric_type, value, timestamp, tier) + VALUES('vm', ?, 'cpu', ?, ?, 'raw') + `) + if err != nil { + _ = tx.Rollback() + t.Fatalf("prepare legacy seed: %v", err) + } + for i := 0; i < rows; i++ { + resourceIndex := i + timestamp := int64(1_700_000_000 + i) + if duplicates { + resourceIndex = 0 + timestamp = 1_700_000_000 + } + if _, err := stmt.Exec(fmt.Sprintf("legacy-%04d", resourceIndex), float64(i), timestamp); err != nil { + _ = stmt.Close() + _ = tx.Rollback() + t.Fatalf("seed legacy row %d: %v", i, err) + } + } + if err := stmt.Close(); err != nil { + _ = tx.Rollback() + t.Fatalf("close legacy seed statement: %v", err) + } + if err := tx.Commit(); err != nil { + t.Fatalf("commit legacy seed: %v", err) + } +} + +func issue1124AssertConsolidatedIndexes(t *testing.T, store *Store) { + t.Helper() + matches, err := store.metricsIndexMatches("idx_metrics_lookup", true, metricsIdentityColumns) + if err != nil { + t.Fatalf("inspect consolidated lookup index: %v", err) + } + if !matches { + t.Fatal("idx_metrics_lookup is not the expected unique identity/range index") + } + exists, err := store.metricsIndexExists("idx_metrics_unique") + if err != nil { + t.Fatalf("inspect obsolete unique index: %v", err) + } + if exists { + t.Fatal("obsolete idx_metrics_unique still exists") + } + + rows, err := store.db.Query(`PRAGMA index_list(metrics)`) + if err != nil { + t.Fatalf("list consolidated indexes: %v", err) + } + defer rows.Close() + indexes := make(map[string]bool) + for rows.Next() { + var ( + sequence int + name string + unique int + origin string + partial int + ) + if err := rows.Scan(&sequence, &name, &unique, &origin, &partial); err != nil { + t.Fatalf("scan consolidated index: %v", err) + } + indexes[name] = unique == 1 + } + if err := rows.Err(); err != nil { + t.Fatalf("consolidated index rows: %v", err) + } + want := map[string]bool{ + "idx_metrics_lookup": true, + "idx_metrics_query_all": false, + "idx_metrics_tier_time": false, + } + if len(indexes) != len(want) { + t.Fatalf("metrics indexes=%v, want %v", indexes, want) + } + for name, unique := range want { + if got, ok := indexes[name]; !ok || got != unique { + t.Fatalf("metrics index %s unique=%v present=%v, want unique=%v", name, got, ok, unique) + } + } +} diff --git a/pkg/metrics/store.go b/pkg/metrics/store.go index 4af524d2d1..8042174416 100644 --- a/pkg/metrics/store.go +++ b/pkg/metrics/store.go @@ -183,7 +183,10 @@ type maintenanceRequest struct { done chan struct{} } -var startupMaintenanceHook func() +var ( + startupMaintenanceHook func() + metricsIdentityMigrationHook func() +) // Store provides persistent metrics storage type Store struct { @@ -201,6 +204,7 @@ type Store struct { doneCh chan struct{} stopOnce sync.Once stopping atomic.Bool + identityMigrationPending atomic.Bool commercialRetentionSeconds atomic.Int64 commercialPurgeEligibleAt atomic.Int64 } @@ -337,10 +341,6 @@ func (s *Store) initSchema() error { tier TEXT NOT NULL DEFAULT 'raw' ); - -- Index for efficient queries by resource and time - CREATE INDEX IF NOT EXISTS idx_metrics_lookup - ON metrics(resource_type, resource_id, metric_type, tier, timestamp); - -- Index for retention pruning CREATE INDEX IF NOT EXISTS idx_metrics_tier_time ON metrics(tier, timestamp); @@ -361,7 +361,7 @@ func (s *Store) initSchema() error { return fmt.Errorf("failed to create schema: %w", err) } - if err := s.ensureMetricsUniqueIndex(); err != nil { + if err := s.ensureMetricsIdentityIndex(); err != nil { return err } @@ -369,56 +369,207 @@ func (s *Store) initSchema() error { return nil } -func (s *Store) ensureMetricsUniqueIndex() error { - const createUniqueIndex = ` - CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_unique - ON metrics(resource_type, resource_id, metric_type, timestamp, tier); - ` +var metricsIdentityColumns = []string{"resource_type", "resource_id", "metric_type", "tier", "timestamp"} - _, err := s.db.Exec(createUniqueIndex) - if err == nil { +// ensureMetricsIdentityIndex keeps one B-tree for both metric identity and +// single-series range queries. Older databases have two indexes containing the +// same five columns in different orders: idx_metrics_lookup serves reads while +// idx_metrics_unique enforces identity. Every insert dirties both trees, which +// materially amplifies WAL and checkpoint writes. +func (s *Store) ensureMetricsIdentityIndex() error { + lookupCurrent, err := s.metricsIndexMatches("idx_metrics_lookup", true, metricsIdentityColumns) + if err != nil { + return fmt.Errorf("inspect metrics identity index: %w", err) + } + legacyUniqueExists, err := s.metricsIndexExists("idx_metrics_unique") + if err != nil { + return fmt.Errorf("inspect legacy metrics unique index: %w", err) + } + + if legacyUniqueExists { + // v6.1.1 and earlier already have an authoritative unique index. Keep + // serving with that crash-safe schema and defer the O(rows) rebuild to + // the startup-maintenance worker so NewStore latency stays bounded. + s.identityMigrationPending.Store(true) + log.Info().Msg("Scheduled metrics identity index consolidation") + return nil + } + if lookupCurrent { return nil } - // If the DB already contains duplicates (from older versions), creating the unique index - // will fail. Deduplicate and retry once. - lower := strings.ToLower(err.Error()) - if !strings.Contains(lower, "unique") && !strings.Contains(lower, "constraint") && !strings.Contains(lower, "duplicate") { - return fmt.Errorf("failed to create unique index for metrics rollups: %w", err) + return s.migrateMetricsIdentityIndex() +} + +func (s *Store) migrateMetricsIdentityIndex() error { + if err := s.replaceMetricsIdentityIndex(); err == nil { + log.Info().Msg("Consolidated metrics lookup and identity indexes") + return nil + } else if !isUniqueConstraintError(err) { + return fmt.Errorf("consolidate metrics identity index: %w", err) } - log.Warn().Err(err).Msg("Duplicate metrics detected; deduplicating before creating unique index") + log.Warn().Msg("Duplicate metrics detected; deduplicating before consolidating identity index") + if err := s.deduplicateMetrics(); err != nil { + return err + } + if err := s.replaceMetricsIdentityIndex(); err != nil { + return fmt.Errorf("consolidate metrics identity index after dedupe: %w", err) + } - tx, txErr := s.db.Begin() - if txErr != nil { - return fmt.Errorf("begin dedupe transaction: %w", txErr) + log.Info().Msg("Metrics deduplicated and identity index consolidated") + return nil +} + +func (s *Store) replaceMetricsIdentityIndex() error { + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("begin identity index migration: %w", err) } defer func() { _ = tx.Rollback() }() - _, txErr = tx.Exec(` - DELETE FROM metrics - WHERE id NOT IN ( - SELECT MIN(id) - FROM metrics - GROUP BY resource_type, resource_id, metric_type, timestamp, tier - ) - `) - if txErr != nil { - return fmt.Errorf("dedupe metrics: %w", txErr) + if _, err := tx.Exec(`DROP INDEX IF EXISTS idx_metrics_lookup`); err != nil { + return fmt.Errorf("drop old lookup index: %w", err) } - - if txErr := tx.Commit(); txErr != nil { - return fmt.Errorf("commit dedupe: %w", txErr) + if metricsIdentityMigrationHook != nil { + metricsIdentityMigrationHook() } - - if _, err := s.db.Exec(createUniqueIndex); err != nil { - return fmt.Errorf("failed to create unique index after dedupe: %w", err) + if _, err := tx.Exec(` + CREATE UNIQUE INDEX idx_metrics_lookup + ON metrics(resource_type, resource_id, metric_type, tier, timestamp) + `); err != nil { + return fmt.Errorf("create consolidated lookup index: %w", err) + } + if _, err := tx.Exec(`DROP INDEX IF EXISTS idx_metrics_unique`); err != nil { + return fmt.Errorf("drop old unique index: %w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit identity index migration: %w", err) } - - log.Info().Msg("Metrics deduplicated and unique index created") return nil } +func (s *Store) deduplicateMetrics() error { + tx, err := s.db.Begin() + if err != nil { + return fmt.Errorf("begin metrics dedupe transaction: %w", err) + } + defer func() { _ = tx.Rollback() }() + + if _, err := tx.Exec(` + DELETE FROM metrics + WHERE rowid NOT IN ( + SELECT MIN(rowid) + FROM metrics + GROUP BY resource_type, resource_id, metric_type, tier, timestamp + ) + `); err != nil { + return fmt.Errorf("dedupe metrics: %w", err) + } + if err := tx.Commit(); err != nil { + return fmt.Errorf("commit metrics dedupe: %w", err) + } + return nil +} + +func isUniqueConstraintError(err error) bool { + if err == nil { + return false + } + lower := strings.ToLower(err.Error()) + return strings.Contains(lower, "unique") || + strings.Contains(lower, "constraint") || + strings.Contains(lower, "duplicate") +} + +func (s *Store) metricsIndexExists(name string) (bool, error) { + rows, err := s.db.Query(`PRAGMA index_list(metrics)`) + if err != nil { + return false, err + } + defer rows.Close() + + for rows.Next() { + var ( + sequence int + index string + unique int + origin string + partial int + ) + if err := rows.Scan(&sequence, &index, &unique, &origin, &partial); err != nil { + return false, err + } + if index == name { + return true, nil + } + } + return false, rows.Err() +} + +func (s *Store) metricsIndexMatches(name string, wantUnique bool, wantColumns []string) (bool, error) { + rows, err := s.db.Query(`PRAGMA index_list(metrics)`) + if err != nil { + return false, err + } + + found := false + for rows.Next() { + var ( + sequence int + index string + unique int + origin string + partial int + ) + if err := rows.Scan(&sequence, &index, &unique, &origin, &partial); err != nil { + _ = rows.Close() + return false, err + } + if index == name { + found = (unique == 1) == wantUnique && partial == 0 + break + } + } + if err := rows.Close(); err != nil { + return false, err + } + if !found { + return false, nil + } + + columnRows, err := s.db.Query(` + SELECT name + FROM pragma_index_info(?) + ORDER BY seqno + `, name) + if err != nil { + return false, err + } + defer columnRows.Close() + + columns := make([]string, 0, len(wantColumns)) + for columnRows.Next() { + var columnName string + if err := columnRows.Scan(&columnName); err != nil { + return false, err + } + columns = append(columns, columnName) + } + if err := columnRows.Err(); err != nil { + return false, err + } + if len(columns) != len(wantColumns) { + return false, nil + } + for i := range columns { + if columns[i] != wantColumns[i] { + return false, nil + } + } + return true, nil +} + // migrateAutoVacuum ensures the database uses incremental auto-vacuum. // SQLite cannot switch from NONE to INCREMENTAL without a full VACUUM to // restructure the file, so we detect and convert on first run after upgrade. @@ -700,8 +851,15 @@ func (s *Store) runStartupMaintenance() { startupMaintenanceHook() } - // Run retention before any deferred vacuum work so restart cleanup trims - // stale rows before SQLite potentially rewrites the file. + if s.identityMigrationPending.Swap(false) { + if err := s.migrateMetricsIdentityIndex(); err != nil { + log.Error().Err(err).Msg("Deferred metrics identity index consolidation failed") + } + } + + // Run retention before the deferred auto-vacuum conversion so restart + // cleanup trims stale rows and redundant-index pages before SQLite + // potentially rewrites the file. s.runRetention() s.migrateAutoVacuum() @@ -834,7 +992,7 @@ func (s *Store) writeBatch(metrics []bufferedMetric) { stmt, err := tx.Prepare(` INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) VALUES (?, ?, ?, ?, ?, ?) - ON CONFLICT(resource_type, resource_id, metric_type, timestamp, tier) + ON CONFLICT(resource_type, resource_id, metric_type, tier, timestamp) DO UPDATE SET value = excluded.value, min_value = excluded.min_value, diff --git a/pkg/metrics/store_query_plan_test.go b/pkg/metrics/store_query_plan_test.go index b85572bda7..b25991ae2d 100644 --- a/pkg/metrics/store_query_plan_test.go +++ b/pkg/metrics/store_query_plan_test.go @@ -82,8 +82,8 @@ func TestQueryPlansUseIndexes(t *testing.T) { AND timestamp >= ? AND timestamp <= ? ORDER BY metric_type, timestamp ASC`, args: []any{"vm", "vm-1", "raw", int64(0), farFuture}, - // The planner uses an indexed SEARCH — it may pick idx_metrics_query_all, - // idx_metrics_unique, or idx_metrics_lookup depending on statistics. + // The planner uses an indexed SEARCH — it may pick idx_metrics_query_all + // or idx_metrics_lookup depending on statistics. // All are valid; the key invariant is that it does a SEARCH, not a SCAN. }, { @@ -110,7 +110,7 @@ func TestQueryPlansUseIndexes(t *testing.T) { ORDER BY resource_id, metric_type, timestamp ASC`, args: []any{"vm", "vm-1", "vm-2", "vm-3", "raw", int64(0), farFuture}, // QueryAllBatch is the anti-N+1 dashboard path. The planner may choose - // idx_metrics_query_all, idx_metrics_unique, or idx_metrics_lookup + // idx_metrics_query_all or idx_metrics_lookup // depending on statistics; the invariant is an indexed SEARCH rather // than a full table scan. }, @@ -191,8 +191,8 @@ func TestQueryPlansUseIndexes(t *testing.T) { args: []any{int64(60), int64(60), "minute", "raw", int64(0), farFuture}, // The SELECT filters on (tier, timestamp) without resource/metric // columns. SQLite uses an index-ordered scan (SCAN USING INDEX) - // on idx_metrics_unique. A TEMP B-TREE may still be used for - // GROUP BY. The INSERT side uses idx_metrics_unique for ON CONFLICT. + // on idx_metrics_lookup. A TEMP B-TREE may still be used for + // GROUP BY. The INSERT side uses the same unique index for ON CONFLICT. allowIndexScan: true, }, { @@ -324,7 +324,7 @@ func containsCoveringIndexScan(plan string) bool { } // containsIndexScan returns true if any plan line shows an index-ordered scan -// on the metrics table (e.g., "SCAN metrics USING INDEX idx_metrics_unique"). +// on the metrics table (e.g., "SCAN metrics USING INDEX idx_metrics_lookup"). // This differs from a covering-index scan in that it reads table rows via the // index, but still avoids a full table scan. func containsIndexScan(plan string) bool { @@ -370,7 +370,7 @@ func newPlanTestDB(t *testing.T) *sql.DB { tier TEXT NOT NULL DEFAULT 'raw' ); - CREATE INDEX IF NOT EXISTS idx_metrics_lookup + CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_lookup ON metrics(resource_type, resource_id, metric_type, tier, timestamp); CREATE INDEX IF NOT EXISTS idx_metrics_tier_time @@ -379,9 +379,6 @@ func newPlanTestDB(t *testing.T) *sql.DB { CREATE INDEX IF NOT EXISTS idx_metrics_query_all ON metrics(resource_type, resource_id, tier, timestamp, metric_type); - CREATE UNIQUE INDEX IF NOT EXISTS idx_metrics_unique - ON metrics(resource_type, resource_id, metric_type, timestamp, tier); - CREATE TABLE IF NOT EXISTS metrics_meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL