mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 12:42:49 +00:00
fix(monitoring): drop replayed unified metric samples
Read-side registry rebuilds re-run the unified metric sync as often as every two seconds, so one poll observation was re-written many times between polls. The metrics store upserts on (resource, metric, tier, timestamp), so each replay UPDATEs the same row and commits a fresh transaction for no new data, churning the SQLite WAL and driving the disk-write amplification reported in #1966. Track the last sample handed to the store per series and skip exact timestamp+value repeats before the batch is enqueued. A corrected value at the same observation time is still written, so the guard cannot mask a real change. Record the behaviour in the monitoring and agent-lifecycle contracts and move the focused regression into the accepted monitor.go proof file. Change-source: pulse-maintainer
This commit is contained in:
@@ -8071,3 +8071,12 @@ This does not change token scopes, agent admission, or existing agent cleanup.
|
||||
Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and
|
||||
`TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`,
|
||||
`TestInitializeSystemSettingsReadError`.
|
||||
|
||||
### Unified metric replay guard does not change agent admission
|
||||
|
||||
The unified metric sync runs on host-agent ingest boundaries and on read-side
|
||||
registry rebuilds. A per-series replay guard now drops exact timestamp+value
|
||||
repeats before the metrics batch is enqueued, reducing WAL churn without
|
||||
changing host admission, token binding or removal-block behaviour. Focused proof
|
||||
lives in `internal/monitoring/monitor_host_agents_test.go`
|
||||
(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
|
||||
|
||||
@@ -4065,3 +4065,18 @@ provider keys, canonical IDs, JSON projection, temperature/size/cadence, separat
|
||||
nodes/controller members, and confirmed inventory removal.
|
||||
`TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
|
||||
This is synthetic runtime evidence, not USB hardware or reporter acceptance.
|
||||
|
||||
### Unified metric replay guard bounds registry-rebuild writes
|
||||
|
||||
Read-side registry rebuilds (/api/state, websocket hydrate) re-run the unified
|
||||
metric sync as often as every two seconds, and storage observations are anchored
|
||||
to the provider's LastSeen. Before this change each rebuild re-issued the same
|
||||
rows as SQLite UPDATEs and committed a fresh transaction for no new data,
|
||||
churning the WAL and driving the disk-write amplification in #1966. Monitoring
|
||||
now tracks the last sample handed to the store per (resource type, resource id,
|
||||
metric type) series and drops exact timestamp+value repeats before the batch is
|
||||
enqueued; a corrected value at the same observation time is still written, so a
|
||||
real change is never masked. The guard is bounded so a long-lived monitor with
|
||||
churning resource IDs cannot pin memory. Focused proof lives in
|
||||
`internal/monitoring/monitor_host_agents_test.go`
|
||||
(`TestDedupeUnifiedMetricWritesDropsExactReplays`).
|
||||
|
||||
@@ -1247,6 +1247,14 @@ type Monitor struct {
|
||||
// that already hold m.mu can consult it without lock re-entrancy.
|
||||
licenseCheckerMu sync.RWMutex
|
||||
licenseChecker func(feature string) bool
|
||||
|
||||
// Unified metric sync replay guard. The sync runs on both ingest
|
||||
// boundaries and read-side registry rebuilds, so one observation is often
|
||||
// replayed many times between polls. Without this every replay re-writes
|
||||
// the same metrics row (an UPDATE in SQLite) and churns the WAL for no new
|
||||
// data, which is the dominant write amplification in #1966.
|
||||
unifiedMetricSyncMu sync.Mutex
|
||||
unifiedMetricSyncLast map[unifiedMetricSampleKey]unifiedMetricSample
|
||||
}
|
||||
|
||||
func (m *Monitor) setRuntimeContext(ctx context.Context, hub *websocket.Hub) {
|
||||
@@ -5171,6 +5179,63 @@ func recordSupplementalResourceChanges(store ResourceStoreInterface, changes []u
|
||||
}
|
||||
}
|
||||
|
||||
// unifiedMetricSampleKey identifies one persisted unified-metric series. The
|
||||
// metrics store keys rows by (resource type, resource id, metric type, tier,
|
||||
// timestamp), so the replay guard must match at least those coordinates.
|
||||
type unifiedMetricSampleKey struct {
|
||||
resourceType string
|
||||
resourceID string
|
||||
metricType string
|
||||
}
|
||||
|
||||
// unifiedMetricSample is the last sample handed to the metrics store for a
|
||||
// series. Timestamps are compared at the store's second resolution so a
|
||||
// sub-second replay of the same observation is still recognised as redundant.
|
||||
type unifiedMetricSample struct {
|
||||
timestampUnix int64
|
||||
value float64
|
||||
}
|
||||
|
||||
// unifiedMetricSyncMaxSeries bounds the replay guard so long-lived monitors
|
||||
// with churning resource IDs cannot pin memory. The map is reset wholesale
|
||||
// when the cap is crossed; the worst case is one extra write per series.
|
||||
const unifiedMetricSyncMaxSeries = 100000
|
||||
|
||||
// dedupeUnifiedMetricWrites drops exact repeats of a sample already handed to
|
||||
// the metrics store. The unified sync runs on ingest boundaries and on
|
||||
// read-side registry rebuilds (which fire as often as every two seconds), so
|
||||
// the same observation is replayed many times between polls. Each replay would
|
||||
// otherwise UPDATE the same row and commit a fresh transaction, churning the
|
||||
// SQLite WAL with no new data (#1966). A changed value at the same timestamp is
|
||||
// still written so corrections are not lost.
|
||||
func (m *Monitor) dedupeUnifiedMetricWrites(writes []metrics.WriteMetric) []metrics.WriteMetric {
|
||||
if len(writes) == 0 {
|
||||
return writes
|
||||
}
|
||||
|
||||
m.unifiedMetricSyncMu.Lock()
|
||||
defer m.unifiedMetricSyncMu.Unlock()
|
||||
if m.unifiedMetricSyncLast == nil || len(m.unifiedMetricSyncLast) > unifiedMetricSyncMaxSeries {
|
||||
m.unifiedMetricSyncLast = make(map[unifiedMetricSampleKey]unifiedMetricSample, len(writes))
|
||||
}
|
||||
|
||||
out := writes[:0]
|
||||
for _, write := range writes {
|
||||
key := unifiedMetricSampleKey{
|
||||
resourceType: write.ResourceType,
|
||||
resourceID: write.ResourceID,
|
||||
metricType: write.MetricType,
|
||||
}
|
||||
sample := unifiedMetricSample{timestampUnix: write.Timestamp.Unix(), value: write.Value}
|
||||
if previous, ok := m.unifiedMetricSyncLast[key]; ok && previous == sample {
|
||||
continue
|
||||
}
|
||||
m.unifiedMetricSyncLast[key] = sample
|
||||
out = append(out, write)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
|
||||
if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
|
||||
return
|
||||
@@ -5276,6 +5341,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
|
||||
appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
|
||||
}
|
||||
}
|
||||
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||
if len(storeWrites) > 0 {
|
||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||
}
|
||||
@@ -5391,6 +5457,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
|
||||
appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
|
||||
}
|
||||
}
|
||||
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||
if len(storeWrites) > 0 {
|
||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||
}
|
||||
@@ -5490,6 +5557,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
|
||||
appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
|
||||
}
|
||||
}
|
||||
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||
if len(storeWrites) > 0 {
|
||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||
}
|
||||
@@ -5697,6 +5765,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
|
||||
appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
|
||||
}
|
||||
}
|
||||
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||
if len(storeWrites) > 0 {
|
||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||
}
|
||||
|
||||
@@ -5884,3 +5884,48 @@ func TestAgentReportsPreserveExplicitDiskIncludes(t *testing.T) {
|
||||
check(t, host.Disks)
|
||||
})
|
||||
}
|
||||
|
||||
// Issue #1966: read-side registry rebuilds re-run the unified metric sync as
|
||||
// often as every two seconds, so one poll observation was re-written many
|
||||
// times between polls. The store upserts on (resource, metric, tier,
|
||||
// timestamp), so each replay committed a fresh transaction for no new data and
|
||||
// churned the SQLite WAL. The replay guard drops exact timestamp+value repeats
|
||||
// before the batch is enqueued while still writing a corrected value.
|
||||
func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) {
|
||||
monitor := &Monitor{}
|
||||
observedAt := time.Date(2026, time.September, 19, 12, 0, 0, 0, time.UTC)
|
||||
write := metrics.WriteMetric{
|
||||
ResourceType: "storage",
|
||||
ResourceID: "pool:tank",
|
||||
MetricType: "usage",
|
||||
Value: 62,
|
||||
Timestamp: observedAt,
|
||||
Tier: metrics.TierRaw,
|
||||
}
|
||||
|
||||
first := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
|
||||
if len(first) != 1 {
|
||||
t.Fatalf("first write count = %d, want 1", len(first))
|
||||
}
|
||||
// The same observation replayed by a read-side registry rebuild must not
|
||||
// reach the store again: that UPDATE is the write amplification in #1966.
|
||||
replay := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{write})
|
||||
if len(replay) != 0 {
|
||||
t.Fatalf("replayed write count = %d, want 0", len(replay))
|
||||
}
|
||||
|
||||
// A corrected value at the same observation time is still written.
|
||||
corrected := write
|
||||
corrected.Value = 63
|
||||
if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{corrected}); len(got) != 1 {
|
||||
t.Fatalf("corrected write count = %d, want 1", len(got))
|
||||
}
|
||||
|
||||
// A newer observation for the same series is written.
|
||||
advanced := write
|
||||
advanced.Value = 63
|
||||
advanced.Timestamp = observedAt.Add(time.Minute)
|
||||
if got := monitor.dedupeUnifiedMetricWrites([]metrics.WriteMetric{advanced}); len(got) != 1 {
|
||||
t.Fatalf("advanced write count = %d, want 1", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user