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
|
Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and
|
||||||
`TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`,
|
`TestInitializeSystemSettingsPreservesExistingBytes`, `TestInitializeSystemSettingsMissing`,
|
||||||
`TestInitializeSystemSettingsReadError`.
|
`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.
|
nodes/controller members, and confirmed inventory removal.
|
||||||
`TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
|
`TestPhysicalDiskReadbackSourceIDFallback` covers missing source metadata.
|
||||||
This is synthetic runtime evidence, not USB hardware or reporter acceptance.
|
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.
|
// that already hold m.mu can consult it without lock re-entrancy.
|
||||||
licenseCheckerMu sync.RWMutex
|
licenseCheckerMu sync.RWMutex
|
||||||
licenseChecker func(feature string) bool
|
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) {
|
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) {
|
func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
|
||||||
if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
|
if store == nil || (m.metricsHistory == nil && m.metricsStore == nil) {
|
||||||
return
|
return
|
||||||
@@ -5276,6 +5341,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) {
|
|||||||
appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
|
appendStoreWrite("agent", targetID, "diskwrite", metric.Value)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||||
if len(storeWrites) > 0 {
|
if len(storeWrites) > 0 {
|
||||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||||
}
|
}
|
||||||
@@ -5391,6 +5457,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) {
|
|||||||
appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
|
appendStoreWrite("vm", targetID, "diskwrite", metric.Value)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||||
if len(storeWrites) > 0 {
|
if len(storeWrites) > 0 {
|
||||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||||
}
|
}
|
||||||
@@ -5490,6 +5557,7 @@ func (m *Monitor) syncUnifiedStorageMetrics(store ResourceStoreInterface) {
|
|||||||
appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
|
appendStoreWrite("storage", targetID, "avail", float64(free), observedAt)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||||
if len(storeWrites) > 0 {
|
if len(storeWrites) > 0 {
|
||||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||||
}
|
}
|
||||||
@@ -5697,6 +5765,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) {
|
|||||||
appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
|
appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
storeWrites = m.dedupeUnifiedMetricWrites(storeWrites)
|
||||||
if len(storeWrites) > 0 {
|
if len(storeWrites) > 0 {
|
||||||
m.metricsStore.WriteBatchBounded(storeWrites)
|
m.metricsStore.WriteBatchBounded(storeWrites)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -5884,3 +5884,48 @@ func TestAgentReportsPreserveExplicitDiskIncludes(t *testing.T) {
|
|||||||
check(t, host.Disks)
|
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