Read native TrueNAS SCALE aggregate CPU usage, including idle zero

The reporting.realtime payload in #3158 places the actual aggregate at fields.cpu.cpu.usage alongside per-core entries. Accept that reading without treating temperature, absent/null usage or a core average as a sample. Existing flat and nested usage/total shapes keep their precedence.

Cover parser presence, legacy compatibility and repeated real TLS JSON-RPC subscription through provider/registry/wire projection, with unavailable and recovery controls. Update the monitoring contract in the same commit.

Change-source: pulse-maintainer

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -17,6 +17,20 @@
 
 ## Purpose
 
+### Native TrueNAS SCALE CPU uses the reported aggregate
+
+`reporting.realtime` may place its aggregate at `fields.cpu.cpu.usage`,
+beside per-core records. That aggregate is an available CPU sample even when
+its value is zero. Missing, null or invalid usage stays unavailable; per-core
+values and aggregate temperature cannot substitute for it. Previously supported
+flat and nested usage/total shapes retain their existing precedence.
+
+`TestRealtimeSCALEAggregateCPU` covers aggregate presence and legacy shapes;
+`TestJSONRPCSCALEAggregateCPUProjection` carries repeated JSON-RPC samples
+through system merge, provider and canonical resource JSON, including idle,
+nonzero, unavailable and recovery. The fixture is not installed-appliance or
+reporter recovery evidence. REST/CORE collection and CPU history are unchanged.
+
 ### Failed guest snapshot reads do not establish removal
 
 A snapshot listing is accepted only when its API call succeeds. In particular,

internal/truenas/client.go

@@ -3272,6 +3272,12 @@ func parseSystemTelemetry(fields map[string]any, intervalSeconds int, collectedA
 			cpuPercent, hasCPU = readFloatValueAny(usage, "percent", "value", "usage")
 		}
 	}
+	if !hasCPU {
+		// SCALE reporting.realtime groups its aggregate alongside cpu0, cpu1,
+		// etc. Read that aggregate directly; cores and temperature are not
+		// substitutes for a missing usage sample. Explicit idle zero is valid.
+		cpuPercent, hasCPU = readFloatValueAny(readMapAny(cpu, "cpu"), "usage")
+	}
 	telemetry.CPUPercent = cpuPercent
 	telemetry.Telemetry.CPU = hasCPU
 

internal/truenas/client_test.go

@@ -2494,6 +2494,8 @@ func TestRealtimeObservationPresence(t *testing.T) {
 		{"empty", nil, SystemTelemetryAvailability{}},
 		{"cpu_zero", map[string]any{"cpu": map[string]any{"usage": 0}}, SystemTelemetryAvailability{CPU: true}},
 		{"cpu_nested_zero", map[string]any{"cpu": map[string]any{"usage": map[string]any{"percent": 0}}}, SystemTelemetryAvailability{CPU: true}},
+		{"cpu_scale_aggregate_zero", map[string]any{"cpu": map[string]any{"cpu": map[string]any{"usage": 0, "temp": 36}}}, SystemTelemetryAvailability{CPU: true}},
+		{"cpu_scale_aggregate_null", map[string]any{"cpu": map[string]any{"cpu": map[string]any{"usage": nil, "temp": 36}}}, SystemTelemetryAvailability{}},
 		{"cpu_null", map[string]any{"cpu": map[string]any{"usage": nil}}, SystemTelemetryAvailability{}},
 		{"capacity_only", map[string]any{"memory": map[string]any{"total": 16}}, SystemTelemetryAvailability{}},
 		{"free_zero", map[string]any{"memory": map[string]any{"total": 16, "available": 0}}, SystemTelemetryAvailability{Memory: true}},

internal/truenas/realtime_aggregate_test.go

@@ -0,0 +1,158 @@
+package truenas
+
+import (
+	"context"
+	"encoding/json"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources"
+)
+
+func TestRealtimeSCALEAggregateCPU(t *testing.T) {
+	for _, tc := range []struct {
+		name, fields string
+		available    bool
+		percent      float64
+	}{
+		{"reported-idle", `{"cpu":{"cpu":{"usage":0,"temp":36}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`, true, 0},
+		{"reported-active", `{"cpu":{"cpu":{"usage":1,"temp":45}}}`, true, 1},
+		{"fractional", `{"cpu":{"cpu":{"usage":12.5}}}`, true, 12.5},
+		{"full", `{"cpu":{"cpu":{"usage":100}}}`, true, 100},
+		{"not-a-core-average", `{"cpu":{"cpu":{"usage":1},"cpu0":{"usage":99},"cpu31":{"usage":75}}}`, true, 1},
+		{"no-cpu", `{}`, false, 0},
+		{"cores-only", `{"cpu":{"cpu0":{"usage":99},"cpu31":{"usage":75}}}`, false, 0},
+		{"temperature-only", `{"cpu":{"cpu":{"temp":36}}}`, false, 0},
+		{"null-aggregate", `{"cpu":{"cpu":null,"cpu0":{"usage":99}}}`, false, 0},
+		{"null-usage", `{"cpu":{"cpu":{"usage":null,"temp":36}}}`, false, 0},
+		{"invalid-usage", `{"cpu":{"cpu":{"usage":"unavailable"},"cpu0":{"usage":99}}}`, false, 0},
+		{"boolean-usage", `{"cpu":{"cpu":{"usage":false}}}`, false, 0},
+		{"flat-compatibility", `{"cpu":{"usage":12.5}}`, true, 12.5},
+		{"nested-usage-compatibility", `{"cpu":{"usage":{"percent":12.5}}}`, true, 12.5},
+		{"nested-total-compatibility", `{"cpu":{"total":{"usage":12.5}}}`, true, 12.5},
+		{"flat-zero-precedence", `{"cpu":{"usage":0,"cpu":{"usage":75}}}`, true, 0},
+		{"nested-zero-precedence", `{"cpu":{"usage":{"percent":0},"cpu":{"usage":75}}}`, true, 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var fields map[string]any
+			if err := json.Unmarshal([]byte(tc.fields), &fields); err != nil {
+				t.Fatal(err)
+			}
+			telemetry := parseSystemTelemetry(fields, 2, time.Unix(1792796400, 0).UTC())
+			if telemetry.Telemetry.CPU != tc.available || telemetry.CPUPercent != tc.percent {
+				t.Fatalf("CPU available=%v percent=%v, want %v/%v", telemetry.Telemetry.CPU, telemetry.CPUPercent, tc.available, tc.percent)
+			}
+		})
+	}
+}
+
+// Use the real TLS websocket subscription, not a direct parser invocation.
+// Samples reuse one ordinary RPC session and acknowledge every unsubscribe.
+func scaleAggregateTelemetryClient(t *testing.T, samples []string) *Client {
+	t.Helper()
+	var subscribes, unsubscribes atomic.Int32
+	fixture := newProtocolFixture(t, func(_ int, request trueNASRPCRequest) protocolFixtureReply {
+		switch request.Method {
+		case "auth.login_ex":
+			return protocolFixtureReply{result: map[string]any{"response_type": "SUCCESS"}}
+		case "reporting.get_data":
+			return protocolFixtureReply{result: []any{}}
+		case "core.subscribe":
+			collection := request.Params.([]any)[0].(string)
+			if !strings.HasPrefix(collection, "reporting.realtime:") {
+				t.Errorf("unexpected subscription %s", collection)
+				return protocolFixtureReply{close: true}
+			}
+			index := int(subscribes.Add(1)) - 1
+			if index >= len(samples) {
+				t.Error("unexpected repeated telemetry read")
+				return protocolFixtureReply{close: true}
+			}
+			var fields map[string]any
+			if err := json.Unmarshal([]byte(samples[index]), &fields); err != nil {
+				t.Error(err)
+				return protocolFixtureReply{close: true}
+			}
+			return protocolFixtureReply{result: "fixture-realtime", notifications: []protocolFixtureNotification{{
+				method: "collection_update", params: map[string]any{"collection": collection, "fields": fields},
+			}}}
+		case "core.unsubscribe":
+			unsubscribes.Add(1)
+			return protocolFixtureReply{result: nil}
+		default:
+			t.Errorf("unexpected RPC method %s", request.Method)
+			return protocolFixtureReply{close: true}
+		}
+	}, nil)
+	client := protocolFixtureClient(t, fixture.server.URL, ClientConfig{APIKey: "readonly-fixture-key", Username: "readonly-fixture", Timeout: 5 * time.Second})
+	t.Cleanup(func() {
+		if fixture.restRequests.Load() != 0 || fixture.sessions.Load() != 1 || int(subscribes.Load()) != len(samples) || subscribes.Load() != unsubscribes.Load() {
+			t.Errorf("sessions=%d REST=%d subscribe/unsubscribe=%d/%d, want 1/0/%d/%d", fixture.sessions.Load(), fixture.restRequests.Load(), subscribes.Load(), unsubscribes.Load(), len(samples), len(samples))
+		}
+	})
+	return client
+}
+
+func scaleAggregateResource(t *testing.T, telemetry *SystemInfo) unifiedresources.Resource {
+	t.Helper()
+	system := systemInfoFromResponse(systemInfoResponse{Hostname: "scale-cpu-fixture", Version: "TrueNAS-SCALE-25.10.7", SystemSerial: "synthetic-3158", Cores: 32, Physmem: 67374374912})
+	mergeSystemTelemetry(system, telemetry)
+	snapshot := FixtureSnapshot{System: *system, CollectedAt: telemetry.CollectedAt}
+	provider := NewProvider(snapshot)
+	records := provider.RecordsFromSnapshot(&snapshot)
+	registry := unifiedresources.NewRegistry(unifiedresources.NewMemoryStore())
+	registry.IngestRecords(unifiedresources.SourceTrueNAS, records)
+	resources := registry.List()
+	if len(resources) != 1 || resources[0].Type != unifiedresources.ResourceTypeAgent {
+		t.Fatalf("system resource projection lost identity: %+v", resources)
+	}
+	return resources[0]
+}
+
+func TestJSONRPCSCALEAggregateCPUProjection(t *testing.T) {
+	previous := IsFeatureEnabled()
+	SetFeatureEnabled(true)
+	t.Cleanup(func() { SetFeatureEnabled(previous) })
+	// Only the first aggregate/memory fragment and second aggregate values
+	// come from #3158. Other frames and per-core conflicts are synthetic controls.
+	samples := []string{
+		`{"cpu":{"cpu":{"usage":0,"temp":36}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`,
+		`{"cpu":{"cpu":{"usage":1,"temp":45},"cpu0":{"usage":99}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`,
+		`{"cpu":{"cpu":{"temp":36},"cpu0":{"usage":99}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`,
+		`{"cpu":{"cpu":{"usage":null},"cpu0":{"usage":99}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`,
+		`{"cpu":{"cpu":{"usage":0}},"memory":{"physical_memory_total":67374374912,"physical_memory_available":4122996736}}`,
+	}
+	client := scaleAggregateTelemetryClient(t, samples)
+	for index, want := range []struct {
+		available bool
+		percent   float64
+	}{{true, 0}, {true, 1}, {false, 0}, {false, 0}, {true, 0}} {
+		telemetry, err := client.GetSystemTelemetry(context.Background())
+		if err != nil {
+			t.Fatal(err)
+		}
+		resource := scaleAggregateResource(t, telemetry)
+		if resource.Metrics == nil || (resource.Metrics.CPU != nil) != want.available || resource.Metrics.Memory == nil {
+			t.Fatalf("frame %d invented or lost CPU/memory: %+v", index, resource.Metrics)
+		}
+		if cpu := resource.Metrics.CPU; cpu != nil && (cpu.Value != want.percent || cpu.Percent != want.percent || cpu.Source != unifiedresources.SourceTrueNAS) {
+			t.Fatalf("frame %d lost the actual aggregate: %+v", index, cpu)
+		}
+		body, err := json.Marshal(resource)
+		if err != nil {
+			t.Fatal(err)
+		}
+		var wire struct {
+			Metrics struct {
+				CPU *struct {
+					Value float64 `json:"value"`
+				} `json:"cpu"`
+			} `json:"metrics"`
+		}
+		if err := json.Unmarshal(body, &wire); err != nil || (wire.Metrics.CPU != nil) != want.available || wire.Metrics.CPU != nil && wire.Metrics.CPU.Value != want.percent {
+			t.Fatalf("frame %d wire availability/value changed: %s, %v", index, body, err)
+		}
+	}
+}