Merge pull request #2168 from rcourtman/maintainer/20260922T184021Z

Keep TrueNAS telemetry available when optional graphs fail

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

@@ -103,6 +103,32 @@ and projection proofs, not native CORE appliance acceptance or reporter
 confirmation.
 
 
+**Legacy reporting failure isolation and memory components — issue #2077**
+
+Live telemetry and history retain the legacy REST graph/query contract and the
+single successful-batch fast path. A batch rejected with HTTP 400, 422 or 500
+falls back to one request per graph (at most six calls including the batch).
+A failed optional graph must not discard successful CPU, memory or ARC graphs.
+Authentication, rate-limit, missing-endpoint, service-unavailable, decoding,
+transport and cancellation errors are not graph fallback triggers; these also
+stop an in-progress split. If every graph fails, the original batch error is
+returned. Missing memory readings remain unavailable, never fabricated usage.
+FreeBSD `active` pages are not a total-used-memory alias. Without an explicit
+used series, history derives usage from the known capacity and reported free
+memory, subtracting separately reported ARC as the live projection does.
+
+`TestRESTReportingGraphFailurePreservesSnapshotTelemetry`,
+`TestRESTReportingGraphFailureBoundaries`,
+`TestRESTReportingPartialAndMalformedResponses` and
+`TestReportingActiveMemoryIsNotTotalUsed` in
+`internal/truenas/client_test.go` exercise request-validating synthetic REST
+responses, repeated full snapshots, CPU/free/ARC projections, matching history,
+partial data, bounded errors and cancellation. They do not establish the native
+CORE request schema, row timestamps/units, appliance acceptance or the reporter's
+exact failure. Native response evidence remains required before claiming that
+this repair resolves the reported telemetry journey.
+
+
 **Availability backfill preserves concurrent discovery changes (7 September 2026)**
 
 The backfill List snapshot is a work list, not an authoritative record to save.

internal/api/unified_agent_offline_preflight_test.go

@@ -0,0 +1,151 @@
+package api
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// This is an offline HTTP/preflight contract, NOT native pfSense acceptance.
+// Only uname is simulated. Bash, curl, the served installer and download
+// handler are real; the agent bytes and signature sidecars are fixtures and
+// are never executed or claimed to be cryptographically qualified artifacts.
+func TestFreeBSDOfflineAgentPreflight(t *testing.T) {
+	for _, tc := range []struct {
+		name             string
+		binary, sidecars bool
+		wantExit         int
+	}{
+		{"bundled", true, true, 0},
+		{"missing_binary", false, false, 11},
+		{"missing_sidecars", true, false, 11},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			dir := t.TempDir()
+			binDir := setupTempPulseBin(t)
+			payload := validTestUnifiedAgentBinary("offline-freebsd-fixture")
+			binaryPath := filepath.Join(binDir, "pulse-agent-freebsd-amd64")
+			write := func(path string, data []byte) {
+				t.Helper()
+				if err := os.WriteFile(path, data, 0700); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if tc.binary {
+				write(binaryPath, payload)
+			}
+			if tc.sidecars {
+				write(binaryPath+".sig", []byte("synthetic-signature"))
+				write(binaryPath+".sshsig", []byte("synthetic-ssh-signature"))
+			}
+			var externalAttempts, healthRequests, downloadRequests atomic.Int32
+			router := &Router{
+				projectRoot: dir, serverVersion: "v6.4.5-rc.1",
+				checksumCache: make(map[string]checksumCacheEntry),
+				installScriptClient: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+					externalAttempts.Add(1)
+					if req.URL.Host != "github.com" || !strings.HasSuffix(req.URL.Path, "/pulse-agent-freebsd-amd64") {
+						t.Errorf("unexpected fallback destination: %s", req.URL)
+					}
+					return nil, errors.New("fixture: server has no external connectivity")
+				})},
+			}
+			script, err := filepath.Abs(filepath.Join("..", "..", "scripts", "install.sh"))
+			if err != nil {
+				t.Fatal(err)
+			}
+			mux := http.NewServeMux()
+			mux.HandleFunc("/install.sh", func(w http.ResponseWriter, r *http.Request) {
+				handleDownloadInstallScriptCommon(w, r, "dev", filepath.Join(dir, "absent.sh"), script, "install.sh", "text/x-shellscript")
+			})
+			mux.HandleFunc("/api/health", func(w http.ResponseWriter, r *http.Request) { healthRequests.Add(1); w.WriteHeader(http.StatusOK) })
+			mux.HandleFunc("/download/pulse-agent", func(w http.ResponseWriter, r *http.Request) {
+				downloadRequests.Add(1)
+				if r.URL.Query().Get("arch") != "freebsd-amd64" {
+					t.Errorf("wrong arch: %s", r.URL)
+				}
+				router.handleDownloadUnifiedAgent(w, r)
+			})
+			server := httptest.NewServer(mux)
+			defer server.Close()
+
+			downloadedScript := filepath.Join(dir, "install.sh")
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+			fetch := exec.CommandContext(ctx, "curl", "--noproxy", "*", "-fsS", server.URL+"/install.sh", "-o", downloadedScript)
+			if out, err := fetch.CombinedOutput(); err != nil {
+				t.Fatalf("fetch installer: %v\n%s", err, out)
+			}
+			original, err := os.ReadFile(script)
+			if err != nil {
+				t.Fatal(err)
+			}
+			downloaded, err := os.ReadFile(downloadedScript)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !bytes.Equal(original, downloaded) {
+				t.Fatal("served installer differs from source")
+			}
+			shimDir := filepath.Join(dir, "shim")
+			if err := os.Mkdir(shimDir, 0700); err != nil {
+				t.Fatal(err)
+			}
+			write(filepath.Join(shimDir, "uname"), []byte("#!/bin/sh\ncase \"$1\" in -s) echo FreeBSD;; -m) echo amd64;; *) exec /usr/bin/uname \"$@\";; esac\n"))
+			cmd := exec.CommandContext(ctx, "bash", downloadedScript, "--url", server.URL, "--preflight-only", "--output", "json", "--non-interactive")
+			cmd.Env = append(os.Environ(), "PATH="+shimDir+":"+os.Getenv("PATH"), "NO_PROXY=*", "no_proxy=*")
+			out, runErr := cmd.CombinedOutput()
+			exitCode := 0
+			if runErr != nil {
+				var ee *exec.ExitError
+				if !errors.As(runErr, &ee) {
+					t.Fatalf("preflight: %v\n%s", runErr, out)
+				}
+				exitCode = ee.ExitCode()
+			}
+			t.Logf("preflight exit=%d; external attempts=%d\n%s", exitCode, externalAttempts.Load(), out)
+			if exitCode != tc.wantExit {
+				t.Fatalf("exit=%d, want %d", exitCode, tc.wantExit)
+			}
+			if healthRequests.Load() != 1 || downloadRequests.Load() != 1 {
+				t.Fatalf("health/download requests = %d/%d", healthRequests.Load(), downloadRequests.Load())
+			}
+			if tc.wantExit != 0 {
+				if externalAttempts.Load() != 1 || !strings.Contains(string(out), `"code":"agent_download_unavailable"`) {
+					t.Fatal("missing artifact must fail visibly at preflight")
+				}
+				return
+			}
+			if externalAttempts.Load() != 0 || !strings.Contains(string(out), `"code":"agent_download_available"`) {
+				t.Fatal("bundled artifact must preflight without external fetch")
+			}
+			response, err := server.Client().Get(server.URL + "/download/pulse-agent?arch=freebsd-amd64")
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer response.Body.Close()
+			data, err := io.ReadAll(response.Body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if response.StatusCode != http.StatusOK || !bytes.Equal(data, payload) || response.Header.Get(checksumHeaderName) != fmt.Sprintf("%x", sha256.Sum256(payload)) {
+				t.Fatal("local payload/checksum mismatch")
+			}
+			if externalAttempts.Load() != 0 {
+				t.Fatal("local download attempted external access")
+			}
+		})
+	}
+}

internal/truenas/client.go

@@ -283,7 +283,7 @@ func (c *Client) getSystemTelemetryREST(ctx context.Context) (*SystemInfo, error
 	if start <= 0 {
 		start = end
 	}
-	response, err := c.getReportingDataREST(ctx, legacyRESTReportingGraphs(), map[string]any{
+	response, err := c.getLegacySystemReportingData(ctx, map[string]any{
 		"aggregate": false,
 		"start":     start,
 		"end":       end,
@@ -360,6 +360,47 @@ func legacyRESTReportingGraphs() []map[string]any {
 	}
 }
 
+// getLegacySystemReportingData keeps a rejected optional graph from discarding
+// usable CPU/memory readings. Keep the successful batch fast path; only split
+// graph-validation/server errors, never authentication, rate-limit, endpoint,
+// transport or cancellation failures. The query and graph set remain unchanged.
+func (c *Client) getLegacySystemReportingData(ctx context.Context, query map[string]any) ([]trueNASReportingGetDataResponse, error) {
+	graphs := legacyRESTReportingGraphs()
+	response, err := c.getReportingDataREST(ctx, graphs, query)
+	if err == nil || !isReportingGraphFailure(err) {
+		return response, err
+	}
+	batchErr := err
+	var collected []trueNASReportingGetDataResponse
+	for _, graph := range graphs {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		response, err := c.getReportingDataREST(ctx, []map[string]any{graph}, query)
+		if err != nil {
+			if !isReportingGraphFailure(err) {
+				return nil, err
+			}
+			continue
+		}
+		collected = append(collected, response...)
+	}
+	if len(collected) == 0 {
+		return nil, batchErr
+	}
+	return collected, nil
+}
+
+func isReportingGraphFailure(err error) bool {
+	var apiErr *APIError
+	if !errors.As(err, &apiErr) {
+		return false
+	}
+	return apiErr.StatusCode == http.StatusBadRequest ||
+		apiErr.StatusCode == http.StatusUnprocessableEntity ||
+		apiErr.StatusCode == http.StatusInternalServerError
+}
+
 // getReportingDataREST issues reporting.get_data over the REST v2.0 transport.
 // The middleware method takes positional parameters (graphs, query); REST
 // serves it as a POST body keyed by parameter name.
@@ -406,7 +447,7 @@ func (c *Client) getSystemMetricHistoryREST(ctx context.Context, duration time.D
 	if start <= 0 {
 		start = end
 	}
-	response, err := c.getReportingDataREST(ctx, legacyRESTReportingGraphs(), map[string]any{
+	response, err := c.getLegacySystemReportingData(ctx, map[string]any{
 		"aggregate": false,
 		"start":     start,
 		"end":       end,
@@ -3171,10 +3212,13 @@ func parseSystemMetricHistory(responses []trueNASReportingGetDataResponse) *Syst
 					history.CPUPercent = appendTimeSeriesPoint(history.CPUPercent, timestamp, value)
 				}
 			case "memory":
+				// FreeBSD active pages are only one used-memory component, not
+				// total usage. Without an explicit used series, let the provider
+				// derive usage from system capacity, free memory and ARC.
 				if value, ok := parseSystemMemoryPercent(values); ok {
 					history.MemoryPercent = appendTimeSeriesPoint(history.MemoryPercent, timestamp, value)
 				}
-				if value, ok := pickReportingValue(values, "used", "memory_used", "used_bytes", "active", "memory"); ok {
+				if value, ok := pickReportingValue(values, "used", "memory_used", "used_bytes", "memory"); ok {
 					history.MemoryUsedBytes = appendTimeSeriesPoint(history.MemoryUsedBytes, timestamp, value)
 				}
 				if value, ok := pickReportingValue(values, "available", "free", "available_bytes", "free_bytes"); ok {
@@ -3442,7 +3486,7 @@ func parseSystemMemoryPercent(values map[string]float64) (float64, bool) {
 	if value, ok := pickReportingValue(values, "usage", "percent", "used_percent", "memory_percent"); ok {
 		return value, true
 	}
-	used, hasUsed := pickReportingValue(values, "used", "memory_used", "used_bytes", "active", "memory")
+	used, hasUsed := pickReportingValue(values, "used", "memory_used", "used_bytes", "memory")
 	total, hasTotal := pickReportingValue(values, "total", "memory_total", "total_bytes", "physical_memory_total")
 	if hasUsed && hasTotal && total > 0 {
 		return (used / total) * 100, true

internal/truenas/client_test.go

@@ -2113,3 +2113,171 @@ func TestTrueNASMemoryWithoutAvailableIsNotReportedAsUsed(t *testing.T) {
 		t.Fatalf("unexpected agent memory: %+v", agent.Memory)
 	}
 }
+
+// This is a request-validating synthetic transport, not an appliance capture.
+// Native CORE request/row schema acceptance remains a separate proof obligation.
+type isolatedReportingTransport struct {
+	routes      alertArgsTransport
+	calls       int
+	status      int
+	cancel      context.CancelFunc
+	rejectGraph string
+	graphStatus int
+	body        string
+}
+
+func (s *isolatedReportingTransport) RoundTrip(r *http.Request) (*http.Response, error) {
+	if r.URL.Path != "/api/v2.0/reporting/get_data" {
+		return s.routes.RoundTrip(r)
+	}
+	s.calls++
+	var request struct {
+		Graphs []map[string]any `json:"graphs"`
+		Query  map[string]any   `json:"query"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+		return nil, err
+	}
+	status, body := s.status, `[]`
+	if r.Method != http.MethodPost || len(request.Graphs) == 0 || request.Query["aggregate"] != false || request.Query["end"].(float64) <= request.Query["start"].(float64) {
+		return nil, fmt.Errorf("invalid reporting request")
+	}
+	if status == 0 {
+		status = http.StatusUnprocessableEntity
+		body = `{"error":"graph requires a device identifier"}`
+		if len(request.Graphs) == 1 {
+			switch request.Graphs[0]["name"] {
+			case "cpu":
+				status, body = 200, `[{"name":"cpu","legend":["user","nice","system","interrupt","idle"],"data":[[1789000060,8,0,3,1,88]]}]`
+			case "memory":
+				status, body = 200, `[{"name":"memory","legend":["active","inactive","wired","laundry","free"],"data":[[1789000060,1073741824,2147483648,11811160064,0,2147483648]]}]`
+			case "arcsize":
+				status, body = 200, `[{"name":"arcsize","legend":["size"],"data":[[1789000060,8589934592]]}]`
+			}
+		}
+	}
+	if len(request.Graphs) == 1 && request.Graphs[0]["name"] == s.rejectGraph {
+		status, body = s.graphStatus, `{"error":"synthetic graph failure"}`
+	}
+	if s.body != "" {
+		body = s.body
+	}
+	if s.cancel != nil {
+		s.cancel()
+	}
+	return (alertArgsTransport{r.URL.Path: apiResponse{status: status, body: body}}).RoundTrip(r)
+}
+
+func TestRESTReportingGraphFailurePreservesSnapshotTelemetry(t *testing.T) {
+	routes := alertArgsTransport(defaultAPIResponses())
+	routes["/api/v2.0/system/info"] = apiResponse{body: `{"hostname":"synthetic-core","version":"TrueNAS-13.0-U6.1","physmem":17179869184}`}
+	transport := &isolatedReportingTransport{routes: routes}
+	client := newLegacyRESTReportingClient(t, routes)
+	client.httpClient.Transport = transport
+	for cycle := 0; cycle < 2; cycle++ {
+		snapshot, err := client.FetchSnapshot(context.Background())
+		if err != nil {
+			t.Fatal(err)
+		}
+		metrics := metricsFromTrueNASSystem(snapshot.System, 0, 0)
+		if metrics.CPU == nil || metrics.CPU.Percent != 12 {
+			t.Fatalf("CPU lost: %+v", metrics.CPU)
+		}
+		if metrics.Memory == nil || metrics.Memory.Used == nil || *metrics.Memory.Used != 6<<30 || metrics.Memory.Percent != 37.5 {
+			t.Fatalf("memory lost or fabricated: %+v", metrics.Memory)
+		}
+		agent := agentDataFromTrueNASSystem("synthetic-core", snapshot.System, nil, nil, false, "", false, "")
+		if agent.Memory == nil || agent.Memory.UsageUnavailable || agent.Memory.Used != 6<<30 || agent.Memory.Free != 2<<30 || agent.Memory.Cache != 8<<30 {
+			t.Fatalf("agent memory: %+v", agent.Memory)
+		}
+		if len(snapshot.Pools) == 0 || len(snapshot.Disks) == 0 {
+			t.Fatal("inventory lost")
+		}
+	}
+	if transport.calls != 12 {
+		t.Fatalf("reporting calls = %d, want bounded batch + five graphs per snapshot", transport.calls)
+	}
+	history, err := client.GetSystemMetricHistory(context.Background(), time.Hour)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if history == nil || len(history.CPUPercent) != 1 || history.CPUPercent[0].Value != 12 {
+		t.Fatalf("CPU history lost: %+v", history)
+	}
+	memory := systemMemoryPercentHistory(history, 16<<30)
+	if len(memory) != 1 || memory[0].Value != 37.5 {
+		t.Fatalf("history must match live free/ARC memory, got %+v", memory)
+	}
+}
+
+func TestRESTReportingGraphFailureBoundaries(t *testing.T) {
+	for _, status := range []int{200, 400, 401, 403, 404, 422, 429, 500, 503} {
+		t.Run(strconv.Itoa(status), func(t *testing.T) {
+			transport := &isolatedReportingTransport{status: status}
+			client := newLegacyRESTReportingClient(t, nil)
+			client.httpClient.Transport = transport
+			if _, err := client.GetSystemTelemetry(context.Background()); err == nil {
+				t.Fatal("all failed graphs must return an error")
+			}
+			want := 1
+			if status == 400 || status == 422 || status == 500 {
+				want = 6
+			}
+			if transport.calls != want {
+				t.Fatalf("calls = %d, want %d", transport.calls, want)
+			}
+		})
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	transport := &isolatedReportingTransport{status: 422, cancel: cancel}
+	client := newLegacyRESTReportingClient(t, nil)
+	client.httpClient.Transport = transport
+	if _, err := client.GetSystemTelemetry(ctx); err == nil || transport.calls != 1 {
+		t.Fatalf("cancellation must stop fallback, calls=%d err=%v", transport.calls, err)
+	}
+}
+
+func TestReportingActiveMemoryIsNotTotalUsed(t *testing.T) {
+	for _, values := range []map[string]float64{
+		{"active": 1 << 30, "free": 2 << 30, "total": 16 << 30},
+		{"active": 1 << 30, "available": 2 << 30, "total": 16 << 30},
+	} {
+		percent, ok := parseSystemMemoryPercent(values)
+		if !ok || percent != 87.5 {
+			t.Fatalf("active pages treated as total used: %v, %v", percent, ok)
+		}
+	}
+	if _, ok := parseSystemMemoryPercent(map[string]float64{"active": 1 << 30, "total": 16 << 30}); ok {
+		t.Fatal("active alone cannot establish total usage")
+	}
+}
+
+func TestRESTReportingPartialAndMalformedResponses(t *testing.T) {
+	for _, status := range []int{422, 401, 429} {
+		transport := &isolatedReportingTransport{rejectGraph: "memory", graphStatus: status}
+		client := newLegacyRESTReportingClient(t, nil)
+		client.httpClient.Transport = transport
+		telemetry, err := client.GetSystemTelemetry(context.Background())
+		if status == 422 {
+			if err != nil || telemetry.CPUPercent != 12 || telemetry.MemoryAvailableBytes != 0 {
+				t.Fatalf("partial readings lost/fabricated: %+v %v", telemetry, err)
+			}
+			telemetry.MemoryTotalBytes = 16 << 30
+			if metricsFromTrueNASSystem(*telemetry, 0, 0).Memory != nil {
+				t.Fatal("missing memory became usage")
+			}
+		} else if err == nil || transport.calls != 3 {
+			t.Fatalf("terminal error after successful CPU must stop collection: status=%d calls=%d err=%v", status, transport.calls, err)
+		}
+	}
+	for _, body := range []string{`{`, legacyRESTReportingMemoryBody()} {
+		transport := &isolatedReportingTransport{status: 200, body: body}
+		client := newLegacyRESTReportingClient(t, nil)
+		client.httpClient.Transport = transport
+		_, err := client.GetSystemTelemetry(context.Background())
+		if (err != nil) != (body == "{") || transport.calls != 1 {
+			t.Fatalf("batch/decode result: calls=%d err=%v", transport.calls, err)
+		}
+	}
+}