From a71f3707317b451bc97ccc8e06b67ed534256db2 Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 29 Mar 2026 21:17:51 +0100 Subject: [PATCH] feat(truenas): add native disk chart fallback --- .../v6/internal/subsystems/monitoring.md | 7 + .../monitoring/canonical_guardrails_test.go | 32 +++ internal/monitoring/monitor.go | 7 + internal/monitoring/monitor_metrics.go | 43 ++++ .../monitor_metrics_chart_fallback_test.go | 70 ++++++ .../monitoring/monitor_metrics_slo_test.go | 67 ++++++ internal/monitoring/truenas_poller.go | 55 +++++ internal/monitoring/truenas_poller_test.go | 74 +++++++ internal/truenas/client.go | 209 +++++++++++++++++- internal/truenas/client_test.go | 72 ++++++ internal/truenas/provider.go | 98 ++++++++ internal/truenas/provider_test.go | 63 +++++- internal/truenas/types.go | 7 + 13 files changed, 794 insertions(+), 10 deletions(-) diff --git a/docs/release-control/v6/internal/subsystems/monitoring.md b/docs/release-control/v6/internal/subsystems/monitoring.md index 4602ff992..1753a0b3e 100644 --- a/docs/release-control/v6/internal/subsystems/monitoring.md +++ b/docs/release-control/v6/internal/subsystems/monitoring.md @@ -245,6 +245,13 @@ physical-disk resources such as TrueNAS disks into the shared `disk` metrics-store contract via the existing SMART-temperature writer, so physical disk charts and disk-health consumers read one history path instead of a TrueNAS-only temperature cache. +That same boundary now also owns native disk-history fallback when Pulse's own +history is shallow. `internal/truenas/client.go`, +`internal/truenas/provider.go`, `internal/monitoring/truenas_poller.go`, and +`internal/monitoring/monitor_metrics.go` must route TrueNAS `disktemp` +reporting history through the shared physical-disk chart path, so canonical +disk charts can render real provider-backed history instead of flat padding +after restarts or immediately after onboarding. That same monitoring boundary now also owns modern TrueNAS app workload telemetry. `internal/truenas/client.go`, `internal/truenas/provider.go`, and `internal/monitoring/monitor.go` must ingest `app.stats` through the official diff --git a/internal/monitoring/canonical_guardrails_test.go b/internal/monitoring/canonical_guardrails_test.go index ecba3af6b..ecff6f29c 100644 --- a/internal/monitoring/canonical_guardrails_test.go +++ b/internal/monitoring/canonical_guardrails_test.go @@ -207,6 +207,38 @@ func TestUnifiedPhysicalDiskMetricsUseCanonicalDiskHistoryPath(t *testing.T) { } } +func TestUnifiedPhysicalDiskMetricsAllowNativeHistoryProviders(t *testing.T) { + monitorData, err := os.ReadFile("monitor.go") + if err != nil { + t.Fatalf("failed to read monitor.go: %v", err) + } + monitorSource := string(monitorData) + monitorSnippets := []string{ + "type MonitorPhysicalDiskTemperatureHistoryProvider interface {", + `PhysicalDiskTemperatureHistory(m *Monitor, orgID string, duration time.Duration) map[string][]MetricPoint`, + } + for _, snippet := range monitorSnippets { + if !strings.Contains(monitorSource, snippet) { + t.Fatalf("monitor.go must contain %q", snippet) + } + } + + pollerData, err := os.ReadFile("truenas_poller.go") + if err != nil { + t.Fatalf("failed to read truenas_poller.go: %v", err) + } + pollerSource := string(pollerData) + pollerSnippets := []string{ + "func (p *TrueNASPoller) PhysicalDiskTemperatureHistory(_ *Monitor, orgID string, duration time.Duration) map[string][]MetricPoint {", + "entry.provider.PhysicalDiskTemperatureHistory(ctx, duration)", + } + for _, snippet := range pollerSnippets { + if !strings.Contains(pollerSource, snippet) { + t.Fatalf("truenas_poller.go must contain %q", snippet) + } + } +} + func TestTrueNASSystemTelemetryUsesCanonicalHostTemperatureModel(t *testing.T) { clientData, err := os.ReadFile(filepath.Join("..", "truenas", "client.go")) if err != nil { diff --git a/internal/monitoring/monitor.go b/internal/monitoring/monitor.go index 514ce8009..f66d575ee 100644 --- a/internal/monitoring/monitor.go +++ b/internal/monitoring/monitor.go @@ -142,6 +142,13 @@ type MonitorSupplementalRecordsProvider interface { SupplementalRecords(m *Monitor, orgID string) []unifiedresources.IngestRecord } +// MonitorPhysicalDiskTemperatureHistoryProvider optionally exposes source-native +// physical-disk temperature history through the canonical monitoring chart +// boundary when Pulse's own stored history is shallow. +type MonitorPhysicalDiskTemperatureHistoryProvider interface { + PhysicalDiskTemperatureHistory(m *Monitor, orgID string, duration time.Duration) map[string][]MetricPoint +} + func getNodeDisplayName(instance *config.PVEInstance, nodeName string) string { baseName := strings.TrimSpace(nodeName) if baseName == "" { diff --git a/internal/monitoring/monitor_metrics.go b/internal/monitoring/monitor_metrics.go index 378a92fd2..037f90551 100644 --- a/internal/monitoring/monitor_metrics.go +++ b/internal/monitoring/monitor_metrics.go @@ -215,6 +215,7 @@ func (m *Monitor) GetPhysicalDiskTemperatureCharts(duration time.Duration) map[s resourceIDs[i] = d.resourceID } batchMetrics := m.queryStoreBatchMetricMapWithGapFill("disk", resourceIDs, duration) + nativeHistory := m.nativePhysicalDiskTemperatureHistory(duration) // Phase 3: Build result entries. result := make(map[string]DiskChartEntry, len(disks)) @@ -225,6 +226,13 @@ func (m *Monitor) GetPhysicalDiskTemperatureCharts(duration time.Duration) map[s tempPoints = pts } } + if nativePoints, ok := nativeHistory[d.resourceID]; ok { + nativePoints = lttb(nativePoints, chartDownsampleTarget) + if chartSeriesCoverageSpan(nativePoints) > chartSeriesCoverageSpan(tempPoints) && + (!hasSufficientChartSeriesCoverage(tempPoints, duration) || len(tempPoints) < 2) { + tempPoints = nativePoints + } + } // Sparklines require >= 2 points. If the store returned 0 or 1 points // but the disk has a live temperature reading, pad to 2 points so the @@ -248,6 +256,41 @@ func (m *Monitor) GetPhysicalDiskTemperatureCharts(duration time.Duration) map[s return result } +func (m *Monitor) nativePhysicalDiskTemperatureHistory(duration time.Duration) map[string][]MetricPoint { + providers := m.supplementalProviderSnapshot() + if len(providers) == 0 { + return nil + } + + orgID := "default" + if m != nil { + if trimmed := strings.TrimSpace(m.GetOrgID()); trimmed != "" { + orgID = trimmed + } + } + + history := make(map[string][]MetricPoint) + for _, provider := range providers { + historyProvider, ok := provider.(MonitorPhysicalDiskTemperatureHistoryProvider) + if !ok { + continue + } + nativeHistory := historyProvider.PhysicalDiskTemperatureHistory(m, orgID, duration) + for resourceID, points := range nativeHistory { + if strings.TrimSpace(resourceID) == "" || len(points) == 0 { + continue + } + if existing, ok := history[resourceID]; !ok || chartSeriesCoverageSpan(points) > chartSeriesCoverageSpan(existing) { + history[resourceID] = points + } + } + } + if len(history) == 0 { + return nil + } + return history +} + func (m *Monitor) currentMetricsTargetStore() MetricsTargetResourceStore { if m == nil { return nil diff --git a/internal/monitoring/monitor_metrics_chart_fallback_test.go b/internal/monitoring/monitor_metrics_chart_fallback_test.go index 71a09cce3..3a31198f7 100644 --- a/internal/monitoring/monitor_metrics_chart_fallback_test.go +++ b/internal/monitoring/monitor_metrics_chart_fallback_test.go @@ -9,6 +9,27 @@ import ( "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" ) +type stubDiskTemperatureHistoryProvider struct { + history map[string][]MetricPoint +} + +func (s *stubDiskTemperatureHistoryProvider) SupplementalRecords(*Monitor, string) []unifiedresources.IngestRecord { + return nil +} + +func (s *stubDiskTemperatureHistoryProvider) PhysicalDiskTemperatureHistory(*Monitor, string, time.Duration) map[string][]MetricPoint { + if len(s.history) == 0 { + return nil + } + result := make(map[string][]MetricPoint, len(s.history)) + for resourceID, points := range s.history { + copied := make([]MetricPoint, len(points)) + copy(copied, points) + result[resourceID] = copied + } + return result +} + func newChartFallbackTestMonitor(t *testing.T) *Monitor { t.Helper() @@ -191,6 +212,55 @@ func TestGetPhysicalDiskTemperatureCharts_UsesUnifiedReadStateDiskViews(t *testi } } +func TestGetPhysicalDiskTemperatureCharts_UsesNativeHistoryWhenStoreCoverageShallow(t *testing.T) { + t.Parallel() + + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(models.StateSnapshot{ + PhysicalDisks: []models.PhysicalDisk{ + { + ID: "disk-1", + Node: "truenas-main", + Instance: "", + DevPath: "/dev/sda", + Model: "Seagate Exos X18", + Serial: "SERIAL-DISK-1", + Temperature: 34, + LastChecked: time.Now().UTC(), + }, + }, + }) + + now := time.Now().UTC().Truncate(time.Second) + monitor := &Monitor{ + state: models.NewState(), + resourceStore: unifiedresources.NewMonitorAdapter(registry), + supplementalProviders: map[unifiedresources.DataSource]MonitorSupplementalRecordsProvider{ + unifiedresources.SourceTrueNAS: &stubDiskTemperatureHistoryProvider{ + history: map[string][]MetricPoint{ + "SERIAL-DISK-1": { + {Timestamp: now.Add(-2 * time.Hour), Value: 29}, + {Timestamp: now.Add(-1 * time.Hour), Value: 31}, + {Timestamp: now, Value: 34}, + }, + }, + }, + }, + } + + charts := monitor.GetPhysicalDiskTemperatureCharts(4 * time.Hour) + entry, ok := charts["SERIAL-DISK-1"] + if !ok { + t.Fatalf("expected chart entry for canonical disk metric id, got %#v", charts) + } + if len(entry.Temperature) != 3 { + t.Fatalf("expected native history points instead of padded fallback, got %+v", entry.Temperature) + } + if entry.Temperature[0].Value != 29 || entry.Temperature[len(entry.Temperature)-1].Value != 34 { + t.Fatalf("expected native history values to win, got %+v", entry.Temperature) + } +} + func TestGetGuestMetricsForChart_UsesGapFillLookbackWhenRequestedRangeIsEmpty(t *testing.T) { t.Parallel() diff --git a/internal/monitoring/monitor_metrics_slo_test.go b/internal/monitoring/monitor_metrics_slo_test.go index ad1121b07..d93596680 100644 --- a/internal/monitoring/monitor_metrics_slo_test.go +++ b/internal/monitoring/monitor_metrics_slo_test.go @@ -7,6 +7,8 @@ import ( "testing" "time" + "github.com/rcourtman/pulse-go-rewrite/internal/models" + "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/metrics" "github.com/rs/zerolog" "github.com/rs/zerolog/log" @@ -27,6 +29,8 @@ const ( SLONodeChartBatchP95 = 35 * time.Millisecond SLOGuestChartBatchGitHubActionsP95 = 220 * time.Millisecond SLONodeChartBatchGitHubActionsP95 = 140 * time.Millisecond + SLOPhysicalDiskChartFallbackP95 = 30 * time.Millisecond + SLOPhysicalDiskChartFallbackGHA = 120 * time.Millisecond monitoringSLOIterations = 120 ) @@ -231,3 +235,66 @@ func TestSLO_GetNodeMetricsForChartBatch(t *testing.T) { t.Errorf("SLO VIOLATION: p95=%v exceeds target %v", p95, target) } } + +func TestSLO_GetPhysicalDiskTemperatureCharts_WithNativeHistoryFallback(t *testing.T) { + skipMonitoringSLOUnderRace(t) + suppressMonitoringTestLogs(t) + + monitor := newChartFallbackTestMonitor(t) + now := time.Now().UTC().Truncate(time.Second) + registry := unifiedresources.NewRegistry(nil) + registry.IngestSnapshot(models.StateSnapshot{ + PhysicalDisks: []models.PhysicalDisk{ + { + ID: "disk-1", + Node: "truenas-main", + DevPath: "/dev/sda", + Model: "Seagate Exos X18", + Serial: "SERIAL-DISK-1", + Temperature: 34, + LastChecked: now, + }, + }, + }) + monitor.resourceStore = unifiedresources.NewMonitorAdapter(registry) + monitor.supplementalProviders = map[unifiedresources.DataSource]MonitorSupplementalRecordsProvider{ + unifiedresources.SourceTrueNAS: &stubDiskTemperatureHistoryProvider{ + history: map[string][]MetricPoint{ + "SERIAL-DISK-1": { + {Timestamp: now.Add(-2 * time.Hour), Value: 29}, + {Timestamp: now.Add(-1 * time.Hour), Value: 31}, + {Timestamp: now, Value: 34}, + }, + }, + }, + } + + sanity := monitor.GetPhysicalDiskTemperatureCharts(4 * time.Hour) + entry, ok := sanity["SERIAL-DISK-1"] + if !ok { + t.Fatalf("sanity: expected chart entry for canonical disk metric id, got %#v", sanity) + } + if len(entry.Temperature) != 3 { + t.Fatalf("sanity: expected native history points instead of padded fallback, got %+v", entry.Temperature) + } + + latencies := measureMonitoringLatencies(t, func() { + result := monitor.GetPhysicalDiskTemperatureCharts(4 * time.Hour) + entry, ok := result["SERIAL-DISK-1"] + if !ok { + t.Fatalf("expected chart entry for canonical disk metric id, got %#v", result) + } + if len(entry.Temperature) != 3 { + t.Fatalf("expected native history points instead of padded fallback, got %+v", entry.Temperature) + } + }) + + target := effectiveMonitoringSLOTarget(SLOPhysicalDiskChartFallbackP95, SLOPhysicalDiskChartFallbackGHA) + p95 := monitoringPercentile(latencies, 0.95) + t.Logf("GetPhysicalDiskTemperatureCharts(native-history fallback) p50=%v p95=%v p99=%v SLO=%v", + monitoringPercentile(latencies, 0.50), p95, monitoringPercentile(latencies, 0.99), target) + + if p95 > target { + t.Errorf("SLO VIOLATION: p95=%v exceeds target %v", p95, target) + } +} diff --git a/internal/monitoring/truenas_poller.go b/internal/monitoring/truenas_poller.go index 51cf7b685..21d7e0cbd 100644 --- a/internal/monitoring/truenas_poller.go +++ b/internal/monitoring/truenas_poller.go @@ -23,6 +23,8 @@ import ( const defaultTrueNASPollInterval = 60 * time.Second +const defaultTrueNASHistoryReadTimeout = 10 * time.Second + // TrueNASPoller manages periodic polling of configured TrueNAS connections. type TrueNASPoller struct { multiTenant *config.MultiTenantPersistence @@ -447,6 +449,59 @@ func (p *TrueNASPoller) SupplementalRecords(_ *Monitor, orgID string) []unifiedr return p.GetCurrentRecordsForOrg(orgID) } +// PhysicalDiskTemperatureHistory exposes native TrueNAS disk temperature +// history through the canonical monitoring chart boundary. +func (p *TrueNASPoller) PhysicalDiskTemperatureHistory(_ *Monitor, orgID string, duration time.Duration) map[string][]MetricPoint { + if p == nil || !truenas.IsFeatureEnabled() { + return nil + } + + orgID = strings.TrimSpace(orgID) + if orgID == "" { + orgID = "default" + } + + entries := p.providerEntriesForOrg(orgID) + if len(entries) == 0 { + return nil + } + + ctx, cancel := context.WithTimeout(context.Background(), defaultTrueNASHistoryReadTimeout) + defer cancel() + + history := make(map[string][]MetricPoint) + for _, entry := range entries { + nativeHistory, err := entry.provider.PhysicalDiskTemperatureHistory(ctx, duration) + if err != nil { + log.Warn(). + Str("component", "truenas_poller"). + Str("action", "disk_temperature_history"). + Str("org_id", orgID). + Str("connection_id", strings.TrimSpace(entry.connectionID)). + Err(err). + Msg("TrueNAS poller failed to read native disk temperature history") + continue + } + for resourceID, points := range nativeHistory { + if strings.TrimSpace(resourceID) == "" || len(points) == 0 { + continue + } + converted := make([]MetricPoint, len(points)) + for i, point := range points { + converted[i] = MetricPoint{ + Timestamp: point.Timestamp, + Value: point.Value, + } + } + history[resourceID] = converted + } + } + if len(history) == 0 { + return nil + } + return history +} + // SnapshotOwnedSources declares source-native ingest ownership for legacy // snapshot suppression (default/global call path). func (p *TrueNASPoller) SnapshotOwnedSources() []unifiedresources.DataSource { diff --git a/internal/monitoring/truenas_poller_test.go b/internal/monitoring/truenas_poller_test.go index 62c767449..2c4c57dc4 100644 --- a/internal/monitoring/truenas_poller_test.go +++ b/internal/monitoring/truenas_poller_test.go @@ -249,6 +249,80 @@ func TestTrueNASPollerRecordsMetrics(t *testing.T) { } } +func TestTrueNASPollerPhysicalDiskTemperatureHistoryUsesTenantScopedProvider(t *testing.T) { + previous := truenas.IsFeatureEnabled() + truenas.SetFeatureEnabled(true) + t.Cleanup(func() { truenas.SetFeatureEnabled(previous) }) + + fixtures := truenas.DefaultFixtures() + now := time.Date(2026, 3, 29, 20, 0, 0, 0, time.UTC) + fetcher := &controllableTrueNASHistoryFetcher{ + snapshot: &fixtures, + history: map[string][]truenas.TimeSeriesPoint{ + "sda": { + {Timestamp: now.Add(-2 * time.Hour), Value: 30}, + {Timestamp: now.Add(-1 * time.Hour), Value: 32}, + {Timestamp: now, Value: 34}, + }, + }, + } + provider := truenas.NewLiveProvider(fetcher) + if err := provider.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh() error = %v", err) + } + + poller := NewTrueNASPoller(nil, time.Minute, nil) + poller.providersByOrg["default"] = map[string]*truenas.Provider{ + "conn-1": provider, + } + + history := poller.PhysicalDiskTemperatureHistory(nil, "default", 4*time.Hour) + points, ok := history["ZL0A1234"] + if !ok { + t.Fatalf("expected canonical metric resource id ZL0A1234, got %#v", history) + } + if len(points) != 3 || points[len(points)-1].Value != 34 { + t.Fatalf("unexpected tenant-scoped disk history: %+v", points) + } +} + +type controllableTrueNASHistoryFetcher struct { + snapshot *truenas.FixtureSnapshot + history map[string][]truenas.TimeSeriesPoint +} + +func (s *controllableTrueNASHistoryFetcher) Fetch(context.Context) (*truenas.FixtureSnapshot, error) { + if s == nil || s.snapshot == nil { + return nil, nil + } + copied := *s.snapshot + copied.Disks = append([]truenas.Disk(nil), s.snapshot.Disks...) + copied.Pools = append([]truenas.Pool(nil), s.snapshot.Pools...) + copied.Datasets = append([]truenas.Dataset(nil), s.snapshot.Datasets...) + copied.Alerts = append([]truenas.Alert(nil), s.snapshot.Alerts...) + copied.Apps = append([]truenas.App(nil), s.snapshot.Apps...) + copied.ZFSSnapshots = append([]truenas.ZFSSnapshot(nil), s.snapshot.ZFSSnapshots...) + copied.ReplicationTasks = append([]truenas.ReplicationTask(nil), s.snapshot.ReplicationTasks...) + return &copied, nil +} + +func (s *controllableTrueNASHistoryFetcher) DiskTemperatureHistory(_ context.Context, identifiers []string, _ time.Duration) (map[string][]truenas.TimeSeriesPoint, error) { + result := make(map[string][]truenas.TimeSeriesPoint) + for _, identifier := range identifiers { + points, ok := s.history[identifier] + if !ok || len(points) == 0 { + continue + } + copied := make([]truenas.TimeSeriesPoint, len(points)) + copy(copied, points) + result[identifier] = copied + } + if len(result) == 0 { + return nil, nil + } + return result, nil +} + type pollerControlFetcher struct { snapshot *truenas.FixtureSnapshot startCalls []string diff --git a/internal/truenas/client.go b/internal/truenas/client.go index b3ef8cff1..fb36b1c7c 100644 --- a/internal/truenas/client.go +++ b/internal/truenas/client.go @@ -348,6 +348,27 @@ func (c *Client) GetDiskTemperatures(ctx context.Context) (map[string]int, error return c.getDiskTemperaturesWithFallback(ctx, nil) } +// GetDiskTemperatureHistory returns recent disk temperature series by TrueNAS +// disk identifier using the native reporting API. +func (c *Client) GetDiskTemperatureHistory(ctx context.Context, identifiers []string, duration time.Duration) (map[string][]TimeSeriesPoint, error) { + identifiers = dedupeStrings(identifiers) + if len(identifiers) == 0 { + return nil, fmt.Errorf("truenas disk temperature history requires disk identifiers") + } + + conn, err := c.dialRPC(ctx) + if err != nil { + return nil, err + } + defer func() { _ = conn.Close() }() + + rpc := &trueNASRPCClient{conn: conn, nextID: 1} + if err := rpc.authenticate(ctx, c.config); err != nil { + return nil, err + } + return rpc.getDiskTemperatureHistory(ctx, identifiers, duration) +} + func (c *Client) getDiskTemperaturesWithFallback(ctx context.Context, identifiers []string) (map[string]int, error) { var response any restErr := c.getJSON(ctx, http.MethodGet, "/disk/temperatures", &response) @@ -1378,6 +1399,43 @@ func (c *trueNASRPCClient) getDiskTemperatureAggregates(ctx context.Context, ide return parseDiskTemperatureAggregates(response, windowDays), nil } +func (c *trueNASRPCClient) getDiskTemperatureHistory(ctx context.Context, identifiers []string, duration time.Duration) (map[string][]TimeSeriesPoint, error) { + if c == nil || c.conn == nil { + return nil, fmt.Errorf("truenas rpc connection is nil") + } + identifiers = dedupeStrings(identifiers) + if len(identifiers) == 0 { + return nil, fmt.Errorf("truenas rpc disk temperature history query requires at least one identifier") + } + if duration <= 0 { + duration = 24 * time.Hour + } + + end := time.Now().Unix() + start := end - int64(duration.Seconds()) + if start <= 0 { + start = end + } + + graphs := make([]map[string]any, 0, len(identifiers)) + for _, identifier := range identifiers { + graphs = append(graphs, map[string]any{ + "name": "disktemp", + "identifier": identifier, + }) + } + + response, err := c.getReportingDataWithQuery(ctx, graphs, map[string]any{ + "aggregate": false, + "start": start, + "end": end, + }) + if err != nil { + return nil, err + } + return parseReportingDiskTemperatureHistory(response), nil +} + func (c *trueNASRPCClient) getReportingData(ctx context.Context, graphs []map[string]any) ([]trueNASReportingGetDataResponse, error) { if c == nil || c.conn == nil { return nil, fmt.Errorf("truenas rpc connection is nil") @@ -1392,15 +1450,29 @@ func (c *trueNASRPCClient) getReportingData(ctx context.Context, graphs []map[st start = end } - var response []trueNASReportingGetDataResponse + return c.getReportingDataWithQuery(ctx, graphs, map[string]any{ + "aggregate": true, + "start": start, + "end": end, + }) +} + +func (c *trueNASRPCClient) getReportingDataWithQuery(ctx context.Context, graphs []map[string]any, query map[string]any) ([]trueNASReportingGetDataResponse, error) { + if c == nil || c.conn == nil { + return nil, fmt.Errorf("truenas rpc connection is nil") + } + if len(graphs) == 0 { + return nil, fmt.Errorf("truenas reporting query requires at least one graph") + } + if len(query) == 0 { + return nil, fmt.Errorf("truenas reporting query requires options") + } + params := []any{ graphs, - map[string]any{ - "aggregate": true, - "start": start, - "end": end, - }, + query, } + var response []trueNASReportingGetDataResponse if err := c.call(ctx, "reporting.get_data", params, &response); err != nil { return nil, err } @@ -1615,6 +1687,41 @@ func parseReportingDiskTemperatures(responses []trueNASReportingGetDataResponse) return temperatures } +func parseReportingDiskTemperatureHistory(responses []trueNASReportingGetDataResponse) map[string][]TimeSeriesPoint { + if len(responses) == 0 { + return nil + } + + history := make(map[string][]TimeSeriesPoint) + for _, response := range responses { + if strings.TrimSpace(strings.ToLower(response.Name)) != "disktemp" { + continue + } + + identifier := readStringAny(map[string]any{"identifier": response.Identifier}, "identifier") + if identifier == "" { + continue + } + + points := make([]TimeSeriesPoint, 0, len(response.Data)) + for _, raw := range response.Data { + timestamp, value, ok := parseReportingSeriesPoint(raw, response.Legend) + if !ok || timestamp.IsZero() { + continue + } + points = append(points, TimeSeriesPoint{Timestamp: timestamp, Value: value}) + } + if len(points) == 0 { + continue + } + history[identifier] = points + } + if len(history) == 0 { + return nil + } + return history +} + func parseDiskTemperatureAggregates(raw any, defaultWindowDays int) map[string]DiskTemperatureAggregate { if defaultWindowDays <= 0 { defaultWindowDays = defaultDiskTemperatureAggregateWindowDays @@ -1703,6 +1810,96 @@ func readFloatValueAny(record map[string]any, keys ...string) (float64, bool) { return 0, false } +func parseReportingSeriesPoint(raw any, legends []string) (time.Time, float64, bool) { + switch typed := raw.(type) { + case []any: + if len(typed) < 2 { + return time.Time{}, 0, false + } + timestamp, ok := parseReportingTimestampAny(typed[0]) + if !ok { + return time.Time{}, 0, false + } + if parsed, ok := parseFloat64Any(typed[1]); ok { + return timestamp, parsed, true + } + values := extractReportingLegendFloatValues(typed[1:], legends) + for _, legend := range legends { + if value, ok := values[legend]; ok { + return timestamp, value, true + } + } + case map[string]any: + timestamp, ok := parseReportingTimestampAny( + firstNonNilMapValue(typed, "timestamp", "time", "ts", "x"), + ) + if !ok { + return time.Time{}, 0, false + } + values := extractReportingLegendFloatValues(typed, legends) + for _, legend := range legends { + if value, ok := values[legend]; ok { + return timestamp, value, true + } + } + if value, ok := readFloatValueAny(typed, "value", "y", "temperature"); ok { + return timestamp, value, true + } + } + return time.Time{}, 0, false +} + +func firstNonNilMapValue(record map[string]any, keys ...string) any { + for _, key := range keys { + if value, ok := record[key]; ok && value != nil { + return value + } + } + return nil +} + +func parseReportingTimestampAny(raw any) (time.Time, bool) { + switch typed := raw.(type) { + case time.Time: + if typed.IsZero() { + return time.Time{}, false + } + return typed.UTC(), true + case string: + trimmed := strings.TrimSpace(typed) + if trimmed == "" { + return time.Time{}, false + } + if parsed, err := time.Parse(time.RFC3339, trimmed); err == nil { + return parsed.UTC(), true + } + if integer, err := strconv.ParseInt(trimmed, 10, 64); err == nil { + return unixReportingTimestamp(integer), true + } + case json.Number: + if integer, err := typed.Int64(); err == nil { + return unixReportingTimestamp(integer), true + } + if value, err := typed.Float64(); err == nil { + return unixReportingTimestamp(int64(math.Round(value))), true + } + default: + if value, ok := parseFloat64Any(raw); ok { + return unixReportingTimestamp(int64(math.Round(value))), true + } + } + return time.Time{}, false +} + +func unixReportingTimestamp(value int64) time.Time { + switch { + case value >= 1_000_000_000_000: + return time.UnixMilli(value).UTC() + default: + return time.Unix(value, 0).UTC() + } +} + func readIntValueAny(record map[string]any, keys ...string) int { for _, key := range keys { value, ok := record[key] diff --git a/internal/truenas/client_test.go b/internal/truenas/client_test.go index 95c2d67f3..99e1c98b4 100644 --- a/internal/truenas/client_test.go +++ b/internal/truenas/client_test.go @@ -763,6 +763,78 @@ func TestGetDisksIncludesDiskTemperatureAggregatesFromRPC(t *testing.T) { } } +func TestGetDiskTemperatureHistoryUsesReportingRPC(t *testing.T) { + server := newMockServerWithRPC(t, defaultAPIResponses(), nil, func(t *testing.T, conn *websocket.Conn) { + authReq := readRPCRequest(t, conn) + if authReq.Method != "auth.login_with_api_key" { + t.Fatalf("expected api-key auth method, got %q", authReq.Method) + } + writeRPCResult(t, conn, authReq.ID, true) + + historyReq := readRPCRequest(t, conn) + if historyReq.Method != "reporting.get_data" { + t.Fatalf("expected reporting.get_data, got %q", historyReq.Method) + } + params, ok := historyReq.Params.([]any) + if !ok || len(params) != 2 { + t.Fatalf("unexpected history params: %#v", historyReq.Params) + } + graphs, ok := params[0].([]any) + if !ok || len(graphs) != 1 { + t.Fatalf("unexpected history graphs: %#v", params[0]) + } + graph, ok := graphs[0].(map[string]any) + if !ok { + t.Fatalf("unexpected history graph entry: %#v", graphs[0]) + } + if got := readStringAny(graph, "name"); got != "disktemp" { + t.Fatalf("expected disktemp graph, got %q", got) + } + if got := readStringAny(graph, "identifier"); got != "sda" { + t.Fatalf("expected sda identifier, got %q", got) + } + query, ok := params[1].(map[string]any) + if !ok { + t.Fatalf("unexpected history query: %#v", params[1]) + } + if aggregate := query["aggregate"]; aggregate != false { + t.Fatalf("expected aggregate=false for history query, got %#v", aggregate) + } + + now := time.Now().UTC().Truncate(time.Second) + writeRPCResult(t, conn, historyReq.ID, []map[string]any{{ + "name": "disktemp", + "identifier": "sda", + "legend": []string{"temperature"}, + "data": []any{ + []any{now.Add(-2 * time.Hour).Unix(), 30.0}, + []any{now.Add(-1 * time.Hour).Unix(), 31.5}, + []any{now.Unix(), 33.0}, + }, + "aggregations": map[string]any{}, + "start": now.Add(-2 * time.Hour).Unix(), + "end": now.Unix(), + }}) + }) + t.Cleanup(server.Close) + + client := mustClientForServer(t, server.URL, ClientConfig{APIKey: "api-key"}) + history, err := client.GetDiskTemperatureHistory(context.Background(), []string{"sda"}, 4*time.Hour) + if err != nil { + t.Fatalf("GetDiskTemperatureHistory() error = %v", err) + } + points, ok := history["sda"] + if !ok { + t.Fatalf("expected history for sda, got %#v", history) + } + if len(points) != 3 { + t.Fatalf("expected 3 history points, got %+v", points) + } + if points[0].Value != 30.0 || points[2].Value != 33.0 { + t.Fatalf("unexpected history values: %+v", points) + } +} + func TestClientHandlesHTTPAndDecodeErrors(t *testing.T) { t.Run("non-2xx response", func(t *testing.T) { server := newMockServer(t, map[string]apiResponse{ diff --git a/internal/truenas/provider.go b/internal/truenas/provider.go index df181d023..bb8d45184 100644 --- a/internal/truenas/provider.go +++ b/internal/truenas/provider.go @@ -56,6 +56,10 @@ type appReadFetcher interface { ReadAppLogs(ctx context.Context, appName, containerID string, tailLines int) ([]AppLogLine, error) } +type physicalDiskHistoryFetcher interface { + DiskTemperatureHistory(ctx context.Context, identifiers []string, duration time.Duration) (map[string][]TimeSeriesPoint, error) +} + // APIFetcher loads snapshots from the live TrueNAS API client. type APIFetcher struct { Client *Client @@ -98,6 +102,13 @@ func (f *APIFetcher) ReadAppLogs(ctx context.Context, appName, containerID strin return f.Client.GetAppLogs(ctx, appName, containerID, tailLines) } +func (f *APIFetcher) DiskTemperatureHistory(ctx context.Context, identifiers []string, duration time.Duration) (map[string][]TimeSeriesPoint, error) { + if f == nil || f.Client == nil { + return nil, fmt.Errorf("truenas api fetcher client is nil") + } + return f.Client.GetDiskTemperatureHistory(ctx, identifiers, duration) +} + // FixtureFetcher loads snapshots from static fixture data. type FixtureFetcher struct { Snapshot FixtureSnapshot @@ -255,6 +266,67 @@ func (p *Provider) GetAppConfig(_ context.Context, appID string) (*AppConfigResu return result, nil } +// PhysicalDiskTemperatureHistory returns canonical physical-disk temperature +// series keyed by the shared physical-disk metrics resource IDs. +func (p *Provider) PhysicalDiskTemperatureHistory(ctx context.Context, duration time.Duration) (map[string][]TimeSeriesPoint, error) { + if p == nil { + return nil, fmt.Errorf("truenas provider is nil") + } + historyFetcher, ok := p.fetcher.(physicalDiskHistoryFetcher) + if !ok { + return nil, fmt.Errorf("truenas provider fetcher does not support physical disk history") + } + + snapshot := p.Snapshot() + if snapshot == nil { + return nil, fmt.Errorf("truenas provider has no cached snapshot") + } + + identifiers := make([]string, 0, len(snapshot.Disks)) + metricIDsByIdentifier := make(map[string]string, len(snapshot.Disks)*3) + for _, disk := range snapshot.Disks { + metricID := trueNASDiskMetricResourceID(disk) + if metricID == "" { + continue + } + if name := strings.TrimSpace(disk.Name); name != "" { + identifiers = append(identifiers, name) + } + for _, key := range trueNASDiskHistoryLookupKeys(disk) { + if _, exists := metricIDsByIdentifier[key]; !exists { + metricIDsByIdentifier[key] = metricID + } + } + } + identifiers = dedupeStrings(identifiers) + if len(identifiers) == 0 { + return nil, nil + } + + nativeHistory, err := historyFetcher.DiskTemperatureHistory(ctx, identifiers, duration) + if err != nil { + return nil, err + } + if len(nativeHistory) == 0 { + return nil, nil + } + + historyByMetricID := make(map[string][]TimeSeriesPoint, len(nativeHistory)) + for identifier, points := range nativeHistory { + metricID := metricIDsByIdentifier[strings.TrimSpace(identifier)] + if metricID == "" || len(points) == 0 { + continue + } + copied := make([]TimeSeriesPoint, len(points)) + copy(copied, points) + historyByMetricID[metricID] = copied + } + if len(historyByMetricID) == 0 { + return nil, nil + } + return historyByMetricID, nil +} + // Close releases resources held by the active fetcher, if supported. func (p *Provider) Close() { if p == nil || p.fetcher == nil { @@ -1336,6 +1408,32 @@ func temperatureAggregateMetaFromTrueNASDisk(disk Disk) *unifiedresources.Temper } } +func trueNASDiskHistoryLookupKeys(disk Disk) []string { + return dedupeStrings([]string{ + strings.TrimSpace(disk.Name), + strings.TrimSpace(disk.ID), + strings.TrimSpace(disk.Serial), + }) +} + +func trueNASDiskMetricResourceID(disk Disk) string { + devPath := "" + if name := strings.TrimSpace(disk.Name); name != "" { + devPath = "/dev/" + name + } + meta := &unifiedresources.PhysicalDiskMeta{ + DevPath: devPath, + Serial: strings.TrimSpace(disk.Serial), + DiskType: strings.TrimSpace(disk.Transport), + SizeBytes: disk.SizeBytes, + } + fallback := strings.TrimSpace(disk.ID) + if fallback == "" { + fallback = strings.TrimSpace(disk.Name) + } + return unifiedresources.PhysicalDiskMetaMetricID(meta, fallback) +} + func parentPoolFromDataset(datasetName string) string { parts := strings.SplitN(strings.TrimSpace(datasetName), "/", 2) if len(parts) == 0 { diff --git a/internal/truenas/provider_test.go b/internal/truenas/provider_test.go index d82bf6907..28301fea4 100644 --- a/internal/truenas/provider_test.go +++ b/internal/truenas/provider_test.go @@ -38,10 +38,11 @@ func (s *closableStubFetcher) Close() { } type controllableStubFetcher struct { - snapshot *FixtureSnapshot - startCalls []string - stopCalls []string - logReads []appLogReadCall + snapshot *FixtureSnapshot + startCalls []string + stopCalls []string + logReads []appLogReadCall + diskHistory map[string][]TimeSeriesPoint } type appLogReadCall struct { @@ -92,6 +93,26 @@ func (s *controllableStubFetcher) ReadAppLogs(_ context.Context, appName, contai }, nil } +func (s *controllableStubFetcher) DiskTemperatureHistory(_ context.Context, identifiers []string, _ time.Duration) (map[string][]TimeSeriesPoint, error) { + if len(s.diskHistory) == 0 { + return nil, nil + } + result := make(map[string][]TimeSeriesPoint) + for _, identifier := range identifiers { + points, ok := s.diskHistory[identifier] + if !ok || len(points) == 0 { + continue + } + copied := make([]TimeSeriesPoint, len(points)) + copy(copied, points) + result[identifier] = copied + } + if len(result) == 0 { + return nil, nil + } + return result, nil +} + func TestFixtureFetcherReturnsSnapshotCopy(t *testing.T) { fixtures := DefaultFixtures() fetcher := &FixtureFetcher{Snapshot: fixtures} @@ -758,3 +779,37 @@ func TestRecordsProjectDiskTemperatureAggregatesIntoCanonicalMetadata(t *testing t.Fatal("expected sda physical disk record") } + +func TestProviderPhysicalDiskTemperatureHistoryUsesCanonicalMetricIDs(t *testing.T) { + fixtures := DefaultFixtures() + now := time.Date(2026, 3, 29, 20, 0, 0, 0, time.UTC) + fetcher := &controllableStubFetcher{ + snapshot: &fixtures, + diskHistory: map[string][]TimeSeriesPoint{ + "sda": { + {Timestamp: now.Add(-2 * time.Hour), Value: 30}, + {Timestamp: now.Add(-1 * time.Hour), Value: 32}, + {Timestamp: now, Value: 34}, + }, + }, + } + provider := NewLiveProvider(fetcher) + if err := provider.Refresh(context.Background()); err != nil { + t.Fatalf("Refresh() error = %v", err) + } + + history, err := provider.PhysicalDiskTemperatureHistory(context.Background(), 4*time.Hour) + if err != nil { + t.Fatalf("PhysicalDiskTemperatureHistory() error = %v", err) + } + points, ok := history["ZL0A1234"] + if !ok { + t.Fatalf("expected canonical disk metric id ZL0A1234, got keys %#v", history) + } + if len(points) != 3 { + t.Fatalf("expected 3 temperature history points, got %+v", points) + } + if points[len(points)-1].Value != 34 { + t.Fatalf("expected latest point value 34, got %+v", points) + } +} diff --git a/internal/truenas/types.go b/internal/truenas/types.go index c8074d191..c91c289eb 100644 --- a/internal/truenas/types.go +++ b/internal/truenas/types.go @@ -81,6 +81,13 @@ type DiskTemperatureAggregate struct { MaxCelsius float64 } +// TimeSeriesPoint stores one provider-native metric point before it is mapped +// onto the canonical monitoring/chart surface. +type TimeSeriesPoint struct { + Timestamp time.Time + Value float64 +} + // Alert mirrors a TrueNAS alert listing entry. type Alert struct { ID string