diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index d34cd67eaf..726ef9d42a 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -2197,7 +2197,7 @@ func TestUnifiedPhysicalDiskMetricsUseCanonicalDiskHistoryPath(t *testing.T) { "m.syncUnifiedPhysicalDiskMetrics(store)", `if target == nil || target.ResourceType != "disk" || strings.TrimSpace(target.ResourceID) == "" {`, `if source == unifiedresources.SourceProxmox || source == unifiedresources.SourceAgent {`, - `m.writeSMARTMetrics(disk, now)`, + `m.writeSMARTMetrics(disk, unifiedResourceObservedAt(resource, now))`, } for _, snippet := range requiredSnippets { if !strings.Contains(source, snippet) { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 378aba84b2..ef9e620dee 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -891,6 +891,11 @@ func hostsFromReadState(readState unifiedresources.ReadState) []models.Host { // writeSMARTMetrics writes SMART temperature history to the in-memory chart // buffer and persists SMART attributes when the metrics store is enabled. +// +// now is the source observation time, not the rebuild wall clock: a read-side +// registry rebuild that re-issues one unchanged observation must map to a +// single sample. The store writes go through the unified replay guard so the +// same observation is not re-committed on every rebuild (#1966). func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) { if shouldSkipNativeMockStateMetricWrites() { return @@ -901,56 +906,88 @@ func (m *Monitor) writeSMARTMetrics(disk models.PhysicalDisk, now time.Time) { return } + // Temperature (always write if > 0) + if disk.Temperature > 0 && m.metricsHistory != nil { + m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now) + } + + writes := m.smartMetricStoreWrites(disk, resourceID, now) + if len(writes) == 0 { + return + } + writes = m.dedupeUnifiedMetricWrites(writes) + if len(writes) > 0 { + m.metricsStore.WriteBatchBounded(writes) + } +} + +// smartMetricStoreWrites builds the persisted SMART writes for one physical +// disk observation. The caller owns batching so the unified sync can anchor the +// observation time and drop exact replays (#1966). +func (m *Monitor) smartMetricStoreWrites(disk models.PhysicalDisk, resourceID string, observedAt time.Time) []metrics.WriteMetric { + if m.metricsStore == nil { + return nil + } + + writes := make([]metrics.WriteMetric, 0, 12) + appendWrite := func(metricType string, value float64) { + writes = append(writes, metrics.WriteMetric{ + ResourceType: "disk", + ResourceID: resourceID, + MetricType: metricType, + Value: value, + Timestamp: observedAt, + Tier: metrics.TierRaw, + }) + } + // Temperature (always write if > 0) if disk.Temperature > 0 { - if m.metricsHistory != nil { - m.metricsHistory.AddDiskMetric(resourceID, "smart_temp", float64(disk.Temperature), now) - } - if m.metricsStore != nil { - m.metricsStore.Write("disk", resourceID, "smart_temp", float64(disk.Temperature), now) - } + appendWrite("smart_temp", float64(disk.Temperature)) } attrs := disk.SmartAttributes - if attrs == nil || m.metricsStore == nil { - return + if attrs == nil { + return writes } // Common if attrs.PowerOnHours != nil { - m.metricsStore.Write("disk", resourceID, "smart_power_on_hours", float64(*attrs.PowerOnHours), now) + appendWrite("smart_power_on_hours", float64(*attrs.PowerOnHours)) } if attrs.PowerCycles != nil { - m.metricsStore.Write("disk", resourceID, "smart_power_cycles", float64(*attrs.PowerCycles), now) + appendWrite("smart_power_cycles", float64(*attrs.PowerCycles)) } // SATA-specific if attrs.ReallocatedSectors != nil { - m.metricsStore.Write("disk", resourceID, "smart_reallocated_sectors", float64(*attrs.ReallocatedSectors), now) + appendWrite("smart_reallocated_sectors", float64(*attrs.ReallocatedSectors)) } if attrs.PendingSectors != nil { - m.metricsStore.Write("disk", resourceID, "smart_pending_sectors", float64(*attrs.PendingSectors), now) + appendWrite("smart_pending_sectors", float64(*attrs.PendingSectors)) } if attrs.OfflineUncorrectable != nil { - m.metricsStore.Write("disk", resourceID, "smart_offline_uncorrectable", float64(*attrs.OfflineUncorrectable), now) + appendWrite("smart_offline_uncorrectable", float64(*attrs.OfflineUncorrectable)) } if attrs.UDMACRCErrors != nil { - m.metricsStore.Write("disk", resourceID, "smart_crc_errors", float64(*attrs.UDMACRCErrors), now) + appendWrite("smart_crc_errors", float64(*attrs.UDMACRCErrors)) } // NVMe-specific if attrs.PercentageUsed != nil { - m.metricsStore.Write("disk", resourceID, "smart_percentage_used", float64(*attrs.PercentageUsed), now) + appendWrite("smart_percentage_used", float64(*attrs.PercentageUsed)) } if attrs.AvailableSpare != nil { - m.metricsStore.Write("disk", resourceID, "smart_available_spare", float64(*attrs.AvailableSpare), now) + appendWrite("smart_available_spare", float64(*attrs.AvailableSpare)) } if attrs.MediaErrors != nil { - m.metricsStore.Write("disk", resourceID, "smart_media_errors", float64(*attrs.MediaErrors), now) + appendWrite("smart_media_errors", float64(*attrs.MediaErrors)) } if attrs.UnsafeShutdowns != nil { - m.metricsStore.Write("disk", resourceID, "smart_unsafe_shutdowns", float64(*attrs.UnsafeShutdowns), now) + appendWrite("smart_unsafe_shutdowns", float64(*attrs.UnsafeShutdowns)) } + + return writes } // PollExecutor defines the contract for executing polling tasks. @@ -5687,7 +5724,10 @@ func (m *Monitor) syncUnifiedPhysicalDiskMetrics(store ResourceStoreInterface) { if disk.Serial == "" { disk.ID = targetID } - m.writeSMARTMetrics(disk, now) + // Anchor to the source observation time so a read-side registry rebuild + // re-issuing the same snapshot maps to one sample instead of a fresh + // wall-clock row (#1966). + m.writeSMARTMetrics(disk, unifiedResourceObservedAt(resource, now)) } } diff --git a/internal/monitoring/monitor_polling_test.go b/internal/monitoring/monitor_polling_test.go index aa2032e0f5..5d5ef4981b 100644 --- a/internal/monitoring/monitor_polling_test.go +++ b/internal/monitoring/monitor_polling_test.go @@ -1138,7 +1138,10 @@ func TestSyncUnifiedPhysicalDiskMetricsRecordsTrueNASDiskHistory(t *testing.T) { defer func() { _ = store.Close() }() resourceStore := unifiedresources.NewMonitorAdapter(nil) - records := truenas.NewProvider(truenas.DefaultFixtures()).Records() + fixtures := truenas.DefaultFixtures() + fixtures.CollectedAt = time.Now().UTC().Truncate(time.Second) + fixtures.System.CollectedAt = fixtures.CollectedAt + records := truenas.NewProvider(fixtures).Records() resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ unifiedresources.SourceTrueNAS: records, }) @@ -1186,6 +1189,79 @@ func TestSyncUnifiedPhysicalDiskMetricsRecordsTrueNASDiskHistory(t *testing.T) { } } +func TestSyncUnifiedPhysicalDiskMetricsUsesSourceObservationTimeAcrossRegistryRebuilds(t *testing.T) { + previous := truenas.IsFeatureEnabled() + truenas.SetFeatureEnabled(true) + t.Cleanup(func() { + truenas.SetFeatureEnabled(previous) + }) + + fixtures := truenas.DefaultFixtures() + fixtures.CollectedAt = time.Now().UTC().Truncate(time.Second) + fixtures.System.CollectedAt = fixtures.CollectedAt + observedAt := fixtures.CollectedAt + + resourceStore := unifiedresources.NewMonitorAdapter(nil) + resourceStore.PopulateSnapshotAndSupplemental(models.StateSnapshot{}, map[unifiedresources.DataSource][]unifiedresources.IngestRecord{ + unifiedresources.SourceTrueNAS: truenas.NewProvider(fixtures).Records(), + }) + + var diskResourceID string + for _, resource := range resourceStore.GetAll() { + if resource.Type != unifiedresources.ResourceTypePhysicalDisk || resource.PhysicalDisk == nil { + continue + } + if strings.TrimSpace(resource.PhysicalDisk.DevPath) == "/dev/sdc" { + diskResourceID = resource.ID + break + } + } + if diskResourceID == "" { + t.Fatal("expected TrueNAS sdc disk resource in unified store") + } + + target := resourceStore.MetricsTargetForResource(diskResourceID) + if target == nil || target.ResourceType != "disk" || target.ResourceID != "WD-WX12A3456" { + t.Fatalf("unexpected physical-disk metrics target %+v", target) + } + + 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.syncUnifiedPhysicalDiskMetrics(resourceStore) + monitor.syncUnifiedPhysicalDiskMetrics(resourceStore) + + inMemory := monitor.GetDiskMetrics(target.ResourceID, "smart_temp", time.Hour) + if len(inMemory) != 1 { + t.Fatalf("in-memory disk smart_temp points = %d, want one per source observation: %+v", len(inMemory), inMemory) + } + if !inMemory[0].Timestamp.Equal(observedAt) { + t.Fatalf("in-memory disk smart_temp timestamp = %s, want source observation %s", inMemory[0].Timestamp, observedAt) + } + + persisted, err := persistentStore.Query("disk", target.ResourceID, "smart_temp", observedAt.Add(-time.Second), observedAt.Add(time.Second), 0) + if err != nil { + t.Fatalf("query persisted disk smart_temp: %v", err) + } + if len(persisted) != 1 { + t.Fatalf("persisted disk smart_temp points = %d, want one per source observation: %+v", len(persisted), persisted) + } + if !persisted[0].Timestamp.Equal(observedAt) { + t.Fatalf("persisted disk smart_temp timestamp = %s, want source observation %s", persisted[0].Timestamp, observedAt) + } +} + func TestSyncUnifiedStorageAndDiskMetricsSkipMockOwnedTrueNASHistoryWhenMockEnabled(t *testing.T) { previousFeature := truenas.IsFeatureEnabled() truenas.SetFeatureEnabled(true)