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)