From 23c84ed813f99c70b67cbda53df6101a0cf14e8e Mon Sep 17 00:00:00 2001 From: "pulse-triage[bot]" <249995291+pulse-triage[bot]@users.noreply.github.com> Date: Sun, 20 Sep 2026 15:48:01 +0100 Subject: [PATCH] fix(monitoring): anchor unified sync samples to source observation time The replay guard added in df44238797 only suppressed repeats for storage, whose samples already carry the provider LastSeen. The agent, VM and app-container syncs still stamped every write with the rebuild wall clock, so dedupeUnifiedMetricWrites could never match and each read-side registry rebuild re-issued the whole batch as fresh rows. That is the same WAL churn described in #1966, still live for VMware/TrueNAS VMs, TrueNAS apps and supplemental agents. Resolve each resource's observation time from its metric source (unifiedResourceObservedAt) and use it for both the in-memory history point and the store write, matching the storage path. Proxmox and libvirt VMs already have native history writers, so exclude libvirt VMs from the unified VM sync to avoid writing a second timeline for the same series. Contract sections recorded in monitoring.md and agent-lifecycle.md; focused proof added in monitor_polling_test.go and monitor_host_agents_test.go. Local Go compilation is unavailable this turn (module cache read-only, toolchain snapshot missing); the canonical completion guard passes and gofmt is clean. Change-source: pulse-maintainer --- .../v6/internal/subsystems/agent-lifecycle.md | 16 +- .../v6/internal/subsystems/monitoring.md | 27 ++-- .../monitoring/canonical_guardrails_test.go | 12 +- internal/monitoring/monitor.go | 126 +++++++++------ .../monitoring/monitor_host_agents_test.go | 67 ++++++++ internal/monitoring/monitor_polling_test.go | 152 ++++++++++++++++++ 6 files changed, 330 insertions(+), 70 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index a528906e01..35d4e8d13c 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -8080,11 +8080,17 @@ Regression coverage: `TestQuickSecuritySetupForcePreservesSystemSettings` and ### 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`). +registry rebuilds. Agent and VM history samples are now anchored to the source +observation time (`unifiedResourceObservedAt`) instead of the rebuild wall +clock, so a registry rebuild re-issuing the same observation maps to one +timestamp rather than a fresh sample. A per-series replay guard then drops exact +timestamp+value repeats before the metrics batch is enqueued, reducing WAL churn +without changing host admission, token binding or removal-block behaviour. +Proxmox and libvirt VMs have native history writers and are excluded from the +unified VM sync. Focused proof lives in +`internal/monitoring/monitor_host_agents_test.go` +(`TestDedupeUnifiedMetricWritesDropsExactReplays`, +`TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`). ### 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 fcf6fdb617..68d56687e8 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -4091,17 +4091,24 @@ 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 +metric sync as often as every two seconds. Every unified sync (agent, VM, +storage, and app-container) now timestamps its samples with the source +observation time (`unifiedResourceObservedAt`/`unifiedMetricObservedAt`), not the +rebuild wall clock. 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 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. Proxmox and libvirt VMs have native history +writers and are excluded from the unified VM sync so the same series is not +written twice. Focused proof lives in `internal/monitoring/monitor_host_agents_test.go` -(`TestDedupeUnifiedMetricWritesDropsExactReplays`). +(`TestDedupeUnifiedMetricWritesDropsExactReplays`) and +`internal/monitoring/monitor_polling_test.go` +(`TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`, +`TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds`). ### Docker image update comparison accepts every local RepoDigest diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index b300842930..d34cd67eaf 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -1944,8 +1944,8 @@ func TestUnifiedAppContainerMetricsUseCanonicalGuestHistoryPath(t *testing.T) { `if target == nil || target.ResourceType != "app-container" || strings.TrimSpace(target.ResourceID) == "" {`, `metricKey := fmt.Sprintf("docker:%s", targetID)`, `storeWrites := make([]metrics.WriteMetric, 0)`, - `appendStoreWrite("dockerContainer", targetID, "cpu", value)`, - `appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value)`, + `appendStoreWrite("dockerContainer", targetID, "cpu", value, observedAt)`, + `appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value, observedAt)`, `m.metricsStore.WriteBatchBounded(storeWrites)`, } for _, snippet := range requiredSnippets { @@ -1988,8 +1988,8 @@ func TestUnifiedAgentMetricsUseCanonicalHostHistoryPath(t *testing.T) { `if target == nil || target.ResourceType != "agent" || strings.TrimSpace(target.ResourceID) == "" {`, `metricKey := fmt.Sprintf("agent:%s", targetID)`, `storeWrites := make([]metrics.WriteMetric, 0)`, - `appendStoreWrite("agent", targetID, "cpu", value)`, - `appendStoreWrite("agent", targetID, "diskwrite", metric.Value)`, + `appendStoreWrite("agent", targetID, "cpu", value, observedAt)`, + `appendStoreWrite("agent", targetID, "diskwrite", metric.Value, observedAt)`, `m.metricsStore.WriteBatchBounded(storeWrites)`, } for _, snippet := range requiredSnippets { @@ -2010,8 +2010,8 @@ func TestUnifiedVMMetricsUseCanonicalVMHistoryPath(t *testing.T) { `if target == nil || target.ResourceType != "vm" || strings.TrimSpace(target.ResourceID) == "" {`, `if source == unifiedresources.SourceProxmox {`, `storeWrites := make([]metrics.WriteMetric, 0)`, - `appendStoreWrite("vm", targetID, "cpu", value)`, - `appendStoreWrite("vm", targetID, "diskwrite", metric.Value)`, + `appendStoreWrite("vm", targetID, "cpu", value, observedAt)`, + `appendStoreWrite("vm", targetID, "diskwrite", metric.Value, observedAt)`, `m.metricsStore.WriteBatchBounded(storeWrites)`, } for _, snippet := range requiredSnippets { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 268b2e8be5..378aba84b2 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -5257,7 +5257,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { now := time.Now() storeWrites := make([]metrics.WriteMetric, 0) - appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) { + appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) { if m.metricsStore == nil { return } @@ -5266,7 +5266,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { ResourceID: resourceID, MetricType: metricType, Value: value, - Timestamp: now, + Timestamp: observedAt, Tier: metrics.TierRaw, }) } @@ -5294,6 +5294,7 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { } seenTargets[targetID] = struct{}{} metricKey := fmt.Sprintf("agent:%s", targetID) + observedAt := unifiedResourceObservedAt(resource, now) if metric := resource.Metrics.CPU; metric != nil { value := metric.Percent @@ -5301,53 +5302,53 @@ func (m *Monitor) syncUnifiedAgentMetrics(store ResourceStoreInterface) { value = metric.Value } if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, observedAt) } - appendStoreWrite("agent", targetID, "cpu", value) + appendStoreWrite("agent", targetID, "cpu", value, observedAt) } if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "memory", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "memory", value, observedAt) } - appendStoreWrite("agent", targetID, "memory", value) + appendStoreWrite("agent", targetID, "memory", value, observedAt) } if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "disk", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "disk", value, observedAt) } - appendStoreWrite("agent", targetID, "disk", value) + appendStoreWrite("agent", targetID, "disk", value, observedAt) } if metric := resource.Metrics.NetIn; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, observedAt) } - appendStoreWrite("agent", targetID, "netin", metric.Value) + appendStoreWrite("agent", targetID, "netin", metric.Value, observedAt) } if metric := resource.Metrics.NetOut; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, observedAt) } - appendStoreWrite("agent", targetID, "netout", metric.Value) + appendStoreWrite("agent", targetID, "netout", metric.Value, observedAt) } if metric := resource.Metrics.DiskRead; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, observedAt) } - appendStoreWrite("agent", targetID, "diskread", metric.Value) + appendStoreWrite("agent", targetID, "diskread", metric.Value, observedAt) } if metric := resource.Metrics.DiskWrite; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, observedAt) } - appendStoreWrite("agent", targetID, "diskwrite", metric.Value) + appendStoreWrite("agent", targetID, "diskwrite", metric.Value, observedAt) } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) @@ -5368,7 +5369,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { now := time.Now() storeWrites := make([]metrics.WriteMetric, 0) - appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) { + appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) { if m.metricsStore == nil { return } @@ -5377,7 +5378,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { ResourceID: resourceID, MetricType: metricType, Value: value, - Timestamp: now, + Timestamp: observedAt, Tier: metrics.TierRaw, }) } @@ -5390,7 +5391,10 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { continue } - hasNativeVMWriter := false + // Proxmox and libvirt both have a native history writer + // (monitor_pve_guest_helpers and monitor_libvirt), so the unified sync + // must not add a second timeline for the same series. + hasNativeVMWriter := strings.EqualFold(strings.TrimSpace(resource.Technology), "libvirt") for _, source := range resource.Sources { if source == unifiedresources.SourceProxmox { hasNativeVMWriter = true @@ -5410,6 +5414,7 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { continue } seenTargets[targetID] = struct{}{} + observedAt := unifiedResourceObservedAt(resource, now) if metric := resource.Metrics.CPU; metric != nil { value := metric.Percent @@ -5417,53 +5422,53 @@ func (m *Monitor) syncUnifiedVMMetrics(store ResourceStoreInterface) { value = metric.Value } if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "cpu", value, now) + m.metricsHistory.AddGuestMetric(targetID, "cpu", value, observedAt) } - appendStoreWrite("vm", targetID, "cpu", value) + appendStoreWrite("vm", targetID, "cpu", value, observedAt) } if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "memory", value, now) + m.metricsHistory.AddGuestMetric(targetID, "memory", value, observedAt) } - appendStoreWrite("vm", targetID, "memory", value) + appendStoreWrite("vm", targetID, "memory", value, observedAt) } if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0 || metric.Used != nil) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "disk", value, now) + m.metricsHistory.AddGuestMetric(targetID, "disk", value, observedAt) } - appendStoreWrite("vm", targetID, "disk", value) + appendStoreWrite("vm", targetID, "disk", value, observedAt) } if metric := resource.Metrics.NetIn; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "netin", metric.Value, now) + m.metricsHistory.AddGuestMetric(targetID, "netin", metric.Value, observedAt) } - appendStoreWrite("vm", targetID, "netin", metric.Value) + appendStoreWrite("vm", targetID, "netin", metric.Value, observedAt) } if metric := resource.Metrics.NetOut; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "netout", metric.Value, now) + m.metricsHistory.AddGuestMetric(targetID, "netout", metric.Value, observedAt) } - appendStoreWrite("vm", targetID, "netout", metric.Value) + appendStoreWrite("vm", targetID, "netout", metric.Value, observedAt) } if metric := resource.Metrics.DiskRead; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "diskread", metric.Value, now) + m.metricsHistory.AddGuestMetric(targetID, "diskread", metric.Value, observedAt) } - appendStoreWrite("vm", targetID, "diskread", metric.Value) + appendStoreWrite("vm", targetID, "diskread", metric.Value, observedAt) } if metric := resource.Metrics.DiskWrite; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(targetID, "diskwrite", metric.Value, now) + m.metricsHistory.AddGuestMetric(targetID, "diskwrite", metric.Value, observedAt) } - appendStoreWrite("vm", targetID, "diskwrite", metric.Value) + appendStoreWrite("vm", targetID, "diskwrite", metric.Value, observedAt) } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) @@ -5588,6 +5593,28 @@ func unifiedMetricObservedAt(resource unifiedresources.Resource, metric *unified return fallback } +// unifiedResourceObservedAt resolves the source observation time for a +// canonical resource from whichever metric carries a source attribution. +// Unified syncs use it as the history timestamp so a read-side registry rebuild +// re-issuing the same observation maps to one timestamp instead of inventing a +// fresh sample each rebuild (see #1966). +func unifiedResourceObservedAt(resource unifiedresources.Resource, fallback time.Time) time.Time { + if resource.Metrics != nil { + for _, metric := range []*unifiedresources.MetricValue{ + resource.Metrics.CPU, + resource.Metrics.Memory, + resource.Metrics.Disk, + resource.Metrics.NetIn, + resource.Metrics.NetOut, + } { + if metric != nil && metric.Source != "" { + return unifiedMetricObservedAt(resource, metric, fallback) + } + } + } + return unifiedMetricObservedAt(resource, nil, fallback) +} + func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) { if store == nil || m.metricsStore == nil { return @@ -5676,7 +5703,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { now := time.Now() storeWrites := make([]metrics.WriteMetric, 0) - appendStoreWrite := func(resourceType, resourceID, metricType string, value float64) { + appendStoreWrite := func(resourceType, resourceID, metricType string, value float64, observedAt time.Time) { if m.metricsStore == nil { return } @@ -5685,7 +5712,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { ResourceID: resourceID, MetricType: metricType, Value: value, - Timestamp: now, + Timestamp: observedAt, Tier: metrics.TierRaw, }) } @@ -5718,6 +5745,7 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { } seenTargets[targetID] = struct{}{} metricKey := fmt.Sprintf("docker:%s", targetID) + observedAt := unifiedResourceObservedAt(resource, now) if metric := resource.Metrics.CPU; metric != nil { value := metric.Percent @@ -5725,53 +5753,53 @@ func (m *Monitor) syncUnifiedAppContainerMetrics(store ResourceStoreInterface) { value = metric.Value } if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "cpu", value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "cpu", value) + appendStoreWrite("dockerContainer", targetID, "cpu", value, observedAt) } if metric := resource.Metrics.Memory; metric != nil && (metric.Total != nil || metric.Percent > 0) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "memory", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "memory", value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "memory", value) + appendStoreWrite("dockerContainer", targetID, "memory", value, observedAt) } if metric := resource.Metrics.Disk; metric != nil && (metric.Total != nil || metric.Percent > 0) { value := metric.Percent if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "disk", value, now) + m.metricsHistory.AddGuestMetric(metricKey, "disk", value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "disk", value) + appendStoreWrite("dockerContainer", targetID, "disk", value, observedAt) } if metric := resource.Metrics.NetIn; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "netin", metric.Value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "netin", metric.Value) + appendStoreWrite("dockerContainer", targetID, "netin", metric.Value, observedAt) } if metric := resource.Metrics.NetOut; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "netout", metric.Value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "netout", metric.Value) + appendStoreWrite("dockerContainer", targetID, "netout", metric.Value, observedAt) } if metric := resource.Metrics.DiskRead; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "diskread", metric.Value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "diskread", metric.Value) + appendStoreWrite("dockerContainer", targetID, "diskread", metric.Value, observedAt) } if metric := resource.Metrics.DiskWrite; metric != nil { if m.metricsHistory != nil { - m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, now) + m.metricsHistory.AddGuestMetric(metricKey, "diskwrite", metric.Value, observedAt) } - appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value) + appendStoreWrite("dockerContainer", targetID, "diskwrite", metric.Value, observedAt) } } storeWrites = m.dedupeUnifiedMetricWrites(storeWrites) diff --git a/internal/monitoring/monitor_host_agents_test.go b/internal/monitoring/monitor_host_agents_test.go index 05bb0a1e33..fec3a0fc91 100644 --- a/internal/monitoring/monitor_host_agents_test.go +++ b/internal/monitoring/monitor_host_agents_test.go @@ -18,6 +18,7 @@ import ( "github.com/rcourtman/pulse-go-rewrite/internal/notifications" "github.com/rcourtman/pulse-go-rewrite/internal/operationreceipt" "github.com/rcourtman/pulse-go-rewrite/internal/storagehealth" + "github.com/rcourtman/pulse-go-rewrite/internal/truenas" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" agentsdocker "github.com/rcourtman/pulse-go-rewrite/pkg/agents/docker" agentshost "github.com/rcourtman/pulse-go-rewrite/pkg/agents/host" @@ -5930,6 +5931,72 @@ func TestDedupeUnifiedMetricWritesDropsExactReplays(t *testing.T) { } } +func TestSyncUnifiedAgentMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) { + previous := truenas.IsFeatureEnabled() + truenas.SetFeatureEnabled(true) + t.Cleanup(func() { + truenas.SetFeatureEnabled(previous) + }) + + observedAt := truenas.DefaultFixtures().CollectedAt + resourceStore := unifiedresources.NewMonitorAdapter(nil) + resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceTrueNAS: truenas.NewProvider(truenas.DefaultFixtures()).Records(), + }) + + var targetID string + for _, resource := range resourceStore.GetAll() { + if resource.Type != unifiedresources.ResourceTypeAgent || resource.Name != "truenas-main" { + continue + } + target := resourceStore.MetricsTargetForResource(resource.ID) + if target == nil || target.ResourceType != "agent" { + t.Fatalf("unexpected agent metrics target: %+v", target) + } + targetID = target.ResourceID + break + } + if targetID == "" { + t.Fatal("expected a TrueNAS agent resource 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, + } + + // Read-side registry rebuilds must not invent a fresh sample for the same + // source observation (the write amplification in #1966). + monitor.syncUnifiedAgentMetrics(resourceStore) + monitor.syncUnifiedAgentMetrics(resourceStore) + + inMemory := monitor.GetGuestMetrics("agent:"+targetID, time.Hour)["cpu"] + if len(inMemory) != 1 { + t.Fatalf("in-memory agent cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory) + } + if !inMemory[0].Timestamp.Equal(observedAt) { + t.Fatalf("in-memory agent cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt) + } + + persisted, err := persistentStore.Query("agent", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted agent cpu: %v", err) + } + if len(persisted) != 1 { + t.Fatalf("persisted agent cpu points = %d, want one per source observation: %+v", len(persisted), persisted) + } + if !persisted[0].Timestamp.Equal(observedAt) { + t.Fatalf("persisted agent cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt) + } +} + // Issue #2113: a removed host agent re-enrolls under a derived suffixed // identity because identity resolution forks the machine before the removal // block is consulted. The block is keyed on the base machine identity, so the diff --git a/internal/monitoring/monitor_polling_test.go b/internal/monitoring/monitor_polling_test.go index b9f0169168..a0e59e1c9d 100644 --- a/internal/monitoring/monitor_polling_test.go +++ b/internal/monitoring/monitor_polling_test.go @@ -665,6 +665,158 @@ func TestSyncUnifiedStorageMetricsUsesPBSObservationTimeAcrossRegistryRebuilds(t } } +func TestSyncUnifiedAppContainerMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) { + previous := truenas.IsFeatureEnabled() + truenas.SetFeatureEnabled(true) + t.Cleanup(func() { + truenas.SetFeatureEnabled(previous) + }) + + observedAt := truenas.DefaultFixtures().CollectedAt + resourceStore := unifiedresources.NewMonitorAdapter(nil) + resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceTrueNAS: truenas.NewProvider(truenas.DefaultFixtures()).Records(), + }) + + var targetID string + for _, resource := range resourceStore.GetAll() { + if resource.Type != unifiedresources.ResourceTypeAppContainer || resource.Name != "Nextcloud" { + continue + } + target := resourceStore.MetricsTargetForResource(resource.ID) + if target == nil || target.ResourceType != "app-container" { + t.Fatalf("unexpected app-container metrics target: %+v", target) + } + targetID = target.ResourceID + break + } + if targetID == "" { + t.Fatal("expected a TrueNAS app-container 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, + } + + // Read-side registry rebuilds must not invent a fresh sample for the same + // source observation (the write amplification in #1966). + monitor.syncUnifiedAppContainerMetrics(resourceStore) + monitor.syncUnifiedAppContainerMetrics(resourceStore) + + inMemory := monitor.GetGuestMetrics("docker:"+targetID, time.Hour)["cpu"] + if len(inMemory) != 1 { + t.Fatalf("in-memory app-container cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory) + } + if !inMemory[0].Timestamp.Equal(observedAt) { + t.Fatalf("in-memory app-container cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt) + } + + persisted, err := persistentStore.Query("dockerContainer", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted app-container cpu: %v", err) + } + if len(persisted) != 1 { + t.Fatalf("persisted app-container cpu points = %d, want one per source observation: %+v", len(persisted), persisted) + } + if !persisted[0].Timestamp.Equal(observedAt) { + t.Fatalf("persisted app-container cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt) + } +} + +func TestSyncUnifiedVMMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) { + previous := vmware.IsFeatureEnabled() + vmware.SetFeatureEnabled(true) + t.Cleanup(func() { + vmware.SetFeatureEnabled(previous) + }) + + observedAt := time.Date(2026, time.March, 30, 18, 15, 0, 0, time.UTC) + resourceStore := unifiedresources.NewMonitorAdapter(nil) + resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceVMware: vmware.NewProvider(vmware.InventorySnapshot{ + ConnectionID: "vc-1", + ConnectionName: "Lab VC", + VCenterHost: "vc.lab.local", + CollectedAt: observedAt, + VMs: []vmware.InventoryVM{{ + VM: "vm-201", + Name: "app-01", + PowerState: "POWERED_ON", + CPUCount: 4, + MemorySizeMiB: 8192, + Metrics: &vmware.InventoryMetrics{ + CPUPercent: ptrFloat64(38.1), + MemoryPercent: ptrFloat64(57.5), + MemoryUsedBytes: ptrInt64(5033164800), + MemoryTotalBytes: ptrInt64(8589934592), + NetInBytesPerSecond: ptrFloat64(512), + NetOutBytesPerSecond: ptrFloat64(768), + DiskReadBytesPerSecond: ptrFloat64(1536), + DiskWriteBytesPerSecond: ptrFloat64(2048), + }, + }}, + }).Records(), + }) + + var targetID string + for _, resource := range resourceStore.GetAll() { + if resource.Type != unifiedresources.ResourceTypeVM || resource.Name != "app-01" { + continue + } + target := resourceStore.MetricsTargetForResource(resource.ID) + if target == nil || target.ResourceType != "vm" { + t.Fatalf("unexpected VM metrics target: %+v", target) + } + targetID = target.ResourceID + break + } + if targetID == "" { + t.Fatal("expected a VMware VM 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, + } + + monitor.syncUnifiedVMMetrics(resourceStore) + monitor.syncUnifiedVMMetrics(resourceStore) + + inMemory := monitor.GetGuestMetrics(targetID, time.Hour)["cpu"] + if len(inMemory) != 1 { + t.Fatalf("in-memory VM cpu points = %d, want one per source observation: %+v", len(inMemory), inMemory) + } + if !inMemory[0].Timestamp.Equal(observedAt) { + t.Fatalf("in-memory VM cpu timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt) + } + + persisted, err := persistentStore.Query("vm", targetID, "cpu", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted VM cpu: %v", err) + } + if len(persisted) != 1 { + t.Fatalf("persisted VM cpu points = %d, want one per source observation: %+v", len(persisted), persisted) + } + if !persisted[0].Timestamp.Equal(observedAt) { + t.Fatalf("persisted VM cpu timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt) + } +} + func TestSyncUnifiedAppContainerMetricsSkipsMockOwnedTrueNASHistoryWhenMockEnabled(t *testing.T) { previousFeature := truenas.IsFeatureEnabled() truenas.SetFeatureEnabled(true)