From 0e7ec69c512dd98d06265a5f3ff24449bf513d16 Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:31:51 +0100 Subject: [PATCH] fix(monitoring): batch unified metric sync writes into one transaction The five unified syncs ran serially and each waited for its own SQLite commit, so a registry rebuild that carried new data for several resource types opened one transaction per type and re-committed the same WAL. Collect the agent, VM, storage and app-container writes through a batch sink and commit them once; the per-series replay guard still runs first, so the series, values and source observation times are unchanged. Standalone callers keep the immediate write. Focused proof in monitor_host_agents_test.go; contract sections recorded in monitoring.md and agent-lifecycle.md. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 4 + .../v6/internal/subsystems/monitoring.md | 16 ++++ internal/monitoring/monitor.go | 76 ++++++++++++------ .../monitoring/monitor_host_agents_test.go | 78 +++++++++++++++++++ 4 files changed, 151 insertions(+), 23 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index 686ef7194c..820656a4cb 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8094,6 +8094,10 @@ sampling does not alter agent admission either. Focused proof lives in (`TestDedupeUnifiedMetricWritesDropsExactReplays`, `TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`, `TestSyncUnifiedPhysicalDiskMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`). +The batched sync path (`syncAllUnifiedMetrics`) collects the agent, VM, storage +and app-container writes into one SQLite transaction without changing the +series, values or observation times, so agent admission, token binding and +removal-block behaviour are unchanged. ### Docker update preflight accepts any local RepoDigest for the planned image diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 398c56a1d0..6f4255c924 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4116,6 +4116,22 @@ and `internal/monitoring/monitor_polling_test.go` (`TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`, `TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`). +### Unified metric sync commits one batch per pass + +The five unified syncs run serially and each previously waited for its own +SQLite commit, so one registry rebuild that carried new data for several +resource types opened one transaction per type and committed the same WAL +repeatedly. `syncAllUnifiedMetrics` now collects the surviving writes of the +agent, VM, storage and app-container syncs through a batch sink and commits them +in a single `WriteBatchBounded` transaction. Each sync still runs its own +per-series replay guard first, so the change only moves the transaction +boundary: the same series, values and source observation times are written. +Standalone callers and tests still pass no sink and keep the immediate, +read-your-writes write. The physical-disk SMART path keeps its own transaction +because it writes directly. Focused proof lives in +`internal/monitoring/monitor_polling_test.go` +(`TestSyncUnifiedStorageMetricsDefersWritesToBatchSink`). + ### Docker image update comparison accepts every local RepoDigest Docker records every RepoDigest an image is known by, and the digest for a diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index ef9e620dee..358f66f909 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -5097,11 +5097,7 @@ func (m *Monitor) updateResourceStoreForRead(state models.StateSnapshot) { return } recordSupplementalResourceChanges(store, m.collectSupplementalChanges()) - m.syncUnifiedAgentMetrics(store) - m.syncUnifiedVMMetrics(store) - m.syncUnifiedStorageMetrics(store) - m.syncUnifiedPhysicalDiskMetrics(store) - m.syncUnifiedAppContainerMetrics(store) + m.syncAllUnifiedMetrics(store) m.syncUnifiedResourceAlertsToState(store.GetAll()) } @@ -5148,11 +5144,7 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) { if atomicStore, ok := store.(AtomicSnapshotResourceStore); ok { atomicStore.PopulateSnapshotAndSupplemental(snapshotForStore, recordsBySource) recordSupplementalResourceChanges(store, supplementalChanges) - m.syncUnifiedAgentMetrics(store) - m.syncUnifiedVMMetrics(store) - m.syncUnifiedStorageMetrics(store) - m.syncUnifiedPhysicalDiskMetrics(store) - m.syncUnifiedAppContainerMetrics(store) + m.syncAllUnifiedMetrics(store) for source, records := range recordsBySource { if len(records) == 0 { continue @@ -5183,11 +5175,7 @@ func (m *Monitor) updateResourceStore(state models.StateSnapshot) { } recordSupplementalResourceChanges(store, supplementalChanges) - m.syncUnifiedAgentMetrics(store) - m.syncUnifiedVMMetrics(store) - m.syncUnifiedStorageMetrics(store) - m.syncUnifiedPhysicalDiskMetrics(store) - m.syncUnifiedAppContainerMetrics(store) + m.syncAllUnifiedMetrics(store) m.syncUnifiedResourceAlertsToState(store.GetAll()) } @@ -5282,7 +5270,41 @@ func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metr return out } -func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { +// syncAllUnifiedMetrics runs the per-resource-type unified syncs and commits +// the surviving store writes of the batched syncs as one transaction. Each sync +// still runs the per-series replay guard, so this only changes the transaction +// boundary, not the series written, their values or their observation times. +// The syncs run serially and each previously waited for its own commit, so a +// rebuild that carried new data for several resource types opened one SQLite +// transaction per type; batching collapses that to one (#1966). The physical +// disk sync keeps its own transaction because its SMART path writes directly. +func (m *Monitor) syncAllUnifiedMetrics(store ResourceStoreInterface) { + if m == nil { + return + } + if m.metricsStore == nil { + // The in-memory history path still needs to run; there is no store + // batch to collect. + m.syncUnifiedAgentMetrics(store) + m.syncUnifiedVMMetrics(store) + m.syncUnifiedStorageMetrics(store) + m.syncUnifiedPhysicalDiskMetrics(store) + m.syncUnifiedAppContainerMetrics(store) + return + } + var batch []metrics.WriteMetric + sink := &batch + m.syncUnifiedAgentMetrics(store, sink) + m.syncUnifiedVMMetrics(store, sink) + m.syncUnifiedStorageMetrics(store, sink) + m.syncUnifiedAppContainerMetrics(store, sink) + m.syncUnifiedPhysicalDiskMetrics(store) + if len(batch) > 0 { + m.metricsStore.WriteBatchBounded(batch) + } +} + +func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) { if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) { return } @@ -5389,12 +5411,14 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) - if len(storeWrites) > 0 { + if len(sinks) > 0 && sinks[0] != nil { + *sinks[0] = append(*sinks[0], storeWrites...) + } else if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } } -func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { +func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) { if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) { return } @@ -5509,12 +5533,14 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) - if len(storeWrites) > 0 { + if len(sinks) > 0 && sinks[0] != nil { + *sinks[0] = append(*sinks[0], storeWrites...) + } else if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } } -func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) { +func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) { if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) { return } @@ -5609,7 +5635,9 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) { } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) - if len(storeWrites) > 0 { + if len(sinks) > 0 && sinks[0] != nil { + *sinks[0] = append(*sinks[0], storeWrites...) + } else if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } } @@ -5731,7 +5759,7 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) { } } -func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { +func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface, sinks ...*[]metrics.WriteMetric) { if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) { return } @@ -5843,7 +5871,9 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) - if len(storeWrites) > 0 { + if len(sinks) > 0 && sinks[0] != nil { + *sinks[0] = append(*sinks[0], storeWrites...) + } else if len(storeWrites) > 0 { m.metricsStore.WriteBatchBounded(storeWrites) } } diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index eadeb0af41..94e157105d 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -5931,6 +5931,84 @@ func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) { } } +func TestSyncUnifiedStorageMetricsDefersWritesToBatchSink(t *testing.T) { + observedAt := time.Now().UTC().Add(-time.Minute).Truncate(time.Second) + resourceStore := unifiedresources.NewMonitorAdapter(nil) + resourceStore.PopulateFromSnapshot(models.StateSnapshot{ + PBSInstances: []models.PBSInstance{{ + ID: "pbs-main", + Name: "pbs-main", + Status: "online", + LastSeen: observedAt, + Datastores: []models.PBSDatastore{{ + Name: "backups", + Status: "available", + Total: 1000, + Used: 400, + Free: 600, + Usage: 40, + }}, + }}, + }) + + var targetID string + for _, resource := range resourceStore.GetAll() { + if resource.Type != unifiedresources.ResourceTypeStorage || resource.Storage == nil || resource.Storage.Platform != "pbs" { + continue + } + target := resourceStore.MetricsTargetForResource(resource.ID) + if target == nil || target.ResourceType != "storage" { + t.Fatalf("unexpected PBS datastore metrics target: %+v", target) + } + targetID = target.ResourceID + break + } + if targetID == "" { + t.Fatal("expected a PBS datastore in the unified resource store") + } + + cfg := metrics.DefaultConfig(t.TempDir()) + persistentStore, err := metrics.NewStore(cfg) + if err != nil { + t.Fatalf("metrics.NewStore() error = %v", err) + } + defer func() { _ = persistentStore.Close() }() + + monitor := &Monitor{ + metricsHistory: NewMetricsHistory(1024, 24*time.Hour), + metricsStore: persistentStore, + } + + // A sink collects the batch so the unified syncs can share one transaction. + // The sink path must not write through to the store itself. + var sink []metrics.WriteMetric + monitor.syncUnifiedStorageMetrics(resourceStore, &sink) + if len(sink) == 0 { + t.Fatal("expected storage writes to be collected in the batch sink") + } + + before, err := persistentStore.Query("storage", targetID, "usage", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted PBS usage: %v", err) + } + if len(before) != 0 { + t.Fatalf("sink path must not write to the store directly, got %+v", before) + } + + persistentStore.WriteBatchBounded(sink) + + after, err := persistentStore.Query("storage", targetID, "usage", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted PBS usage: %v", err) + } + if len(after) != 1 { + t.Fatalf("persisted PBS usage points = %d, want one after the batch flush: %+v", len(after), after) + } + if !after[0].Timestamp.Equal(observedAt) { + t.Fatalf("persisted PBS usage timestamp = %s, want source observation %s", after[0].Timestamp, observedAt) + } +} + func TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) { previous := truenas.IsFeatureEnabled() truenas.SetFeatureEnabled(true)