From a319ff6a5f7ceca2465e723f378f58d20980e623 Mon Sep 17 00:00:00 2001 From: "Patrick W. Healy" Date: Thu, 17 Sep 2026 14:06:28 +0000 Subject: [PATCH] net CLI: list summaries and explicitly retrieve refreshed node diagnostics Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: d2243398-6c36-4c3d-969e-7ed7bfb5b459 --- cmd/kubectl-unbounded/app/net/controller.go | 48 +- .../app/net/controller_test.go | 260 +++++++++++ .../app/net/detail_client.go | 209 +++++++++ .../app/net/detail_client_test.go | 143 ++++++ .../app/net/detail_poll_test.go | 190 ++++++++ cmd/kubectl-unbounded/app/net/node.go | 270 ++++++++---- .../app/net/node_detail_command_test.go | 412 ++++++++++++++++++ cmd/kubectl-unbounded/app/net/node_render.go | 45 +- .../app/net/summary_poll_test.go | 168 +++++++ .../app/net/summary_types.go | 24 +- cmd/kubectl-unbounded/app/net/watch.go | 193 ++++++-- 11 files changed, 1790 insertions(+), 172 deletions(-) create mode 100644 cmd/kubectl-unbounded/app/net/controller_test.go create mode 100644 cmd/kubectl-unbounded/app/net/detail_client.go create mode 100644 cmd/kubectl-unbounded/app/net/detail_client_test.go create mode 100644 cmd/kubectl-unbounded/app/net/detail_poll_test.go create mode 100644 cmd/kubectl-unbounded/app/net/node_detail_command_test.go create mode 100644 cmd/kubectl-unbounded/app/net/summary_poll_test.go diff --git a/cmd/kubectl-unbounded/app/net/controller.go b/cmd/kubectl-unbounded/app/net/controller.go index 53b3f6785..df40c015d 100644 --- a/cmd/kubectl-unbounded/app/net/controller.go +++ b/cmd/kubectl-unbounded/app/net/controller.go @@ -119,7 +119,7 @@ func newControllerLogsCommand(rt *pluginRuntime) *cobra.Command { return cmd } -// newControllerStatusJSONCommand dumps raw /status/json from the controller. +// newControllerStatusJSONCommand exports the controller's cluster overview. func newControllerStatusJSONCommand(rt *pluginRuntime) *cobra.Command { var pretty bool @@ -127,12 +127,19 @@ func newControllerStatusJSONCommand(rt *pluginRuntime) *cobra.Command { cmd := &cobra.Command{ Use: "status-json", - Short: "Dump raw /status/json from the controller", - Args: cobra.NoArgs, + Short: "Dump summary JSON from the controller", + Long: "Export /status/json, preserving controller metadata and summary fields. " + + "Legacy full-node responses are projected into summaries without diagnostic arrays.", + Args: cobra.NoArgs, RunE: func(cmd *cobra.Command, _ []string) error { opts := nodeStatusFetchFromCommand(cmd).merged(fetch) - status, err := fetchClusterStatus(rt, cmd, opts) + raw, err := fetchClusterStatusRaw(rt, cmd, opts) + if err != nil { + return err + } + + status, err := clusterSummaryJSON(raw) if err != nil { return err } @@ -158,3 +165,36 @@ func newControllerStatusJSONCommand(rt *pluginRuntime) *cobra.Command { return cmd } + +// clusterSummaryJSON preserves unknown fields while removing legacy node details. +func clusterSummaryJSON(raw []byte) (json.RawMessage, error) { + summary, err := decodeClusterSummary(raw) + if err != nil { + return nil, err + } + + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + return nil, err + } + + if _, legacy := fields["nodes"]; !legacy { + return raw, nil + } + + delete(fields, "nodes") + + if _, current := fields["nodeSummaries"]; !current { + fields["nodeSummaries"], err = json.Marshal(summary.NodeSummaries) + if err != nil { + return nil, err + } + + fields["nodeCount"], err = json.Marshal(summary.NodeCount) + if err != nil { + return nil, err + } + } + + return json.Marshal(fields) +} diff --git a/cmd/kubectl-unbounded/app/net/controller_test.go b/cmd/kubectl-unbounded/app/net/controller_test.go new file mode 100644 index 000000000..ac0da6faa --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/controller_test.go @@ -0,0 +1,260 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" +) + +func TestControllerStatusJSONPreservesSummary(t *testing.T) { + nodes := make([]string, 11) + for i := range nodes { + nodes[i] = fmt.Sprintf(`{"name":"node-%d","k8sReady":"Ready","futureNodeField":{"value":9007199254740993}}`, i) + } + + fixture := `{ + "seq":9007199254740993,"nodeCount":11,"siteCount":2,"pullEnabled":false, + "buildInfo":{"commit":"024fd308","futureBuildField":"preserved"}, + "leaderInfo":{"podName":"leader"},"sites":[{"name":"site"}], + "gatewayPools":[],"warnings":["warning"],"errors":["explicit error"], + "futureMetadata":{"precision":9007199254740993}, + "nodeSummaries":[` + strings.Join(nodes, ",") + `]}` + + for _, pretty := range []bool{true, false} { + t.Run(fmt.Sprintf("pretty=%t", pretty), func(t *testing.T) { + var calls atomic.Int32 + + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + assertControllerOverviewRequest(t, r) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, fixture) + })) + defer server.Close() + + cmd := newControllerRootCommand(newControllerStatusTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"status-json", fmt.Sprintf("--pretty=%t", pretty)}) + + if err := cmd.Execute(); err != nil { + t.Fatal(err) + } + + var want, got bytes.Buffer + if err := json.Compact(&want, []byte(fixture)); err != nil { + t.Fatal(err) + } + + if err := json.Compact(&got, out.Bytes()); err != nil { + t.Fatal(err) + } + + if !bytes.Equal(want.Bytes(), got.Bytes()) { + t.Fatalf("summary fields changed:\nwant %s\ngot %s", want.String(), got.String()) + } + + if calls.Load() != 1 { + t.Errorf("export made %d requests, want one overview request", calls.Load()) + } + + if !strings.HasSuffix(out.String(), "\n") || + strings.Contains(strings.TrimSuffix(out.String(), "\n"), "\n") != pretty { + t.Errorf("incorrect pretty=%t formatting: %s", pretty, out.String()) + } + }) + } +} + +func TestControllerStatusJSONLegacyCompatibility(t *testing.T) { + for _, fixture := range []string{ + strings.Replace(legacyOverviewFixture, `"seq":7`, `"buildInfo":{"commit":"old"},"futureMetadata":42,"seq":7`, 1), + `{"nodeSummaries":[{"name":"current","futureNodeField":42}],"nodeCount":1,"nodes":"ignored legacy details"}`, + `{"nodes":[]}`, + `{"nodeSummaries":[],"nodeCount":0}`, + } { + for _, pretty := range []bool{true, false} { + t.Run(fmt.Sprintf("%s/pretty=%t", fixture, pretty), func(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assertControllerOverviewRequest(t, r) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, fixture) + })) + defer server.Close() + + cmd := newControllerRootCommand(newControllerStatusTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"status-json", fmt.Sprintf("--pretty=%t", pretty)}) + + if err := cmd.Execute(); err != nil { + t.Fatal(err) + } + + var fields map[string]json.RawMessage + if err := json.Unmarshal(out.Bytes(), &fields); err != nil { + t.Fatal(err) + } + + if fields["nodeSummaries"] == nil || fields["nodeCount"] == nil { + t.Fatalf("missing summary fields: %s", out.String()) + } + + for _, forbidden := range []string{`"nodes"`, `"peers"`, `"routingTable"`, `"bpfEntries"`, "secret-detail", "ignored legacy details"} { + if strings.Contains(out.String(), forbidden) { + t.Errorf("export leaked %q: %s", forbidden, out.String()) + } + } + + summary, err := decodeClusterSummary(out.Bytes()) + if err != nil { + t.Fatal(err) + } + + if strings.Contains(fixture, `"buildInfo"`) { + if summary.NodeCount != 1 || summary.NodeSummaries[0].PeerCount != 2 || + summary.NodeSummaries[0].FirstError != "not ready" || summary.Seq != 7 { + t.Fatalf("lost projected summary facts: %+v", summary) + } + + var build map[string]string + if err := json.Unmarshal(fields["buildInfo"], &build); err != nil { + t.Fatal(err) + } + + if build["commit"] != "old" || string(fields["futureMetadata"]) != "42" { + t.Fatalf("lost legacy metadata: %s", out.String()) + } + } + + if strings.Contains(fixture, `"current"`) && !strings.Contains(out.String(), `"futureNodeField"`) { + t.Fatalf("lost current summary field: %s", out.String()) + } + }) + } + } +} + +func TestControllerStatusJSONFailures(t *testing.T) { + for _, tc := range []struct { + name string + status int + body string + }{ + {name: "invalid JSON", body: `{`}, + {name: "missing nodes", body: `{}`}, + {name: "null", body: `null`}, + {name: "invalid summaries", body: `{"nodeSummaries":"bad"}`}, + {name: "invalid legacy nodes", body: `{"nodes":"bad"}`}, + {name: "unauthorized", status: http.StatusUnauthorized, body: "Unauthorized"}, + {name: "forbidden", status: http.StatusForbidden, body: "Forbidden"}, + {name: "server error", status: http.StatusInternalServerError, body: "controller failed"}, + } { + for _, pretty := range []bool{true, false} { + t.Run(fmt.Sprintf("%s/pretty=%t", tc.name, pretty), func(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || strings.Contains(r.URL.Path, "/details") { + t.Errorf("failure requested diagnostics: %s %s", r.Method, r.URL) + } + + w.Header().Set("Content-Type", "application/json") + + if tc.status != 0 { + w.WriteHeader(tc.status) + } else { + assertControllerOverviewRequest(t, r) + } + + _, _ = io.WriteString(w, tc.body) + })) + defer server.Close() + + cmd := newControllerRootCommand(newControllerStatusTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SilenceUsage = true + cmd.SetArgs([]string{"status-json", fmt.Sprintf("--pretty=%t", pretty)}) + + wantError := "cluster overview" + if tc.status != 0 { + wantError = "fetch /status/json failed" + } + + if err := cmd.Execute(); err == nil || !strings.Contains(err.Error(), wantError) { + t.Fatalf("failed response error = %v, want %q", err, wantError) + } + + if out.Len() != 0 { + t.Fatalf("failed export wrote output: %s", out.String()) + } + }) + } + } +} + +func TestControllerStatusJSONOutputFailure(t *testing.T) { + server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + assertControllerOverviewRequest(t, r) + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"nodeSummaries":[],"nodeCount":0}`) + })) + defer server.Close() + + for _, pretty := range []bool{true, false} { + cmd := newControllerRootCommand(newControllerStatusTestRuntime(t, server.URL)) + outputErr := errors.New("output closed") + cmd.SetOut(controllerStatusErrorWriter{err: outputErr}) + cmd.SetErr(io.Discard) + cmd.SilenceUsage = true + cmd.SetArgs([]string{"status-json", fmt.Sprintf("--pretty=%t", pretty)}) + + if err := cmd.Execute(); !errors.Is(err, outputErr) { + t.Fatalf("output failure = %v, want %v", err, outputErr) + } + } +} + +type controllerStatusErrorWriter struct { + err error +} + +func (w controllerStatusErrorWriter) Write([]byte) (int, error) { + return 0, w.err +} + +func assertControllerOverviewRequest(t *testing.T, r *http.Request) { + t.Helper() + + if r.Method != http.MethodGet || r.URL.Path != "/apis/status.net.unbounded-cloud.io/v1alpha1/status/json" { + t.Errorf("export requested non-overview endpoint: %s %s", r.Method, r.URL) + } + + if r.Header.Get("Authorization") != "Bearer test-token" { + t.Error("export did not reuse Kubernetes authentication") + } +} + +func newControllerStatusTestRuntime(t *testing.T, serverURL string) *pluginRuntime { + t.Helper() + + rt := newDetailTestRuntime(t, serverURL) + *rt.configFlags.Insecure = true + + return rt +} diff --git a/cmd/kubectl-unbounded/app/net/detail_client.go b/cmd/kubectl-unbounded/app/net/detail_client.go new file mode 100644 index 000000000..c0f9342f2 --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/detail_client.go @@ -0,0 +1,209 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "k8s.io/apimachinery/pkg/util/validation" + "k8s.io/client-go/kubernetes" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +type statusRequest func(context.Context, string, string, []byte) ([]byte, error) + +type nodeDetailClient struct { + request statusRequest + pollInterval time.Duration +} + +func (c nodeDetailClient) fetch(ctx context.Context, nodeName string, forceRefresh bool) (*statusv1alpha1.NodeStatusResponse, error) { + if len(validation.IsDNS1123Subdomain(nodeName)) != 0 { + return nil, fmt.Errorf("invalid node name %q", nodeName) + } + + body, err := json.Marshal(struct { + ForceRefresh bool `json:"forceRefresh"` + }{ForceRefresh: forceRefresh}) + if err != nil { + return nil, err + } + + path := "/status/node/" + nodeName + "/details" + + raw, requestErr := c.request(ctx, http.MethodPost, path, body) + if ctx.Err() != nil { + return nil, ctx.Err() + } + + initial, err := decodeNodeDetailResult(raw, requestErr, nodeName, "") + if err != nil { + return nil, err + } + + if initial.State == statusv1alpha1.NodeDetailComplete { + return validateNodeDetails(initial, time.Now()) + } + + requestCtx, cancel := context.WithDeadline(ctx, initial.Deadline) + defer cancel() + + for { + interval := c.pollInterval + if interval <= 0 { + interval = time.Second + } + + timer := time.NewTimer(interval) + select { + case <-requestCtx.Done(): + timer.Stop() + return nil, fmt.Errorf("node %q detail request canceled or deadline exceeded: %w", nodeName, requestCtx.Err()) + case <-timer.C: + } + + raw, requestErr = c.request(requestCtx, http.MethodGet, path+"?requestId="+url.QueryEscape(initial.RequestID), nil) + if requestCtx.Err() != nil { + return nil, fmt.Errorf("node %q detail request canceled or deadline exceeded: %w", nodeName, requestCtx.Err()) + } + + result, err := decodeNodeDetailResult(raw, requestErr, nodeName, initial.RequestID) + if err != nil { + return nil, err + } + + if result.State == statusv1alpha1.NodeDetailComplete { + return validateNodeDetails(result, time.Now()) + } + + if !result.Deadline.Equal(initial.Deadline) { + return nil, fmt.Errorf("malformed node detail response: request deadline changed") + } + } +} + +func decodeNodeDetailResult(raw []byte, requestErr error, nodeName, requestID string) (statusv1alpha1.NodeDetailResult, error) { + var result statusv1alpha1.NodeDetailResult + + decodeErr := json.Unmarshal(raw, &result) + if requestErr != nil && (decodeErr != nil || result.State == "") { + return result, fmt.Errorf("node %q detail API unavailable or unsupported: %w", nodeName, requestErr) + } + + if decodeErr != nil { + return result, fmt.Errorf("malformed node detail response: %w", decodeErr) + } + + if result.NodeName != nodeName || (requestID != "" && result.RequestID != requestID) { + return result, fmt.Errorf("malformed node detail response: node or request identity mismatch") + } + + switch result.State { + case statusv1alpha1.NodeDetailComplete, statusv1alpha1.NodeDetailPending: + if requestErr != nil { + return result, fmt.Errorf("node detail request failed: %w", requestErr) + } + + if result.State == statusv1alpha1.NodeDetailPending && + (result.RequestID == "" || result.Deadline.IsZero() || result.Details != nil) { + return result, fmt.Errorf("malformed pending node detail response") + } + + return result, nil + case statusv1alpha1.NodeDetailExpired, statusv1alpha1.NodeDetailUnavailable, statusv1alpha1.NodeDetailRetryable: + return result, fmt.Errorf("node %q details %s: %s", nodeName, result.State, result.Error) + default: + return result, fmt.Errorf("malformed or unsupported node detail state %q: %s", result.State, result.Error) + } +} + +func validateNodeDetails(result statusv1alpha1.NodeDetailResult, now time.Time) (*statusv1alpha1.NodeStatusResponse, error) { + details := result.Details + if result.RequestID == "" || details == nil || details.Status == nil || + details.NodeName != result.NodeName || details.RequestID != result.RequestID || + details.Status.NodeInfo.Name != result.NodeName || details.CollectedAt.IsZero() || + details.ReceivedAt.IsZero() || !details.ExpiresAt.After(details.ReceivedAt) || result.Error != "" { + return nil, fmt.Errorf("malformed completed node detail response") + } + + if !details.ExpiresAt.After(now) { + return nil, fmt.Errorf("node %q details expired; request them again", result.NodeName) + } + + if details.Status.FetchError != "" { + return nil, fmt.Errorf("node %q detail collection failed: %s", result.NodeName, details.Status.FetchError) + } + + return details.Status, nil +} + +// newStatusRequest reuses kubectl credentials and authenticated port-forward fallback. +// Each HTTP attempt is bounded independently; the caller owns the overall deadline. +func newStatusRequest(rt *pluginRuntime, opts nodeStatusFetchOptions) (statusRequest, error) { + ns, err := rt.namespace() + if err != nil { + return nil, err + } + + client, err := rt.kubeClient() + if err != nil { + return nil, err + } + + cfg, err := rt.restConfig() + if err != nil { + return nil, err + } + + return func(ctx context.Context, method, path string, body []byte) ([]byte, error) { + attemptCtx, cancel := context.WithTimeout(ctx, opts.timeout) + defer cancel() + + raw, err := requestStatusViaAggregatedAPI(attemptCtx, client, method, path, body) + if err == nil || len(raw) > 0 || ctx.Err() != nil { + return raw, err + } + + fallbackCtx, fallbackCancel := context.WithTimeout(ctx, opts.timeout) + defer fallbackCancel() + + return requestStatusViaPortForward(fallbackCtx, client, cfg, ns, + opts.controllerDeploy, opts.controllerSelector, opts.controllerPort, opts.timeout, + method, path, body) + }, nil +} + +func requestStatusViaAggregatedAPI(ctx context.Context, client *kubernetes.Clientset, method, path string, body []byte) ([]byte, error) { + target, err := url.ParseRequestURI(path) + if err != nil || target.IsAbs() || target.Host != "" { + return nil, fmt.Errorf("invalid controller status path %q", path) + } + + aggregatedPath := target.Path + if strings.HasPrefix(aggregatedPath, "/status/node/") && strings.HasSuffix(aggregatedPath, "/details") { + aggregatedPath = "/nodes/" + strings.TrimPrefix(aggregatedPath, "/status/node/") + } + + request := client.CoreV1().RESTClient().Verb(method). + AbsPath("/apis/status.net.unbounded-cloud.io/v1alpha1" + aggregatedPath) + + for key, values := range target.Query() { + for _, value := range values { + request.Param(key, value) + } + } + + if body != nil { + request.SetHeader("Content-Type", "application/json").Body(body) + } + + return request.DoRaw(ctx) +} diff --git a/cmd/kubectl-unbounded/app/net/detail_client_test.go b/cmd/kubectl-unbounded/app/net/detail_client_test.go new file mode 100644 index 000000000..a068baf9d --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/detail_client_test.go @@ -0,0 +1,143 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" +) + +func newDetailTestRuntime(t *testing.T, serverURL string) *pluginRuntime { + t.Helper() + + path := filepath.Join(t.TempDir(), "kubeconfig") + + data := fmt.Sprintf(`apiVersion: v1 +kind: Config +current-context: test +clusters: +- name: test + cluster: + server: %s +contexts: +- name: test + context: + cluster: test + user: test +users: +- name: test + user: + token: test-token +`, serverURL) + if err := os.WriteFile(path, []byte(data), 0o600); err != nil { + t.Fatal(err) + } + + rt := newPluginRuntime() + *rt.configFlags.KubeConfig = path + *rt.configFlags.Namespace = "unbounded-system" + + return rt +} + +func TestStatusRequestAggregatedTransport(t *testing.T) { + for _, method := range []string{http.MethodPost, http.MethodGet} { + t.Run(method, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != method || r.URL.Path != "/apis/status.net.unbounded-cloud.io/v1alpha1/nodes/node-a/details" { + t.Errorf("unexpected request %s %s", r.Method, r.URL) + } + + if r.Header.Get("Authorization") != "Bearer test-token" { + t.Error("request did not reuse Kubernetes authentication") + } + + if method == http.MethodPost { + body, err := io.ReadAll(r.Body) + if err != nil || string(body) != `{"forceRefresh":true}` || r.Header.Get("Content-Type") != "application/json" { + t.Errorf("unexpected request body %s, %v", body, err) + } + } else if r.URL.Query().Get("requestId") != "id/+ ?" { + t.Errorf("request ID changed: %q", r.URL.Query().Get("requestId")) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusAccepted) + _, _ = io.WriteString(w, `{"state":"pending"}`) + })) + defer server.Close() + + client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL, BearerToken: "test-token"}) + if err != nil { + t.Fatal(err) + } + + path := "/status/node/node-a/details" + + var body []byte + if method == http.MethodPost { + body = []byte(`{"forceRefresh":true}`) + } else { + path += "?requestId=id%2F%2B+%3F" + } + + raw, err := requestStatusViaAggregatedAPI(context.Background(), client, method, path, body) + if err != nil || string(raw) != `{"state":"pending"}` { + t.Fatalf("request = %s, %v", raw, err) + } + }) + } +} + +func TestStatusRequestPreservesHTTPFailure(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = io.WriteString(w, `{"state":"unavailable","error":"leadership changed"}`) + })) + defer server.Close() + + client, err := kubernetes.NewForConfig(&rest.Config{Host: server.URL}) + if err != nil { + t.Fatal(err) + } + + raw, err := requestStatusViaAggregatedAPI(context.Background(), client, http.MethodGet, "/status/node/node-a/details", nil) + if err == nil || !strings.Contains(string(raw), "leadership changed") { + t.Fatalf("lost explicit controller failure: %s, %v", raw, err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := requestStatusViaAggregatedAPI(ctx, client, http.MethodGet, "/status/node/node-a/details", nil); err == nil { + t.Fatal("canceled request succeeded") + } + + for _, path := range []string{"https://other.invalid/status", "://invalid"} { + if _, err := requestStatusViaAggregatedAPI(context.Background(), client, http.MethodGet, path, nil); err == nil { + t.Errorf("accepted invalid path %q", path) + } + } + + request, err := newStatusRequest(newDetailTestRuntime(t, server.URL), defaultNodeStatusFetchOptions()) + if err != nil { + t.Fatal(err) + } + + raw, err = request(context.Background(), http.MethodGet, "/status/node/node-a/details", nil) + if err == nil || !strings.Contains(string(raw), "leadership changed") { + t.Fatalf("fallback hid controller failure: %s, %v", raw, err) + } +} diff --git a/cmd/kubectl-unbounded/app/net/detail_poll_test.go b/cmd/kubectl-unbounded/app/net/detail_poll_test.go new file mode 100644 index 000000000..7ceeab5b3 --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/detail_poll_test.go @@ -0,0 +1,190 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/url" + "strings" + "testing" + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func detailResultFixture() statusv1alpha1.NodeDetailResult { + now := time.Now().UTC() + + return statusv1alpha1.NodeDetailResult{ + State: statusv1alpha1.NodeDetailComplete, NodeName: "node-a", RequestID: "request/+ ?", + Deadline: now.Add(-time.Second), + Details: &statusv1alpha1.NodeDetailSnapshot{ + NodeName: "node-a", RequestID: "request/+ ?", CollectedAt: now, ReceivedAt: now, + ExpiresAt: now.Add(time.Minute), + Status: &statusv1alpha1.NodeStatusResponse{ + Timestamp: now, NodeInfo: statusv1alpha1.NodeInfo{Name: "node-a"}, + Peers: []statusv1alpha1.PeerStatus{{Name: "peer"}}, + }, + }, + } +} + +func TestNodeDetailClientCachedAndPending(t *testing.T) { + for _, pending := range []bool{false, true} { + for _, refresh := range []bool{false, true} { + result := detailResultFixture() + calls := 0 + client := nodeDetailClient{ + pollInterval: time.Nanosecond, + request: func(_ context.Context, method, path string, body []byte) ([]byte, error) { + calls++ + if calls == 1 { + if method != http.MethodPost || path != "/status/node/node-a/details" { + t.Fatalf("initial request = %s %s", method, path) + } + + var args struct { + Refresh *bool `json:"forceRefresh"` + } + if err := json.Unmarshal(body, &args); err != nil || args.Refresh == nil || *args.Refresh != refresh { + t.Fatalf("refresh body = %s, %v", body, err) + } + + if pending { + result.Deadline = time.Now().Add(time.Minute) + + return json.Marshal(statusv1alpha1.NodeDetailResult{ + State: statusv1alpha1.NodeDetailPending, NodeName: result.NodeName, + RequestID: result.RequestID, Deadline: result.Deadline, + }) + } + } else { + target, err := url.Parse(path) + if err != nil || method != http.MethodGet || target.Query().Get("requestId") != result.RequestID || body != nil { + t.Fatalf("poll request = %s %s, %v", method, path, err) + } + } + + return json.Marshal(result) + }, + } + + got, err := client.fetch(context.Background(), "node-a", refresh) + if err != nil || got.NodeInfo.Name != "node-a" || len(got.Peers) != 1 { + t.Fatalf("fetch = %+v, %v", got, err) + } + + wantCalls := 1 + if pending { + wantCalls = 2 + } + + if calls != wantCalls { + t.Errorf("made %d calls, want %d", calls, wantCalls) + } + } + } +} + +func TestNodeDetailClientFailures(t *testing.T) { + for _, tc := range []struct { + name string + change func(*statusv1alpha1.NodeDetailResult) + want string + }{ + {"wrong node", func(r *statusv1alpha1.NodeDetailResult) { r.NodeName = "other" }, "identity mismatch"}, + {"wrong snapshot", func(r *statusv1alpha1.NodeDetailResult) { r.Details.NodeName = "other" }, "malformed"}, + {"wrong request", func(r *statusv1alpha1.NodeDetailResult) { r.Details.RequestID = "other" }, "malformed"}, + {"wrong payload", func(r *statusv1alpha1.NodeDetailResult) { r.Details.Status.NodeInfo.Name = "other" }, "malformed"}, + {"missing details", func(r *statusv1alpha1.NodeDetailResult) { r.Details = nil }, "malformed"}, + {"missing payload", func(r *statusv1alpha1.NodeDetailResult) { r.Details.Status = nil }, "malformed"}, + {"missing collection time", func(r *statusv1alpha1.NodeDetailResult) { r.Details.CollectedAt = time.Time{} }, "malformed"}, + {"missing receipt time", func(r *statusv1alpha1.NodeDetailResult) { r.Details.ReceivedAt = time.Time{} }, "malformed"}, + {"missing request", func(r *statusv1alpha1.NodeDetailResult) { r.RequestID = "" }, "malformed"}, + {"fetch failure", func(r *statusv1alpha1.NodeDetailResult) { r.Details.Status.FetchError = "unreachable" }, "collection failed"}, + {"expired snapshot", func(r *statusv1alpha1.NodeDetailResult) { + r.Details.ReceivedAt = time.Now().Add(-time.Hour) + r.Details.ExpiresAt = time.Now().Add(-time.Second) + }, "expired"}, + {"missing pending ID", func(r *statusv1alpha1.NodeDetailResult) { + r.State, r.RequestID, r.Details = statusv1alpha1.NodeDetailPending, "", nil + }, "malformed"}, + {"missing pending deadline", func(r *statusv1alpha1.NodeDetailResult) { + r.State, r.Deadline, r.Details = statusv1alpha1.NodeDetailPending, time.Time{}, nil + }, "malformed"}, + } { + t.Run(tc.name, func(t *testing.T) { + result := detailResultFixture() + tc.change(&result) + + client := nodeDetailClient{request: func(context.Context, string, string, []byte) ([]byte, error) { + return json.Marshal(result) + }} + if _, err := client.fetch(context.Background(), "node-a", false); err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("fetch error = %v, want %q", err, tc.want) + } + }) + } + + for _, state := range []statusv1alpha1.NodeDetailState{ + statusv1alpha1.NodeDetailExpired, statusv1alpha1.NodeDetailUnavailable, statusv1alpha1.NodeDetailRetryable, "unsupported", "", + } { + client := nodeDetailClient{request: func(context.Context, string, string, []byte) ([]byte, error) { + return json.Marshal(statusv1alpha1.NodeDetailResult{NodeName: "node-a", State: state, Error: "controller explanation"}) + }} + if _, err := client.fetch(context.Background(), "node-a", false); err == nil || + !strings.Contains(err.Error(), string(state)) || !strings.Contains(err.Error(), "controller explanation") { + t.Errorf("state %q error = %v", state, err) + } + } +} + +func TestNodeDetailClientDeadlineAndCancellation(t *testing.T) { + result := detailResultFixture() + result.State, result.Details, result.Deadline = statusv1alpha1.NodeDetailPending, nil, time.Now().Add(-time.Second) + calls := 0 + + client := nodeDetailClient{request: func(context.Context, string, string, []byte) ([]byte, error) { + calls++ + return json.Marshal(result) + }} + if _, err := client.fetch(context.Background(), "node-a", false); !errors.Is(err, context.DeadlineExceeded) || calls != 1 { + t.Fatalf("expired deadline = %v, calls=%d", err, calls) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if _, err := client.fetch(ctx, "node-a", false); !errors.Is(err, context.Canceled) { + t.Fatalf("cancellation = %v", err) + } + + client.pollInterval = time.Nanosecond + result.Deadline = time.Now().Add(time.Minute) + + client.request = func(context.Context, string, string, []byte) ([]byte, error) { + result.Deadline = result.Deadline.Add(time.Second) + return json.Marshal(result) + } + if _, err := client.fetch(context.Background(), "node-a", false); err == nil || !strings.Contains(err.Error(), "deadline changed") { + t.Fatalf("deadline extension accepted: %v", err) + } + + client.request = func(context.Context, string, string, []byte) ([]byte, error) { return []byte(`not JSON`), nil } + if _, err := client.fetch(context.Background(), "node-a", false); err == nil || !strings.Contains(err.Error(), "malformed") { + t.Fatalf("malformed response = %v", err) + } + + client.request = func(context.Context, string, string, []byte) ([]byte, error) { return nil, errors.New("HTTP 404") } + if _, err := client.fetch(context.Background(), "node-a", false); err == nil || !strings.Contains(err.Error(), "unsupported") { + t.Fatalf("unsupported API = %v", err) + } + + if _, err := client.fetch(context.Background(), "../other", false); err == nil || !strings.Contains(err.Error(), "invalid node name") { + t.Fatalf("invalid node = %v", err) + } +} diff --git a/cmd/kubectl-unbounded/app/net/node.go b/cmd/kubectl-unbounded/app/net/node.go index 42ac8fa9d..67e3a6203 100644 --- a/cmd/kubectl-unbounded/app/net/node.go +++ b/cmd/kubectl-unbounded/app/net/node.go @@ -4,6 +4,7 @@ package net import ( + "bytes" "context" "crypto/tls" "crypto/x509" @@ -27,6 +28,7 @@ import ( "k8s.io/client-go/rest" "k8s.io/client-go/tools/remotecommand" + netstatus "github.com/Azure/unbounded/internal/net/status" statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" ) @@ -178,16 +180,16 @@ func runNodeList(rt *pluginRuntime, cmd *cobra.Command, baseFetch nodeStatusFetc fetchOpts.timeout = override.timeout } - status, err := fetchClusterStatus(rt, cmd, fetchOpts) + status, err := fetchClusterSummary(rt, cmd, fetchOpts) if err != nil { return err } - rows := buildNodeRows(status) + rows := buildNodeRowsFromSummary(status) useColor := shouldUseColor(cmd.OutOrStdout(), color) if !suppressWarnings { - printWarnings(cmd.OutOrStdout(), collectWarnings(status), useColor) + printWarnings(cmd.OutOrStdout(), collectWarningsFromSummary(status), useColor) } switch output { @@ -213,19 +215,32 @@ func runNodeList(rt *pluginRuntime, cmd *cobra.Command, baseFetch nodeStatusFetc func fetchClusterStatus(rt *pluginRuntime, cmd *cobra.Command, opts nodeStatusFetchOptions) (clusterStatusResponse, error) { var status clusterStatusResponse - ns, err := rt.namespace() + raw, err := fetchClusterStatusRaw(rt, cmd, opts) if err != nil { return status, err } + if err := json.Unmarshal(raw, &status); err != nil { + return status, fmt.Errorf("decode /status/json: %w", err) + } + + return status, nil +} + +func fetchClusterStatusRaw(rt *pluginRuntime, cmd *cobra.Command, opts nodeStatusFetchOptions) ([]byte, error) { + ns, err := rt.namespace() + if err != nil { + return nil, err + } + client, err := rt.kubeClient() if err != nil { - return status, err + return nil, err } cfg, err := rt.restConfig() if err != nil { - return status, err + return nil, err } ctx, cancel := context.WithTimeout(cmd.Context(), opts.timeout) @@ -235,15 +250,89 @@ func fetchClusterStatus(rt *pluginRuntime, cmd *cobra.Command, opts nodeStatusFe if err != nil { raw, err = fetchStatusViaPortForward(ctx, client, cfg, ns, opts.controllerDeploy, opts.controllerSelector, opts.controllerPort, opts.timeout) if err != nil { - return status, fmt.Errorf("fetch /status/json failed via service proxy (%s) and pod port-forward (%s)", opts.controllerService, err) + return nil, fmt.Errorf("fetch /status/json failed via service proxy (%s) and pod port-forward (%s)", opts.controllerService, err) } } - if err := json.Unmarshal(raw, &status); err != nil { - return status, fmt.Errorf("decode /status/json: %w", err) + return raw, nil +} + +func fetchClusterSummary(rt *pluginRuntime, cmd *cobra.Command, opts nodeStatusFetchOptions) (clusterSummary, error) { + raw, err := fetchClusterStatusRaw(rt, cmd, opts) + if err != nil { + return clusterSummary{}, err } - return status, nil + return decodeClusterSummary(raw) +} + +// decodeClusterSummary projects legacy full responses once, never retaining details. +func decodeClusterSummary(raw []byte) (clusterSummary, error) { + var shape map[string]json.RawMessage + if err := json.Unmarshal(raw, &shape); err != nil { + return clusterSummary{}, fmt.Errorf("decode cluster overview: %w", err) + } + + var summary clusterSummary + if err := json.Unmarshal(raw, &summary); err != nil { + return summary, fmt.Errorf("decode cluster overview: %w", err) + } + + if _, ok := shape["nodeSummaries"]; ok { + return summary, nil + } + + if _, ok := shape["nodes"]; !ok { + return summary, fmt.Errorf("malformed cluster overview: missing nodeSummaries or nodes") + } + + var legacy clusterStatusResponse + if err := json.Unmarshal(raw, &legacy); err != nil { + return summary, fmt.Errorf("decode legacy cluster overview: %w", err) + } + + summary.NodeSummaries = make([]nodeSummary, 0, len(legacy.Nodes)) + + now := time.Now() + for _, node := range legacy.Nodes { + overview := netstatus.OverviewFromStatus(&node, now) + // Keep the metadata allocation independent of the full diagnostic payload. + info := node.NodeInfo + + entry := nodeSummary{ + NodeInfo: &info, LastPushTime: node.LastPushTime, + WireGuardOnline: node.NodeInfo.WireGuard != nil && node.NodeInfo.WireGuard.Interface != "", + Name: node.NodeInfo.Name, SiteName: node.NodeInfo.SiteName, + IsGateway: node.NodeInfo.IsGateway, K8sReady: node.NodeInfo.K8sReady, + StatusSource: node.StatusSource, FetchError: node.FetchError, + PeerCount: overview.PeerCount, HealthyPeers: overview.HealthyPeers, RouteCount: overview.RouteCount, + RouteMismatch: overview.RouteMismatch, ErrorCount: len(node.NodeErrors), + CniStatus: cniStatusLabel(node, legacy.PullEnabled), CniTone: statusTone(node, legacy.PullEnabled), + } + if len(node.NodeErrors) > 0 { + entry.FirstError = node.NodeErrors[0].Message + } + + summary.NodeSummaries = append(summary.NodeSummaries, entry) + } + + summary.NodeCount = len(summary.NodeSummaries) + + return summary, nil +} + +// summaryStatusMetadata adapts overview identity data for existing detail renderers. +func summaryStatusMetadata(summary clusterSummary) clusterStatusResponse { + status := clusterStatusResponse{ + PullEnabled: summary.PullEnabled, LeaderInfo: summary.LeaderInfo, + Sites: summary.Sites, GatewayPools: summary.GatewayPools, Warnings: summary.Warnings, + Nodes: make([]statusv1alpha1.NodeStatusResponse, 0, len(summary.NodeSummaries)), + } + for _, node := range summary.NodeSummaries { + status.Nodes = append(status.Nodes, node.statusMetadata()) + } + + return status } // newNodeLogsCommand shows CNI node-agent logs for a specific Kubernetes node. @@ -506,110 +595,91 @@ func listNodeNamesForCompletion(rt *pluginRuntime, ctx context.Context, prefix s // listNodePeeringsForCompletion returns existing peering destination node names for a source node. func listNodePeeringsForCompletion(rt *pluginRuntime, cmd *cobra.Command, baseFetch nodeStatusFetchOptions, sourceNode, prefix string) ([]string, error) { - status, err := fetchClusterStatus(rt, cmd, nodeStatusFetchFromCommand(cmd).merged(baseFetch)) + candidates, err := listNodeNamesForCompletion(rt, cmd.Context(), prefix) if err != nil { return nil, err } - node, ok := nodeStatusByName(status, sourceNode) - if !ok { - return nil, fmt.Errorf("node %q not found in status", sourceNode) - } - - seen := make(map[string]struct{}) - - names := make([]string, 0, len(node.Peers)) - for _, peer := range node.Peers { - name := strings.TrimSpace(peer.Name) - if name == "" || !strings.HasPrefix(name, prefix) { - continue - } - - if _, exists := seen[name]; exists { - continue + names := make([]string, 0, len(candidates)) + for _, name := range candidates { + if name != sourceNode { + names = append(names, name) } - - seen[name] = struct{}{} - names = append(names, name) } return names, nil } -// newNodeShowCommand prints node info and detail tables from cluster status. +// newNodeShowCommand explicitly obtains one node's expiring diagnostics. func newNodeShowCommand(rt *pluginRuntime, baseFetch nodeStatusFetchOptions) *cobra.Command { var ( - color string - watch bool + color string + watch bool + refresh bool ) cmd := &cobra.Command{ Use: "show NODE_NAME [peer|peers|route|routes|bpf|json] [PEER_NODE]", - Short: "Show node info, peers, routes, BPF entries, or raw JSON from status data", + Short: "Load node diagnostics: info, peers, routes, BPF entries, or raw JSON", Args: cobra.RangeArgs(1, 3), RunE: func(cmd *cobra.Command, args []string) error { if watch { - fetchOpts := nodeStatusFetchFromCommand(cmd).merged(baseFetch) - nodeName := args[0] - mode := "" - peerName := "" + return fmt.Errorf("node show does not watch diagnostic data; use node list --watch for overview updates, or repeat node show --refresh") + } - if len(args) >= 2 { - mode = strings.ToLower(args[1]) - } + mode := "" + if len(args) >= 2 { + mode = strings.ToLower(args[1]) + } - if len(args) == 3 { - peerName = args[2] + switch mode { + case "", "peer", "peers", "route", "routes", "bpf", "json": + default: + return fmt.Errorf("unsupported show subcommand %q, expected peer(s), route(s), bpf, or json", args[1]) + } + + fetchOpts := nodeStatusFetchFromCommand(cmd).merged(baseFetch) + + var ( + status clusterStatusResponse + overview *nodeSummary + ) + + if mode == "" || mode == "peer" || mode == "peers" { + summary, err := fetchClusterSummary(rt, cmd, fetchOpts) + if err != nil { + return err } - return runWatch(cmd.Context(), rt, cmd, fetchOpts, color, func(w io.Writer, status clusterStatusResponse, useColor bool) error { - node, ok := nodeStatusByName(status, nodeName) - if !ok { - return fmt.Errorf("node %q not found in controller status", nodeName) + status = summaryStatusMetadata(summary) + for i := range summary.NodeSummaries { + if summary.NodeSummaries[i].Name == args[0] { + overview = &summary.NodeSummaries[i] + break } + } - switch mode { - case "peer", "peers": - return printNodePeerings(w, status, node, peerName, useColor) - case "route", "routes": - return printNodeRoutes(w, node, useColor, rt.interfaceNames) - case "bpf": - return printNodeBpf(w, node) - case "json": - data, jsonErr := json.MarshalIndent(node, "", " ") - if jsonErr != nil { - return jsonErr - } - - _, _ = fmt.Fprintf(w, "%s\n", data) //nolint:errcheck - - return nil - default: - return printNodeInfoPane(w, status, node, useColor) - } - }, nil) + if mode == "" && overview == nil { + return fmt.Errorf("node %q not found in cluster overview", args[0]) + } } - useColor := shouldUseColor(cmd.OutOrStdout(), color) - - status, err := fetchClusterStatus(rt, cmd, nodeStatusFetchFromCommand(cmd).merged(baseFetch)) + request, err := newStatusRequest(rt, fetchOpts) if err != nil { return err } - nodeName := args[0] - - node, ok := nodeStatusByName(status, nodeName) - if !ok { - return fmt.Errorf("node %q not found in controller status", nodeName) + details, err := (nodeDetailClient{request: request}).fetch(cmd.Context(), args[0], refresh) + if err != nil { + return err } - if len(args) == 1 { - return printNodeInfoPane(cmd.OutOrStdout(), status, node, useColor) - } + node := *details + useColor := shouldUseColor(cmd.OutOrStdout(), color) - mode := strings.ToLower(args[1]) switch mode { + case "": + return printNodeInfoPane(cmd.OutOrStdout(), status, *overview, useColor) case "peer", "peers": var peerName string if len(args) == 3 { @@ -637,7 +707,8 @@ func newNodeShowCommand(rt *pluginRuntime, baseFetch nodeStatusFetchOptions) *co } cmd.ValidArgsFunction = nodeShowCompletion(rt, baseFetch) cmd.Flags().StringVarP(&color, "color", "C", "auto", "Colorize output: auto|always|never (pass -C with no value for always)") - cmd.Flags().BoolVarP(&watch, "watch", "w", false, "Watch live updates via WebSocket") + cmd.Flags().BoolVarP(&watch, "watch", "w", false, "Unsupported for diagnostics; use node list --watch") + cmd.Flags().BoolVar(&refresh, "refresh", false, "Collect fresh node diagnostics instead of reusing an unexpired snapshot") if flag := cmd.Flags().Lookup("color"); flag != nil { flag.NoOptDefVal = "always" @@ -768,16 +839,6 @@ func execInPod( } // nodeStatusByName returns the node status entry with the given node name. -func nodeStatusByName(status clusterStatusResponse, nodeName string) (statusv1alpha1.NodeStatusResponse, bool) { - for _, n := range status.Nodes { - if strings.EqualFold(n.NodeInfo.Name, nodeName) { - return n, true - } - } - - return statusv1alpha1.NodeStatusResponse{}, false -} - // fetchStatusViaServiceProxy fetches status through the aggregated API endpoint. func fetchStatusViaServiceProxy(ctx context.Context, client *kubernetes.Clientset, ns, service, port string) ([]byte, error) { return client.CoreV1().RESTClient(). @@ -798,6 +859,18 @@ func fetchStatusViaPortForward( selector string, remotePort string, timeout time.Duration, +) ([]byte, error) { + return requestStatusViaPortForward(ctx, client, cfg, ns, deployName, selector, remotePort, timeout, http.MethodGet, "/status/json", nil) +} + +func requestStatusViaPortForward( + ctx context.Context, + client *kubernetes.Clientset, + cfg *rest.Config, + ns, deployName, selector, remotePort string, + timeout time.Duration, + method, path string, + body []byte, ) ([]byte, error) { pods, err := podsForController(ctx, client, ns, deployName, selector) if err != nil { @@ -825,6 +898,8 @@ func fetchStatusViaPortForward( } stopCh := make(chan struct{}, 1) + defer close(stopCh) + readyCh := make(chan struct{}) errCh := make(chan error, 1) @@ -845,8 +920,6 @@ func fetchStatusViaPortForward( return nil, ctx.Err() } - defer close(stopCh) - fwdPorts, err := fw.GetPorts() if err != nil { return nil, err @@ -861,11 +934,15 @@ func fetchStatusViaPortForward( reqCtx, cancel := context.WithTimeout(ctx, timeout) defer cancel() - req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, fmt.Sprintf("https://127.0.0.1:%d/status/json", localPort), nil) + req, err := http.NewRequestWithContext(reqCtx, method, fmt.Sprintf("https://127.0.0.1:%d%s", localPort, path), bytes.NewReader(body)) if err != nil { return nil, err } + if body != nil { + req.Header.Set("Content-Type", "application/json") + } + // Request an HMAC viewer token for authentication. When port-forwarding // directly to the controller pod, the API server front-proxy is bypassed // so the controller requires an HMAC token. @@ -892,6 +969,7 @@ func fetchStatusViaPortForward( }, }, } + defer tlsClient.CloseIdleConnections() resp, err := tlsClient.Do(req) if err != nil { @@ -900,16 +978,16 @@ func fetchStatusViaPortForward( defer func() { _ = resp.Body.Close() }() //nolint:errcheck - body, err := io.ReadAll(resp.Body) + responseBody, err := io.ReadAll(resp.Body) if err != nil { return nil, err } if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return nil, fmt.Errorf("controller /status/json returned %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + return responseBody, fmt.Errorf("controller %s returned %d: %s", path, resp.StatusCode, strings.TrimSpace(string(responseBody))) } - return body, nil + return responseBody, nil } // podsForController returns controller pods from deployment selector, falling back to label selector. diff --git a/cmd/kubectl-unbounded/app/net/node_detail_command_test.go b/cmd/kubectl-unbounded/app/net/node_detail_command_test.go new file mode 100644 index 000000000..dab56642d --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/node_detail_command_test.go @@ -0,0 +1,412 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/spf13/cobra" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) + +func TestNodeShowRequestsNamedDetails(t *testing.T) { + for _, mode := range []string{"", "peer", "peers", "route", "routes", "bpf", "json"} { + for _, refresh := range []bool{false, true} { + t.Run(mode, func(t *testing.T) { + result := detailResultFixture() + result.Details.Status.RoutingTable.Routes = []statusv1alpha1.RouteEntry{{ + Destination: "10.20.0.0/24", NextHops: []statusv1alpha1.NextHop{{Gateway: "10.1.0.1"}}, + }} + result.Details.Status.BpfEntries = []statusv1alpha1.BpfEntry{{CIDR: "10.30.0.0/24", Node: "peer"}} + + var posts atomic.Int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + + switch r.URL.Path { + case "/apis/status.net.unbounded-cloud.io/v1alpha1/status/json": + if mode != "" && mode != "peer" && mode != "peers" { + t.Error("detail-only output fetched cluster status unnecessarily") + } + + if r.Method != http.MethodGet { + t.Errorf("metadata request method %s", r.Method) + } + + _, _ = io.WriteString(w, `{"nodeSummaries":[{"name":"node-a"},{"name":"peer","siteName":"site","k8sReady":"Ready"}]}`) + case "/apis/status.net.unbounded-cloud.io/v1alpha1/nodes/node-a/details": + if r.Method != http.MethodPost { + t.Errorf("cached show must POST once, got %s", r.Method) + } + + posts.Add(1) + + var body struct { + Refresh *bool `json:"forceRefresh"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil || body.Refresh == nil || *body.Refresh != refresh { + t.Errorf("refresh = %+v, %v; want %v", body, err, refresh) + } + + if err := json.NewEncoder(w).Encode(result); err != nil { + t.Error(err) + } + default: + t.Errorf("unexpected endpoint %s", r.URL.Path) + http.NotFound(w, r) + } + })) + defer server.Close() + + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + + args := []string{"show", "node-a", "--color=never"} + if mode != "" { + args = append(args, mode) + } + + if refresh { + args = append(args, "--refresh") + } + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs(args) + + if err := cmd.Execute(); err != nil { + t.Fatal(err) + } + + if posts.Load() != 1 || out.Len() == 0 { + t.Fatalf("show did not load/render exactly one named snapshot: calls=%d, output=%s", posts.Load(), out.String()) + } + + want := map[string]string{"peer": "peer", "peers": "peer", "route": "10.20.0.0/24", "routes": "10.20.0.0/24", "bpf": "10.30.0.0/24"}[mode] + if want != "" && !strings.Contains(out.String(), want) { + t.Fatalf("diagnostic content lost: %s", out.String()) + } + + if mode == "json" { + var snapshot statusv1alpha1.NodeStatusResponse + if err := json.Unmarshal(out.Bytes(), &snapshot); err != nil || !reflect.DeepEqual(&snapshot, result.Details.Status) { + t.Fatalf("raw diagnostic content changed: %s, %v", out.String(), err) + } + + var decoded map[string]json.RawMessage + if err := json.Unmarshal(out.Bytes(), &decoded); err != nil { + t.Fatal(err) + } + + if decoded["nodeInfo"] == nil || decoded["peers"] == nil || decoded["routingTable"] == nil || + decoded["bpfEntries"] == nil || decoded["details"] != nil || decoded["state"] != nil { + t.Fatalf("raw node JSON shape changed: %s", out.String()) + } + } + }) + } + } +} + +func TestNodeShowOverviewConsistency(t *testing.T) { + for _, tc := range []struct { + name, ready, cni string + old, missingInfo, legacy bool + }{ + {name: "unenriched diagnostics", ready: "Ready", cni: "Healthy"}, + {name: "older diagnostics", ready: "Ready", cni: "Healthy", old: true}, + {name: "stale", ready: "NotReady", cni: "Stale"}, + {name: "no data", ready: "NotReady", cni: "No data"}, + {name: "fallback", ready: "Ready", cni: "Fallback"}, + {name: "errors", ready: "Ready", cni: "Errors"}, + {name: "route mismatch", ready: "Ready", cni: "Route mismatch"}, + {name: "missing metadata", ready: "Ready", cni: "Healthy", missingInfo: true, old: true}, + {name: "missing health", missingInfo: true, old: true}, + {name: "legacy overview", ready: "Ready", cni: "Healthy", legacy: true}, + } { + t.Run(tc.name, func(t *testing.T) { + result := detailResultFixture() + now := result.Details.ReceivedAt + info := &statusv1alpha1.NodeInfo{ + Name: "node-a", K8sReady: tc.ready, SiteName: "site", + OSImage: "Ubuntu 24.04.4", Kernel: "6.8.0-1067-azure", Kubelet: "v1.35.7", + InternalIPs: []string{"10.224.0.103"}, PodCIDRs: []string{"10.20.0.0/24"}, + K8sUpdatedAt: &now, BuildInfo: &statusv1alpha1.BuildInfo{Commit: "current-build"}, + WireGuard: &statusv1alpha1.WireGuardStatusInfo{PublicKey: "current-key"}, + K8sLabels: map[string]string{ + "node.kubernetes.io/instance-type": "Standard_D2ads_v6", + "topology.kubernetes.io/region": "canadacentral", + "topology.kubernetes.io/zone": "canadacentral-1", + }, + } + + entry := nodeSummary{ + Name: "node-a", SiteName: "site", K8sReady: tc.ready, CniStatus: tc.cni, + StatusSource: "ws", NodeInfo: info, LastPushTime: &now, + } + if tc.cni == "Errors" { + entry.ErrorCount, entry.FirstError, entry.FetchError = 2, "current error", "current fetch error" + } + + if tc.missingInfo { + entry.NodeInfo, entry.LastPushTime = nil, nil + } + + if tc.old { + result.Details.Status.NodeInfo = statusv1alpha1.NodeInfo{ + Name: "node-a", K8sReady: "NotReady", OSImage: "old-image", + Kernel: "old-kernel", Kubelet: "old-kubelet", InternalIPs: []string{"10.0.0.1"}, + K8sLabels: map[string]string{"topology.kubernetes.io/region": "old-region"}, + } + result.Details.Status.StatusSource = "stale-cache" + result.Details.Status.NodeErrors = []statusv1alpha1.NodeError{{Message: "old diagnostic error"}} + } + + var overview any = clusterSummary{NodeSummaries: []nodeSummary{entry}} + if tc.legacy { + overview = clusterStatusResponse{Nodes: []statusv1alpha1.NodeStatusResponse{entry.statusMetadata()}} + } + + var summaries, details atomic.Int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + + switch { + case strings.HasSuffix(r.URL.Path, "/status/json") && r.Method == http.MethodGet: + summaries.Add(1) + + _ = json.NewEncoder(w).Encode(overview) + case strings.HasSuffix(r.URL.Path, "/nodes/node-a/details") && r.Method == http.MethodPost: + details.Add(1) + + _ = json.NewEncoder(w).Encode(result) + default: + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + http.NotFound(w, r) + } + })) + defer server.Close() + + execute := func(args ...string) string { + t.Helper() + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs(args) + + if err := cmd.Execute(); err != nil { + t.Fatal(err) + } + + return out.String() + } + + var rows []nodeListRow + if err := json.Unmarshal([]byte(execute("list", "-o", "json", "--suppress-warnings")), &rows); err != nil || len(rows) != 1 { + t.Fatalf("list rows: %+v, %v", rows, err) + } + + output := execute("show", "node-a", "--color=never") + fields := map[string]string{} + + for _, line := range strings.Split(output, "\n") { + if key, value, ok := strings.Cut(line, " "); ok { + fields[strings.TrimSpace(key)] = strings.TrimSpace(value) + } + } + + if fields["K8s Status"] != rows[0].K8sStatus || fields["UN Status"] != rows[0].WGStatus || + fields["K8s Status"] != valueOr(tc.ready, "Unknown") || fields["UN Status"] != valueOr(tc.cni, "Unknown") { + t.Fatalf("list/show health disagreement: %+v\n%s", rows, output) + } + + want := map[string]string{ + "Node Image": "Ubuntu 24.04.4", "Kernel": "6.8.0-1067-azure", "Kubelet Version": "v1.35.7", + "Instance Type": "Standard_D2ads_v6", "Region": "canadacentral", "Availability Zone": "canadacentral-1", + "Internal IPs": "10.224.0.103", "Pod CIDRs": "10.20.0.0/24", "WireGuard Public Key": "current-key", + "Node Agent Build": "Commit: current-build", + } + for key, value := range want { + if tc.missingInfo { + value = "-" + } + + if fields[key] != value { + t.Errorf("%s = %q, want %q", key, fields[key], value) + } + } + + for _, key := range []string{"K8s Node Updated", "Status Push Updated"} { + if (fields[key] == "Never") != tc.missingInfo || fields[key] == "" { + t.Errorf("%s did not use summary timestamp: %q", key, fields[key]) + } + } + + if tc.cni == "Errors" && (fields["Node Error Count"] != "2" || fields["First Node Error"] != "current error" || + fields["Fetch Error"] != "current fetch error") { + t.Fatalf("current errors lost: %s", output) + } + + if strings.Contains(output, "old") || summaries.Load() != 2 || details.Load() != 1 { + t.Fatalf("stale metadata or extra requests: summaries=%d details=%d\n%s", summaries.Load(), details.Load(), output) + } + }) + } +} + +func TestNodeShowOverviewFailures(t *testing.T) { + for _, tc := range []struct { + body, want string + status int + }{ + {body: `{"nodeSummaries":[]}`, want: "not found in cluster overview"}, + {body: `{}`, want: "missing nodeSummaries or nodes"}, + {body: `{`, want: "decode cluster overview"}, + {body: "forbidden", status: http.StatusForbidden, want: "failed"}, + } { + t.Run(tc.want, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.Contains(r.URL.Path, "/details") { + t.Error("overview failure requested diagnostics") + } + + w.Header().Set("Content-Type", "application/json") + + if tc.status != 0 { + w.WriteHeader(tc.status) + } + + _, _ = io.WriteString(w, tc.body) + })) + defer server.Close() + + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"show", "node-a", "--color=never"}) + + cmd.SilenceUsage = true + if err := cmd.Execute(); err == nil || !strings.Contains(err.Error(), tc.want) || out.Len() != 0 { + t.Fatalf("overview failure not surfaced: %v, %s", err, out.String()) + } + }) + } +} + +func TestNodeShowRejectsExpiredDiagnostics(t *testing.T) { + result := detailResultFixture() + result.Details.ReceivedAt = result.Details.ReceivedAt.Add(-2 * time.Minute) + result.Details.ExpiresAt = result.Details.ExpiresAt.Add(-2 * time.Minute) + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if !strings.HasSuffix(r.URL.Path, "/nodes/node-a/details") || r.Method != http.MethodPost { + t.Errorf("unexpected request: %s %s", r.Method, r.URL) + } + + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(result) + })) + defer server.Close() + + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + cmd.SetOut(io.Discard) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"show", "node-a", "json"}) + + if err := cmd.Execute(); err == nil || !strings.Contains(err.Error(), "expired") { + t.Fatalf("expired diagnostics rendered: %v", err) + } +} + +func TestNodeShowFailureDoesNotFallBackToFullStatus(t *testing.T) { + var calls atomic.Int32 + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + + if !strings.HasSuffix(r.URL.Path, "/nodes/node-a/details") || r.Method != http.MethodPost { + t.Errorf("unexpected fallback request: %s %s", r.Method, r.URL) + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _, _ = io.WriteString(w, `{"state":"unavailable","nodeName":"node-a","error":"node is unreachable"}`) + })) + defer server.Close() + + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"show", "node-a", "json"}) + + err := cmd.Execute() + if err == nil || !strings.Contains(err.Error(), "unreachable") || !strings.Contains(err.Error(), "unavailable") { + t.Fatalf("explicit diagnostic failure not surfaced: %v", err) + } + + if calls.Load() != 1 { + t.Errorf("failure triggered fallback requests: %d", calls.Load()) + } +} + +func TestNodePeerCompletionDoesNotLoadDiagnostics(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/api/v1/nodes" { + t.Errorf("completion requested diagnostic endpoint: %s %s", r.Method, r.URL) + } + + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, `{"apiVersion":"v1","kind":"NodeList","items":[{"metadata":{"name":"node-a"}},{"metadata":{"name":"node-b"}}]}`) + })) + defer server.Close() + + cmd := &cobra.Command{} + cmd.SetContext(context.Background()) + + names, err := listNodePeeringsForCompletion(newDetailTestRuntime(t, server.URL), cmd, + defaultNodeStatusFetchOptions(), "node-a", "node") + if err != nil || len(names) != 1 || names[0] != "node-b" { + t.Fatalf("completion names = %v, %v", names, err) + } +} + +func TestNodeShowRejectsWatchAndInvalidModeWithoutRequests(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + t.Errorf("invalid/watch command requested %s", r.URL) + })) + defer server.Close() + + for _, args := range [][]string{ + {"show", "node-a", "json", "--watch"}, + {"show", "node-a", "unknown"}, + } { + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + cmd.SetOut(io.Discard) + cmd.SetErr(io.Discard) + cmd.SetArgs(args) + + if err := cmd.Execute(); err == nil { + t.Errorf("accepted invalid/watch invocation %v", args) + } + } +} diff --git a/cmd/kubectl-unbounded/app/net/node_render.go b/cmd/kubectl-unbounded/app/net/node_render.go index 266b71ddc..21ea76bdb 100644 --- a/cmd/kubectl-unbounded/app/net/node_render.go +++ b/cmd/kubectl-unbounded/app/net/node_render.go @@ -62,7 +62,8 @@ type nodeListRow struct { } // printNodeInfoPane prints fields similar to the UI node information panel. -func printNodeInfoPane(w io.Writer, status clusterStatusResponse, node statusv1alpha1.NodeStatusResponse, useColor bool) error { +func printNodeInfoPane(w io.Writer, status clusterStatusResponse, overview nodeSummary, useColor bool) error { + node := overview.statusMetadata() gatewayByNode := map[string]string{} for _, pool := range status.GatewayPools { @@ -121,12 +122,12 @@ func printNodeInfoPane(w io.Writer, status clusterStatusResponse, node statusv1a buildStr = "-" } - k8sReady := valueOr(node.NodeInfo.K8sReady, "NotReady") - wgStatus := cniStatusLabel(node, status.PullEnabled) + k8sReady := valueOr(overview.K8sReady, "Unknown") + wgStatus := valueOr(overview.CniStatus, "Unknown") if useColor { k8sReady = colorize(k8sReady, k8sTone(k8sReady)) - wgStatus = colorize(wgStatus, statusTone(node, status.PullEnabled)) + wgStatus = colorize(wgStatus, valueOr(overview.CniTone, "yellow")) } rows := [][2]string{ @@ -136,6 +137,9 @@ func printNodeInfoPane(w io.Writer, status clusterStatusResponse, node statusv1a {"Pool", valueOr(gatewayByNode[node.NodeInfo.Name], "-")}, {"K8s Status", k8sReady}, {"UN Status", wgStatus}, + {"Node Error Count", fmt.Sprint(overview.ErrorCount)}, + {"First Node Error", valueOr(overview.FirstError, "-")}, + {"Fetch Error", valueOr(overview.FetchError, "-")}, {"WireGuard Public Key", func() string { if node.NodeInfo.WireGuard != nil { return valueOr(node.NodeInfo.WireGuard.PublicKey, "-") @@ -157,37 +161,8 @@ func printNodeInfoPane(w io.Writer, status clusterStatusResponse, node statusv1a {"K8s Node Updated", formatAgePtr(node.NodeInfo.K8sUpdatedAt)}, {"Status Push Updated", formatAgePtr(node.LastPushTime)}, } - if err := printKVRows(w, rows); err != nil { - return err - } - - if len(node.NodeErrors) > 0 { - if _, err := fmt.Fprintln(w); err != nil { - return err - } - - if _, err := fmt.Fprintln(w, " Node Errors:"); err != nil { - return err - } - - for _, ne := range node.NodeErrors { - msg := strings.TrimSpace(ne.Message) - if msg == "" { - continue - } - - line := fmt.Sprintf(" - [%s] %s", ne.Type, msg) - if useColor { - line = colorize(line, "red") - } - if _, err := fmt.Fprintln(w, line); err != nil { - return err - } - } - } - - return nil + return printKVRows(w, rows) } // printNodePeerings prints a peering table and optional details for one peer. @@ -873,7 +848,7 @@ func buildNodeRowsFromSummary(summary clusterSummary) []nodeListRow { k8s := ns.K8sReady if k8s == "" { - k8s = "NotReady" + k8s = "Unknown" } cniStatus := ns.CniStatus diff --git a/cmd/kubectl-unbounded/app/net/summary_poll_test.go b/cmd/kubectl-unbounded/app/net/summary_poll_test.go new file mode 100644 index 000000000..845780ae0 --- /dev/null +++ b/cmd/kubectl-unbounded/app/net/summary_poll_test.go @@ -0,0 +1,168 @@ +// Copyright (c) Microsoft Corporation. +// SPDX-License-Identifier: Apache-2.0 + +package net + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "testing" +) + +const legacyOverviewFixture = `{ + "seq":7,"pullEnabled":false,"leaderInfo":{"podName":"leader"}, + "sites":[{"name":"site","manageCniPlugin":true}], + "gatewayPools":[{"name":"pool","gateways":["node-a"]}], + "warnings":["controller warning"], + "nodes":[{ + "nodeInfo":{"name":"node-a","siteName":"site","isGateway":true,"k8sReady":"Ready"}, + "statusSource":"ws","nodeErrors":[{"type":"cni","message":"not ready"}], + "peers":[{"name":"peer","healthCheck":{"enabled":true,"status":"up"}},{"name":"down"}], + "routingTable":{"routes":[{"nextHops":[{"expected":true,"present":false}]}]}, + "bpfEntries":[{"cidr":"10.0.0.0/24","node":"secret-detail"}] + }] +}` + +func TestDecodeClusterSummaryCompatibility(t *testing.T) { + summary, err := decodeClusterSummary([]byte(legacyOverviewFixture)) + if err != nil { + t.Fatal(err) + } + + if summary.Seq != 7 || summary.LeaderInfo.PodName != "leader" || len(summary.NodeSummaries) != 1 { + t.Fatalf("missing cluster metadata: %+v", summary) + } + + node := summary.NodeSummaries[0] + if node.PeerCount != 2 || node.HealthyPeers != 1 || node.RouteCount != 1 || !node.RouteMismatch || + node.ErrorCount != 1 || node.FirstError != "not ready" || node.Name != "node-a" { + t.Fatalf("overview facts changed: %+v", node) + } + + raw, err := json.Marshal(summary) + if err != nil { + t.Fatal(err) + } + + for _, forbidden := range []string{`"nodes"`, `"peers"`, `"routingTable"`, `"bpfEntries"`, "secret-detail", "10.0.0.0/24"} { + if strings.Contains(string(raw), forbidden) { + t.Errorf("retained detail %q in summary: %s", forbidden, raw) + } + } + + got, err := decodeClusterSummary(raw) + if err != nil || !reflect.DeepEqual(got.NodeSummaries[0], node) { + t.Fatalf("summary round trip changed: %+v, %v", got, err) + } + + for _, raw := range []string{`{`, `{}`, `null`, `{"nodeSummaries":"bad"}`, `{"nodes":"bad"}`} { + if _, err := decodeClusterSummary([]byte(raw)); err == nil { + t.Errorf("accepted malformed response %s", raw) + } + } + + if got, err := decodeClusterSummary([]byte(`{"nodeSummaries":[],"nodes":"ignored legacy field"}`)); err != nil || len(got.NodeSummaries) != 0 { + t.Fatalf("summary must take precedence: %+v, %v", got, err) + } +} + +func TestMergeClusterSummaryDelta(t *testing.T) { + summary := clusterSummary{ + Seq: 1, PullEnabled: true, NodeCount: 2, + NodeSummaries: []nodeSummary{ + {Name: "keep", PeerCount: 3, HealthyPeers: 2}, + {Name: "remove", PeerCount: 4}, + }, + Warnings: []string{"old warning"}, + } + + raw := []byte(`{ + "seq":2,"nodeCount":2,"pullEnabled":false,"warnings":[], + "removedNodes":["remove"], + "nodeSummaries":[{"name":"new","peerCount":7,"healthyPeers":5}] + }`) + if err := mergeClusterSummaryDelta(&summary, raw); err != nil { + t.Fatal(err) + } + + if summary.Seq != 2 || summary.PullEnabled || len(summary.Warnings) != 0 || len(summary.NodeSummaries) != 2 { + t.Fatalf("delta metadata not applied: %+v", summary) + } + + nodes := make(map[string]nodeSummary) + for _, node := range summary.NodeSummaries { + nodes[node.Name] = node + } + + if nodes["keep"].PeerCount != 3 || nodes["new"].HealthyPeers != 5 { + t.Fatalf("delta lost existing/updated facts: %+v", nodes) + } + + if _, ok := nodes["remove"]; ok { + t.Fatal("removed node still retained") + } + + if err := mergeClusterSummaryDelta(&summary, []byte(`{"nodeSummaries":[{"name":"keep","peerCount":0}]}`)); err != nil { + t.Fatal(err) + } + + for _, node := range summary.NodeSummaries { + if node.Name == "keep" && (node.PeerCount != 0 || node.HealthyPeers != 0) { + t.Fatalf("zero-valued update not applied: %+v", node) + } + } + + for _, raw := range []string{`null`, `{`, `{"nodeSummaries":"bad"}`, `{"pullEnabled":"bad"}`} { + if err := mergeClusterSummaryDelta(&summary, []byte(raw)); err == nil { + t.Errorf("accepted malformed summary delta %s", raw) + } + } +} + +func TestNodeListUsesOnlyOverview(t *testing.T) { + for _, fixture := range []string{ + legacyOverviewFixture, + `{"nodeSummaries":[{"name":"node-a","peerCount":2,"healthyPeers":1}],"sites":[],"gatewayPools":[]}`, + } { + for _, output := range []string{"json", "table", "wide"} { + t.Run(output, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet || r.URL.Path != "/apis/status.net.unbounded-cloud.io/v1alpha1/status/json" { + t.Errorf("list requested non-overview endpoint: %s %s", r.Method, r.URL) + http.Error(w, "unexpected request", http.StatusBadRequest) + + return + } + + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, fixture) + })) + defer server.Close() + + cmd := newNodeRootCommand(newDetailTestRuntime(t, server.URL)) + + var out bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(io.Discard) + cmd.SetArgs([]string{"list", "-o", output, "--color=never", "--suppress-warnings"}) + + if err := cmd.Execute(); err != nil { + t.Fatal(err) + } + + if !strings.Contains(out.String(), "node-a") || !strings.Contains(out.String(), "1/2") { + t.Fatalf("missing summary row: %s", out.String()) + } + + if strings.Contains(out.String(), "secret-detail") { + t.Fatal("list leaked details") + } + }) + } + } +} diff --git a/cmd/kubectl-unbounded/app/net/summary_types.go b/cmd/kubectl-unbounded/app/net/summary_types.go index d0c916c13..5e98a2959 100644 --- a/cmd/kubectl-unbounded/app/net/summary_types.go +++ b/cmd/kubectl-unbounded/app/net/summary_types.go @@ -3,7 +3,11 @@ package net -import "time" +import ( + "time" + + statusv1alpha1 "github.com/Azure/unbounded/internal/net/status/v1alpha1" +) // clusterSummary is the lightweight cluster overview received via the WS // summary protocol. It mirrors the controller's ClusterSummary type, carrying @@ -26,6 +30,10 @@ type clusterSummary struct { // controller's NodeSummary type and provides pre-computed status fields so the // plugin can render a useful table without full per-node data. type nodeSummary struct { + NodeInfo *statusv1alpha1.NodeInfo `json:"nodeInfo,omitempty"` + LastPushTime *time.Time `json:"lastPushTime,omitempty"` + WireGuardOnline bool `json:"wireGuardOnline"` + Name string `json:"name"` SiteName string `json:"siteName,omitempty"` IsGateway bool `json:"isGateway,omitempty"` @@ -42,6 +50,20 @@ type nodeSummary struct { FetchError string `json:"fetchError,omitempty"` } +func (n nodeSummary) statusMetadata() statusv1alpha1.NodeStatusResponse { + var info statusv1alpha1.NodeInfo + if n.NodeInfo != nil { + info = *n.NodeInfo + } + + info.Name, info.SiteName, info.IsGateway, info.K8sReady = n.Name, n.SiteName, n.IsGateway, n.K8sReady + + return statusv1alpha1.NodeStatusResponse{ + NodeInfo: info, LastPushTime: n.LastPushTime, + StatusSource: n.StatusSource, FetchError: n.FetchError, + } +} + // wsClientMessage is the envelope for client-to-controller WebSocket messages // such as subscription requests. type wsClientMessage struct { diff --git a/cmd/kubectl-unbounded/app/net/watch.go b/cmd/kubectl-unbounded/app/net/watch.go index 3d41ad388..dae9f2bf7 100644 --- a/cmd/kubectl-unbounded/app/net/watch.go +++ b/cmd/kubectl-unbounded/app/net/watch.go @@ -47,11 +47,74 @@ type watchOpts struct { // renderSummary, when non-nil, enables the WS summary protocol. On // connect the client sends cluster_summary_subscribe and subsequent // updates arrive as cluster_summary messages rendered via this callback. - // The full-status render callback is still used during HTTP polling - // fallback because the polling endpoint returns full status. + // HTTP polling also projects older controller payloads into this summary. renderSummary func(io.Writer, *clusterSummary, bool) error } +func mergeClusterSummaryDelta(current *clusterSummary, raw []byte) error { + var fields map[string]json.RawMessage + if err := json.Unmarshal(raw, &fields); err != nil { + return err + } + + if fields == nil { + return fmt.Errorf("malformed cluster summary delta") + } + + var nodes struct { + Updated []nodeSummary `json:"nodeSummaries"` + Removed []string `json:"removedNodes"` + } + if err := json.Unmarshal(raw, &nodes); err != nil { + return err + } + + delete(fields, "nodeSummaries") + delete(fields, "removedNodes") + + base, err := json.Marshal(current) + if err != nil { + return err + } + + metadata, err := json.Marshal(fields) + if err != nil { + return err + } + + merged, err := shallowMergeJSON(base, metadata) + if err != nil { + return err + } + + var result clusterSummary + if err := json.Unmarshal(merged, &result); err != nil { + return err + } + + byName := make(map[string]nodeSummary, len(result.NodeSummaries)) + for _, node := range result.NodeSummaries { + byName[node.Name] = node + } + + for _, name := range nodes.Removed { + delete(byName, name) + } + + for _, node := range nodes.Updated { + byName[node.Name] = node + } + + result.NodeSummaries = make([]nodeSummary, 0, len(byName)) + for _, node := range byName { + result.NodeSummaries = append(result.NodeSummaries, node) + } + + *current = result + + return nil +} + // mergeStatusDelta applies an incremental delta to the current cluster status. func mergeStatusDelta(current *clusterStatusResponse, deltaRaw json.RawMessage) error { var delta clusterStatusDelta @@ -411,7 +474,7 @@ func renderWatchScreenSummary( // runWatch connects via WebSocket and renders live-updating data to the terminal. // On WebSocket failure it reconnects with exponential backoff and falls back to -// HTTP polling via fetchClusterStatus while disconnected. +// HTTP overview polling while disconnected. // // When wopts is non-nil and wopts.renderSummary is set, the client subscribes // to the lightweight cluster_summary protocol. If the controller supports it, @@ -439,6 +502,8 @@ func runWatch( ctx, stop := signal.NotifyContext(ctx, os.Interrupt) defer stop() + cmd.SetContext(ctx) + // Enter raw mode for 'q' keypress detection. Cleanup is in the main // function scope so the terminal is always restored, even on error exits. var terminalRestored bool @@ -506,6 +571,7 @@ func runWatch( var currentSummary *clusterSummary summaryInitialized := false + summaryProtocol := false const ( backoffMin = 1 * time.Second @@ -515,6 +581,48 @@ func runWatch( backoff := backoffMin + poll := func() error { + var pollErr error + + if summaryMode { + var summary clusterSummary + + summary, pollErr = fetchClusterSummary(rt, cmd, fetchOpts) + if pollErr == nil { + currentSummary = &summary + summaryInitialized = true + lastSeq = summary.Seq + } + } else { + var status clusterStatusResponse + + status, pollErr = fetchClusterStatus(rt, cmd, fetchOpts) + if pollErr == nil { + current = status + initialized = true + } + } + + if pollErr == nil { + lastUpdate = time.Now() + connState = "Polling" + } else if !initialized && !summaryInitialized { + connState = "Disconnected" + } + + if summaryMode && summaryInitialized { + return renderWatchScreenSummary(os.Stdout, currentSummary, connState, + lastSeq, lastUpdate, useColor, wopts.renderSummary) + } + + if initialized { + return renderWatchScreen(os.Stdout, os.Stdout, current, connState, + lastSeq, lastUpdate, useColor, render) + } + + return nil + } + for { if ctx.Err() != nil { _, _ = fmt.Fprintln(os.Stdout) //nolint:errcheck @@ -537,6 +645,8 @@ func runWatch( conn.SetReadLimit(32 * 1024 * 1024) + summaryProtocol = false + // Subscribe to the summary protocol when configured. If the // controller does not support it the message is silently ignored // and the client keeps receiving cluster_status / cluster_status_delta. @@ -568,6 +678,19 @@ func runWatch( switch msg.Type { case "cluster_status": + if summaryMode { + summary, err := decodeClusterSummary(msg.Data) + if err != nil { + continue + } + + currentSummary = &summary + summaryInitialized = true + lastUpdate = time.Now() + + break + } + if err := json.Unmarshal(msg.Data, ¤t); err != nil { continue } @@ -575,6 +698,20 @@ func runWatch( initialized = true lastUpdate = time.Now() case "cluster_status_delta": + if summaryMode { + if summaryProtocol { + continue + } + + // An old controller's partial detail delta needs a heavy base. + // Refresh the overview instead of retaining that base. + if err := poll(); err != nil { + return err + } + + continue + } + if !initialized { continue } @@ -602,8 +739,21 @@ func runWatch( currentSummary = &summary summaryInitialized = true + summaryProtocol = true lastSeq = summary.Seq lastUpdate = time.Now() + case "cluster_summary_delta": + if !summaryMode || !summaryInitialized { + continue + } + + if err := mergeClusterSummaryDelta(currentSummary, msg.Data); err != nil { + continue + } + + summaryProtocol = true + lastSeq = currentSummary.Seq + lastUpdate = time.Now() default: continue } @@ -627,24 +777,8 @@ func runWatch( // Polling fallback while waiting to reconnect WebSocket. // Poll once before applying the backoff wait. - pollStatus, pollErr := fetchClusterStatus(rt, cmd, fetchOpts) - if pollErr == nil { - current = pollStatus - initialized = true - lastUpdate = time.Now() - connState = "Polling" - } else { - if !initialized { - connState = "Disconnected" - } - // Keep connState as "Polling" if we had data before. - } - - if initialized { - if err := renderWatchScreen(os.Stdout, os.Stdout, current, connState, - lastSeq, lastUpdate, useColor, render); err != nil { - return err - } + if err := poll(); err != nil { + return err } // Backoff wait before attempting WebSocket reconnection. @@ -666,21 +800,8 @@ func runWatch( waited += sleepDur // Poll during the backoff window. - pollStatus, pollErr := fetchClusterStatus(rt, cmd, fetchOpts) - if pollErr == nil { - current = pollStatus - initialized = true - lastUpdate = time.Now() - connState = "Polling" - } else if !initialized { - connState = "Disconnected" - } - - if initialized { - if err := renderWatchScreen(os.Stdout, os.Stdout, current, connState, - lastSeq, lastUpdate, useColor, render); err != nil { - return err - } + if err := poll(); err != nil { + return err } }