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 afd715022..1ee3bf824 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -1229,7 +1229,14 @@ aggregate-only min/max columns stay unset. Rollups must not be hard-coded to a 5-minute disk-write cadence: the default cadence should favor set-based transactions, remain bounded by raw retention so data is aggregated before pruning, and stay overrideable for operators who deliberately trade freshness -for lower write frequency. Rollup transactions must also keep a bounded +for lower write frequency. Startup cleanup, rollup, and retention must run on a +lifecycle-owned maintenance worker separate from the worker draining the +bounded live-ingestion queue. SQLite remains the serialization boundary for +their transactions, but maintenance CPU or read work must not stop buffered +writes or `Flush` barriers from progressing; shutdown must join both workers +before closing their shared database. `pkg/metrics/store_additional_test.go` +must prove ingestion remains live while startup maintenance is blocked. Rollup +transactions must also keep a bounded time-window working set and resume from checkpoints across invocations instead of aggregating every pending series and bucket in one global SQLite GROUP BY that can spike RSS on agent-heavy installs. The runtime must also allow an explicit diff --git a/internal/hostagent/smartctl_standby_guard_test.go b/internal/hostagent/smartctl_standby_guard_test.go index 4c1f7162c..b9e7102db 100644 --- a/internal/hostagent/smartctl_standby_guard_test.go +++ b/internal/hostagent/smartctl_standby_guard_test.go @@ -6,6 +6,7 @@ import ( "fmt" "io/fs" "os/exec" + "path/filepath" "slices" "strings" "testing" @@ -193,7 +194,8 @@ func TestIssue1653DSMDeviceGetsSATRetry(t *testing.T) { } func TestIssue1653SmartctlPathOverride(t *testing.T) { - t.Setenv("PULSE_SMARTCTL_PATH", "/opt/syno/bin/smartctl") + configuredPath := filepath.Join(t.TempDir(), "smartctl") + t.Setenv("PULSE_SMARTCTL_PATH", configuredPath) originalLookPath := execLookPath t.Cleanup(func() { execLookPath = originalLookPath }) execLookPath = func(string) (string, error) { @@ -204,7 +206,7 @@ func TestIssue1653SmartctlPathOverride(t *testing.T) { if err != nil { t.Fatalf("resolveSmartctlPath() error = %v", err) } - if path != "/opt/syno/bin/smartctl" { + if path != configuredPath { t.Fatalf("resolveSmartctlPath() = %q", path) } } diff --git a/pkg/metrics/issue1601_read_concurrency_test.go b/pkg/metrics/issue1601_read_concurrency_test.go deleted file mode 100644 index f50e71ba3..000000000 --- a/pkg/metrics/issue1601_read_concurrency_test.go +++ /dev/null @@ -1,123 +0,0 @@ -package metrics - -import ( - "sync" - "testing" - "time" -) - -// A connection pool of one queued every UI history read behind metric-flush -// commits, freezing charts whenever a commit picked up a WAL checkpoint -// (#1601). Reads must proceed on their own connections while a write -// transaction holds the WAL write lock. -func TestQueriesDoNotQueueBehindWriteTransaction(t *testing.T) { - store, err := NewStore(DefaultConfig(t.TempDir())) - if err != nil { - t.Fatalf("NewStore: %v", err) - } - defer store.Close() - - // Let startup maintenance (checkpoint/vacuum) finish so its locks don't - // entangle with the write transaction opened below. - if err := store.WaitForMaintenance(10 * time.Second); err != nil { - t.Fatalf("WaitForMaintenance: %v", err) - } - - if got := store.db.DB.Stats().MaxOpenConnections; got < 2 { - t.Fatalf("MaxOpenConnections = %d, want at least 2 so reads do not serialize behind writes", got) - } - - // Hold the WAL write lock on one connection. - tx, err := store.db.Begin() - if err != nil { - t.Fatalf("Begin: %v", err) - } - defer func() { _ = tx.Rollback() }() - if _, err := tx.Exec( - `INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) - VALUES ('node', 'n1', 'cpu', 1.0, ?, 'raw')`, - time.Now().Unix(), - ); err != nil { - t.Fatalf("write inside transaction: %v", err) - } - - done := make(chan error, 1) - go func() { - _, err := store.Query("node", "n1", "cpu", time.Now().Add(-time.Hour), time.Now(), 60) - done <- err - }() - - select { - case err := <-done: - if err != nil { - t.Fatalf("Query during open write transaction: %v", err) - } - case <-time.After(5 * time.Second): - t.Fatal("Query blocked behind the open write transaction") - } -} - -// Startup maintenance used to execute on the same worker that drained -// writeCh. A large retention/VACUUM pass could therefore fill the bounded -// write queue and drop live samples. Keep maintenance blocked here and prove -// that ingestion and its Flush barrier remain operational. -func TestStartupMaintenanceDoesNotBlockIngestionWorker(t *testing.T) { - maintenanceEntered := make(chan struct{}) - releaseMaintenance := make(chan struct{}) - var releaseOnce sync.Once - release := func() { - releaseOnce.Do(func() { close(releaseMaintenance) }) - } - t.Cleanup(release) - - previousHook := startupMaintenanceHook - startupMaintenanceHook = func() { - close(maintenanceEntered) - <-releaseMaintenance - } - t.Cleanup(func() { startupMaintenanceHook = previousHook }) - - config := DefaultConfig(t.TempDir()) - config.WriteBufferSize = 1 - config.FlushInterval = time.Hour - - store, err := NewStore(config) - if err != nil { - t.Fatalf("NewStore: %v", err) - } - defer store.Close() - - select { - case <-maintenanceEntered: - case <-time.After(5 * time.Second): - t.Fatal("startup maintenance did not begin") - } - - timestamp := time.Now().UTC().Truncate(time.Second) - store.Write("node", "n1", "cpu", 42, timestamp) - - flushed := make(chan struct{}) - go func() { - store.Flush() - close(flushed) - }() - - select { - case <-flushed: - case <-time.After(5 * time.Second): - t.Fatal("metric ingestion blocked behind startup maintenance") - } - - release() - if err := store.WaitForMaintenance(10 * time.Second); err != nil { - t.Fatalf("WaitForMaintenance: %v", err) - } - - points, err := store.Query("node", "n1", "cpu", timestamp.Add(-time.Second), timestamp.Add(time.Second), 0) - if err != nil { - t.Fatalf("Query: %v", err) - } - if len(points) != 1 || points[0].Value != 42 { - t.Fatalf("persisted points = %+v, want one value 42", points) - } -} diff --git a/pkg/metrics/store_additional_test.go b/pkg/metrics/store_additional_test.go index dfaf52e3f..dc63c9a8b 100644 --- a/pkg/metrics/store_additional_test.go +++ b/pkg/metrics/store_additional_test.go @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "strings" + "sync" "syscall" "testing" "time" @@ -524,6 +525,119 @@ func TestStoreWaitForMaintenanceWaitsForQueuedStartupWork(t *testing.T) { } } +// A connection pool of one queued every UI history read behind metric-flush +// commits, freezing charts whenever a commit picked up a WAL checkpoint +// (#1601). Reads must proceed on their own connections while a write +// transaction holds the WAL write lock. +func TestQueriesDoNotQueueBehindWriteTransaction(t *testing.T) { + store, err := NewStore(DefaultConfig(t.TempDir())) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + defer store.Close() + + if err := store.WaitForMaintenance(10 * time.Second); err != nil { + t.Fatalf("WaitForMaintenance: %v", err) + } + + if got := store.db.DB.Stats().MaxOpenConnections; got < 2 { + t.Fatalf("MaxOpenConnections = %d, want at least 2 so reads do not serialize behind writes", got) + } + + tx, err := store.db.Begin() + if err != nil { + t.Fatalf("Begin: %v", err) + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.Exec( + `INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) + VALUES ('node', 'n1', 'cpu', 1.0, ?, 'raw')`, + time.Now().Unix(), + ); err != nil { + t.Fatalf("write inside transaction: %v", err) + } + + done := make(chan error, 1) + go func() { + _, err := store.Query("node", "n1", "cpu", time.Now().Add(-time.Hour), time.Now(), 60) + done <- err + }() + + select { + case err := <-done: + if err != nil { + t.Fatalf("Query during open write transaction: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Query blocked behind the open write transaction") + } +} + +// Startup maintenance used to execute on the same worker that drained +// writeCh. A large retention/VACUUM pass could therefore fill the bounded +// write queue and drop live samples. Keep maintenance blocked here and prove +// that ingestion and its Flush barrier remain operational. +func TestStartupMaintenanceDoesNotBlockIngestionWorker(t *testing.T) { + maintenanceEntered := make(chan struct{}) + releaseMaintenance := make(chan struct{}) + var releaseOnce sync.Once + release := func() { + releaseOnce.Do(func() { close(releaseMaintenance) }) + } + t.Cleanup(release) + + previousHook := startupMaintenanceHook + startupMaintenanceHook = func() { + close(maintenanceEntered) + <-releaseMaintenance + } + t.Cleanup(func() { startupMaintenanceHook = previousHook }) + + config := DefaultConfig(t.TempDir()) + config.WriteBufferSize = 1 + config.FlushInterval = time.Hour + + store, err := NewStore(config) + if err != nil { + t.Fatalf("NewStore: %v", err) + } + defer store.Close() + + select { + case <-maintenanceEntered: + case <-time.After(5 * time.Second): + t.Fatal("startup maintenance did not begin") + } + + timestamp := time.Now().UTC().Truncate(time.Second) + store.Write("node", "n1", "cpu", 42, timestamp) + + flushed := make(chan struct{}) + go func() { + store.Flush() + close(flushed) + }() + + select { + case <-flushed: + case <-time.After(5 * time.Second): + t.Fatal("metric ingestion blocked behind startup maintenance") + } + + release() + if err := store.WaitForMaintenance(10 * time.Second); err != nil { + t.Fatalf("WaitForMaintenance: %v", err) + } + + points, err := store.Query("node", "n1", "cpu", timestamp.Add(-time.Second), timestamp.Add(time.Second), 0) + if err != nil { + t.Fatalf("Query: %v", err) + } + if len(points) != 1 || points[0].Value != 42 { + t.Fatalf("persisted points = %+v, want one value 42", points) + } +} + func TestStoreMigratesLegacyHostResourceTypeToAgent(t *testing.T) { dir := t.TempDir() cfg := DefaultConfig(dir)