From 6fce50b9b3861484462293abc3dd0b59a56bcb9b Mon Sep 17 00:00:00 2001 From: Ravi Shankar Date: Fri, 7 Aug 2026 14:28:00 -0700 Subject: [PATCH] feat(dsx): provider implementation Signed-off-by: Ravi Shankar --- docs/overview.md | 2 + docs/providers/dsx.md | 170 ++++++ pkg/providers/dsx/client.go | 109 ++++ pkg/providers/dsx/client_test.go | 256 ++++++++ pkg/providers/dsx/instance_topology.go | 194 ++++-- pkg/providers/dsx/instance_topology_test.go | 618 ++++++++++++++++++++ pkg/providers/dsx/provider.go | 68 ++- pkg/providers/dsx/provider_sim.go | 60 +- pkg/providers/dsx/provider_sim_test.go | 29 +- pkg/providers/dsx/provider_test.go | 91 +++ pkg/registry/registry.go | 1 + 11 files changed, 1515 insertions(+), 83 deletions(-) create mode 100644 docs/providers/dsx.md create mode 100644 pkg/providers/dsx/client.go create mode 100644 pkg/providers/dsx/client_test.go create mode 100644 pkg/providers/dsx/instance_topology_test.go create mode 100644 pkg/providers/dsx/provider_test.go diff --git a/docs/overview.md b/docs/overview.md index 93e552f1..22030e6f 100644 --- a/docs/overview.md +++ b/docs/overview.md @@ -39,6 +39,7 @@ Currently supported providers: - [Nebius](./providers/nebius.md) - [Nscale](./providers/nscale.md) - [Lambda](./providers/lambdai.md) +- [DSX](./providers/dsx.md) - [NetQ](./providers/netq.md) - [DRA](./providers/dra.md) — provides Slinky block topology from pre-existing `nvidia.com/gpu.clique` labels; it does not discover the backend switch fabric - [InfiniBand (bare-metal)](./providers/infiniband.md#infiniband-bm-bare-metal) @@ -58,6 +59,7 @@ Currently supported engines: | Scenario | Recommended provider | |---|---| | Cloud cluster (AWS, GCP, OCI, Nebius, Nscale, Lambda) | Use the matching CSP provider | +| DSX cluster | [DSX](./providers/dsx.md) | | Spectrum-X fabric | [NetQ](./providers/netq.md) | | Multi-Node NVLink (MNNVL), including cross-partition fabric locality | [NetQ](./providers/netq.md) or [InfiniBand (Kubernetes)](./providers/infiniband.md#infiniband-k8s-kubernetes) | | MNNVL with Slinky, workloads contained within one NVLink partition, and `nvidia.com/gpu.clique` present | [DRA](./providers/dra.md) | diff --git a/docs/providers/dsx.md b/docs/providers/dsx.md new file mode 100644 index 00000000..2ab2cc93 --- /dev/null +++ b/docs/providers/dsx.md @@ -0,0 +1,170 @@ +# DSX Topology Provider + +The `dsx` topology provider reads topology data from the **DSX Topology API** and converts it into Topograph's canonical topology graph. + +The provider calls `GET /v1/topology/nodes`, which returns an ordered list of switch adjacency entries. Each entry maps a switch name to the set of downstream switches and compute nodes it serves. From this it builds a switch tree (for Slurm `topology/tree` or Kubernetes labels) and, when NVLink domain IDs are present, an accelerator domain map (for `topology/block`). + +All `topology.ComputeInstances` groups in the request are aggregated into a single node-ID list sent to this endpoint. The `Region` field on each group and any VPC ID are not used — the DSX service infers the scope from the caller's identity. + +Authentication and rate limiting are enforced by the Envoy proxy sidecar before requests reach this service. Topograph sends an optional `Authorization: Bearer ` header; when no token is configured, the Envoy sidecar is expected to supply SVID-based authentication transparently for in-cluster callers. + +## When to Use This Provider + +Use this provider for **DSX clusters** where the DSX Topology API is the topology source. It works with both the Slurm engine (generating `topology.conf`) and the Kubernetes engine (labeling nodes). + +The request's `nodes` list maps provider node IDs to hostnames. All entries across every group are merged into a single paginated sequence of API calls against `GET /v1/topology/nodes`. + +## Prerequisites + +- The DSX Topology API endpoint reachable from the Topograph host (`base_url`) +- The caller's SVID injected by Envoy (in-cluster, zero-config), **or** a Bearer token with permission to read topology + +## Credentials + +| Field | Required | Description | +|---|---|---| +| `token` | No | Bearer token sent as `Authorization: Bearer `. Omit when running in-cluster and relying on the Envoy sidecar for SVID-based authentication | + +Store the token in a YAML file when needed: + +```yaml +token: +``` + +Reference that file from the Topograph config: + +```yaml +credentialsPath: /etc/topograph/dsx-credentials.yaml +``` + +Credentials can also be supplied directly in the topology request payload under `provider.creds`. + +## Parameters + +| Field | Required | Description | +|---|---|---| +| `base_url` | Yes | Base URL for the DSX Topology API, for example `https://topology.example.com` | +| `trimTiers` | No | Number of highest topology tiers to trim from output. Defaults to `0` | + +The top-level Topograph `pageSize` setting controls the `page_size` query parameter for paginated topology requests (default 100, max 1000 per the API). The `region` field on each `nodes` entry is not used by this provider — the DSX service infers the scope from the caller's identity. + +## Configuration + +Example Topograph config for Slurm: + +```yaml +http: + port: 49021 + ssl: false + +provider: dsx +engine: slurm + +requestAggregationDelay: 15s +credentialsPath: /etc/topograph/dsx-credentials.yaml + +providerParams: + base_url: https://topology.example.com + +engineParams: + plugin: topology/tree + topologyConfigPath: /etc/slurm/topology.conf +``` + +Example request payload: + +```json +{ + "provider": { + "name": "dsx", + "creds": { + "token": "" + }, + "params": { + "base_url": "https://topology.example.com" + } + }, + "engine": { + "name": "slurm", + "params": { + "plugin": "topology/tree" + } + }, + "nodes": [ + { + "region": "", + "instances": { + "": "node001", + "": "node002" + } + } + ] +} +``` + +When running in-cluster without a token, omit the `creds` field entirely — the Envoy sidecar supplies SVID authentication. + +## How It Works + +The provider aggregates node IDs from all `topology.ComputeInstances` groups in the request (the `Region` field on each group is ignored) and sends them as the `node_ids` comma-separated query parameter. It pages through the single global endpoint until the response carries an empty `next_page_token`: + +```text +GET /v1/topology/nodes?node_ids=,&page_size= +Authorization: Bearer # omitted when using Envoy SVID +``` + +The response envelope: + +```json +{ + "switches": [ + { "": { "switches": [""], "nodes": [] } }, + { "": { "switches": [], "nodes": [{ "node_id": "", "accelerated_network_id": "" }] } } + ], + "next_page_token": "" +} +``` + +`switches` is an **ordered list of single-key objects**. Non-leaf entries carry `switches` (their downstream switches); leaf entries carry `nodes` (the compute nodes attached to them). This ordering reflects the fabric hierarchy from core to leaf. + +Switch entries from every page are accumulated before any graph is built. Only after the final page (empty `next_page_token`) are parent-child relationships resolved and instance topologies emitted. This ensures cross-page ancestry is correct — for example, a spine switch returned on page 1 is correctly recognised as the parent of a leaf switch returned on page 2. + +Each node is translated as follows: + +| API field | Topograph field | +|---|---| +| `node_id` | Instance ID (matched against the request's instance-to-hostname map) | +| Switch that lists the node under its `nodes` | Leaf (tier 0, closest to node) | +| Parent of the leaf (from `switches` adjacency) | Spine (tier 1) | +| Parent of the spine | Core (tier 2) | +| `accelerated_network_id` | Accelerator / NVLink domain (`XclrDomainID`) | + +Tier assignment is closest-first: tier 0 is the leaf switch directly attached to the node, tier 1 is the spine, and tier 2 is the core. When `accelerated_network_id` is non-empty the node is placed into that NVLink domain, enabling `topology/block` output. + +## Verifying the Output + +Sanity-check the API directly (in-cluster, Envoy supplies auth): + +```bash +curl -s "$BASE_URL/v1/topology/nodes?node_ids=node1,node2" | jq . +``` + +Or with an explicit token: + +```bash +curl -s -H "Authorization: Bearer $TOKEN" \ + "$BASE_URL/v1/topology/nodes?node_ids=node1,node2" | jq . +``` + +Then trigger topology generation and read the result: + +```bash +id=$(curl -s -X POST -H "Content-Type: application/json" -d @payload.json http://localhost:49021/v1/generate) +curl -s "http://localhost:49021/v1/topology?uid=$id" +``` + +For the Slurm engine, verify the generated `topology.conf` reflects the expected switch hierarchy for your nodes. + +## Simulation + +A `dsx-sim` provider variant is registered for testing without a live API. Instead of calling the topology API, it reads a YAML simulation model and serves it through the same translation path. Select it with `provider: dsx-sim` and point it at a model file via the `modelFileName` parameter; see [Test Mode and Test Provider](./test.md) for the model-file format and simulation parameters. diff --git a/pkg/providers/dsx/client.go b/pkg/providers/dsx/client.go new file mode 100644 index 00000000..1f810089 --- /dev/null +++ b/pkg/providers/dsx/client.go @@ -0,0 +1,109 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dsx + +import ( + "context" + "encoding/json" + "net/http" + "strconv" + "strings" + "time" + + "k8s.io/klog/v2" + + "github.com/NVIDIA/topograph/internal/httperr" + "github.com/NVIDIA/topograph/internal/httpreq" +) + +const ( + pathNodes = "/v1/topology/nodes" + + // requestTimeout bounds each individual page fetch — connection, response + // headers, and full body — including up to httpreq.maxRetries (5) retries + // with httpreq.maxRetryAfter (5 min) Retry-After back-off. The 30-minute + // ceiling allows legitimate rate-limit retry sequences to complete while + // still terminating a permanently stalled connection. + requestTimeout = 30 * time.Minute + + // minPageSize is the floor applied to caller-supplied page sizes. The DSX + // API defaults to 100; values below this minimum create excessive pages, + // which can exhaust maxPaginationPages before a finite response completes. + minPageSize = 100 +) + +type httpClient struct { + baseURL string + token string +} + +// NewHTTPClient returns a Client that calls the DSX Topology API. +// If token is empty the Authorization header is omitted and the Envoy sidecar +// is expected to supply SVID-based authentication transparently. +func NewHTTPClient(baseURL, token string) *httpClient { + return &httpClient{baseURL: baseURL, token: token} +} + +func (c *httpClient) GetTopology(ctx context.Context, vpcID string, nodeIDs []string, pageSize int, pageToken string) (*TopologyResponse, error) { + // Derive callCtx from context.Background() so requestTimeout is the true + // per-page deadline, independent of any total-generation deadline already + // set on ctx. context.AfterFunc propagates ctx cancellation so that a + // caller-side abort (e.g. genCtx expiry) still terminates the page fetch. + callCtx, cancel := context.WithTimeout(context.Background(), requestTimeout) + defer cancel() + stopPropagation := context.AfterFunc(ctx, cancel) + defer stopPropagation() + + path := pathNodes + if vpcID != "" { + path = "/v1/topology/vpcs/" + vpcID + "/nodes" + } + + headers := map[string]string{} + if c.token != "" { + headers["Authorization"] = "Bearer " + c.token + } + + if pageSize > 0 && pageSize < minPageSize { + klog.Warningf("DSX page size %d is below minimum %d; clamping to avoid excessive pagination", pageSize, minPageSize) + pageSize = minPageSize + } + + query := map[string]string{} + if len(nodeIDs) > 0 { + query["node_ids"] = strings.Join(nodeIDs, ",") + } + if pageSize > 0 { + query["page_size"] = strconv.Itoa(pageSize) + } + if pageToken != "" { + query["page_token"] = pageToken + } + + f := httpreq.GetRequestFunc(callCtx, http.MethodGet, headers, query, nil, c.baseURL, path) + body, httpErr := httpreq.DoRequestWithRetries(f, false) + if httpErr != nil { + return nil, httpErr + } + + var resp TopologyResponse + if err := json.Unmarshal(body, &resp); err != nil { + return nil, httperr.NewError(http.StatusBadGateway, err.Error()) + } + + return &resp, nil +} diff --git a/pkg/providers/dsx/client_test.go b/pkg/providers/dsx/client_test.go new file mode 100644 index 00000000..8cee6f6f --- /dev/null +++ b/pkg/providers/dsx/client_test.go @@ -0,0 +1,256 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dsx + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// capture holds the server-side view of one request. +type capture struct { + path string + query url.Values + authHdr string +} + +// newTestServer starts an httptest server that records each request into a +// *capture and responds with the given status and body. +func newTestServer(t *testing.T, status int, body []byte) (*httptest.Server, *capture) { + t.Helper() + cap := &capture{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cap.path = r.URL.Path + cap.query = r.URL.Query() + cap.authHdr = r.Header.Get("Authorization") + w.WriteHeader(status) + _, _ = w.Write(body) + })) + t.Cleanup(srv.Close) + return srv, cap +} + +// TestHTTPClientGetTopologyContextCancellation verifies that cancelling the +// caller's context causes GetTopology to return promptly with an error rather +// than waiting for requestTimeout. +func TestHTTPClientGetTopologyContextCancellation(t *testing.T) { + ready := make(chan struct{}) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(ready) + <-r.Context().Done() // hold until the client disconnects + })) + defer srv.Close() + + ctx, cancel := context.WithCancel(context.Background()) + client := NewHTTPClient(srv.URL, "") + + type result struct { + resp *TopologyResponse + err error + } + ch := make(chan result, 1) + go func() { + resp, err := client.GetTopology(ctx, "", nil, 0, "") + ch <- result{resp, err} + }() + + <-ready // server has the request; now cancel the caller's context + cancel() + + select { + case res := <-ch: + require.Error(t, res.err) + require.Nil(t, res.resp) + case <-time.After(5 * time.Second): + t.Fatal("GetTopology did not return promptly after context cancellation") + } +} + +func validResponseBody(t *testing.T) ([]byte, TopologyResponse) { + t.Helper() + resp := TopologyResponse{ + Switches: []map[string]SwitchAdjacency{{"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}}, + NextPageToken: "tok2", + } + b, err := json.Marshal(resp) + require.NoError(t, err) + return b, resp +} + +func TestHTTPClientGetTopology(t *testing.T) { + validBody, validResp := validResponseBody(t) + + tests := []struct { + name string + token string + vpcID string + nodeIDs []string + pageSize int + pageToken string + status int + body []byte + wantPath string + wantQuery map[string]string // key → expected value; "" means must be absent + wantAuth string // expected Authorization header value, or "" if none + wantErr bool + wantResp *TopologyResponse + }{ + { + name: "global nodes path when vpcID is empty", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantResp: &validResp, + }, + { + name: "VPC path when vpcID is set", + vpcID: "vpc-123", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/vpcs/vpc-123/nodes", + wantResp: &validResp, + }, + { + name: "query params: node_ids comma-joined, page_size and page_token encoded", + nodeIDs: []string{"n1", "n2", "n3"}, + pageSize: 200, + pageToken: "cursor-abc", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantQuery: map[string]string{ + "node_ids": "n1,n2,n3", + "page_size": "200", + "page_token": "cursor-abc", + }, + wantResp: &validResp, + }, + { + name: "page_size below minimum is clamped to minPageSize", + pageSize: 1, + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantQuery: map[string]string{ + "page_size": strconv.Itoa(minPageSize), + }, + wantResp: &validResp, + }, + { + name: "Authorization header sent when token is set", + token: "secret-token", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantAuth: "Bearer secret-token", + wantResp: &validResp, + }, + { + name: "no Authorization header when token is empty", + token: "", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantAuth: "", // expect header absent + wantResp: &validResp, + }, + { + name: "empty nodeIDs omits node_ids param", + nodeIDs: nil, + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantQuery: map[string]string{"node_ids": ""}, + wantResp: &validResp, + }, + { + name: "zero pageSize omits page_size param", + pageSize: 0, + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantQuery: map[string]string{"page_size": ""}, + wantResp: &validResp, + }, + { + name: "non-2xx response returns error", + status: http.StatusBadRequest, + body: []byte(`{"detail":"bad node_ids"}`), + wantPath: "/v1/topology/nodes", + wantErr: true, + }, + { + name: "malformed JSON with 200 OK returns error", + status: http.StatusOK, + body: []byte(`{not valid json`), + wantPath: "/v1/topology/nodes", + wantErr: true, + }, + { + name: "complete valid response is parsed correctly", + token: "tok", + status: http.StatusOK, + body: validBody, + wantPath: "/v1/topology/nodes", + wantAuth: "Bearer tok", + wantResp: &validResp, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + srv, cap := newTestServer(t, tc.status, tc.body) + + client := NewHTTPClient(srv.URL, tc.token) + got, err := client.GetTopology(context.Background(), tc.vpcID, tc.nodeIDs, tc.pageSize, tc.pageToken) + + // --- server-side assertions --- + require.Equal(t, tc.wantPath, cap.path) + + for param, want := range tc.wantQuery { + if want == "" { + // Empty expected value means the param must be absent entirely. + // url.Values.Get returns "" for both absent and empty-valued params, + // so check the raw slice to distinguish the two cases. + require.Empty(t, cap.query[param], "query param %q should be absent", param) + } else { + require.Equal(t, want, cap.query.Get(param), + "query param %q: got %q, want %q", param, cap.query.Get(param), want) + } + } + + require.Equal(t, tc.wantAuth, cap.authHdr) + + // --- client-side assertions --- + if tc.wantErr { + require.Error(t, err) + require.Nil(t, got) + } else { + require.NoError(t, err) + require.Equal(t, tc.wantResp, got) + } + }) + } +} diff --git a/pkg/providers/dsx/instance_topology.go b/pkg/providers/dsx/instance_topology.go index 53e123b9..c25839cf 100644 --- a/pkg/providers/dsx/instance_topology.go +++ b/pkg/providers/dsx/instance_topology.go @@ -1,6 +1,17 @@ /* - * Copyright 2026, NVIDIA CORPORATION - * SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. */ package dsx @@ -9,6 +20,8 @@ import ( "context" "fmt" "net/http" + "sort" + "time" "k8s.io/klog/v2" @@ -16,71 +29,180 @@ import ( "github.com/NVIDIA/topograph/pkg/topology" ) +// totalGenerationTimeout is the wall-clock budget for one complete topology +// fetch across all pages. It is the primary bound on how long a single +// topology generation can run: even an API that returns an endless stream of +// unique tokens cannot keep generation active beyond this deadline. +// +// At the default page size of 100 on a healthy server (< 1 s/page), 10 min +// comfortably covers clusters up to 60 000 nodes with margin for retries. +// It is a var so tests can lower it without sleeping for minutes. +var totalGenerationTimeout = 10 * time.Minute + +// maxPaginationPages is a secondary defence: it terminates runaway pagination +// independently of wall-clock time. The client enforces minPageSize=100, so +// 10 000 pages × 100 entries/page covers 1 000 000 entries — well beyond any +// realistic cluster, even with switch-only pages inflating the count. +// It is a var so tests can lower it without allocating thousands of responses. +var maxPaginationPages = 10_000 + func (p *baseProvider) generateInstanceTopology(ctx context.Context, pageSize *int, cis []topology.ComputeInstances) (*topology.ClusterTopology, *httperr.Error) { client, err := p.clientFactory() if err != nil { return nil, httperr.NewError(http.StatusBadGateway, fmt.Sprintf("failed to get client: %v", err)) } + // Build want set and nodeIDs once. Deduplicate IDs that appear in more than + // one ComputeInstances group, and sort for deterministic API requests. + want := make(map[string]struct{}) var nodeIDs []string for _, ci := range cis { for instanceID := range ci.Instances { - nodeIDs = append(nodeIDs, instanceID) + if instanceID == "" { + klog.Warningf("skipping empty instance ID in ComputeInstances") + continue + } + if _, exists := want[instanceID]; !exists { + want[instanceID] = struct{}{} + nodeIDs = append(nodeIDs, instanceID) + } } } + sort.Strings(nodeIDs) + + if len(nodeIDs) == 0 { + return topology.NewClusterTopology(), nil + } pageSizeVal := 0 if pageSize != nil { pageSizeVal = *pageSize } - response, apiErr := client.GetTopology(ctx, "", nodeIDs, pageSizeVal, "") - if apiErr != nil { - return nil, httperr.NewError(http.StatusBadGateway, fmt.Sprintf("API error: %v", apiErr)) + // Phase 1: accumulate all switch entries across pages before resolving topology. + // Building parentOf page-by-page would lose cross-page ancestry (e.g. spine on + // page 1, leaves on page 2). + // + // genCtx provides a single wall-clock deadline for the entire multi-page fetch. + // This is the primary termination bound: no matter how many unique tokens the + // API returns, generation cannot run past totalGenerationTimeout. + // + // seenTokens and the page counter are secondary defences against token cycles + // and runaway page counts within the deadline. + genCtx, cancel := context.WithTimeout(ctx, totalGenerationTimeout) + defer cancel() + + seenTokens := map[string]struct{}{"": {}} + var allSwitches []map[string]SwitchAdjacency + var pageToken string + for page := 1; ; page++ { + if page > maxPaginationPages { + // Deterministic: retrying would hit the same limit. + return nil, httperr.NewError(http.StatusUnprocessableEntity, + fmt.Sprintf("DSX API exceeded maximum page limit (%d)", maxPaginationPages)) + } + response, apiErr := client.GetTopology(genCtx, "", nodeIDs, pageSizeVal, pageToken) + if apiErr != nil { + if ctx.Err() != nil { + // The caller's context was cancelled — report that, not our deadline. + return nil, httperr.NewError(http.StatusUnprocessableEntity, + fmt.Sprintf("context cancelled: %v", apiErr)) + } + if genCtx.Err() != nil { + // Our own total-generation deadline fired. A retry would restart + // with a fresh deadline, repeating the full wait — not retryable. + return nil, httperr.NewError(http.StatusUnprocessableEntity, + fmt.Sprintf("DSX topology generation deadline exceeded: %v", apiErr)) + } + return nil, httperr.NewError(http.StatusBadGateway, fmt.Sprintf("API error: %v", apiErr)) + } + if response == nil { + return nil, httperr.NewError(http.StatusBadGateway, "DSX API returned nil response without error") + } + allSwitches = append(allSwitches, response.Switches...) + if response.NextPageToken == "" { + break + } + if _, seen := seenTokens[response.NextPageToken]; seen { + // Deterministic: retrying would encounter the same cycle. + return nil, httperr.NewError(http.StatusUnprocessableEntity, "DSX API returned a page token cycle") + } + seenTokens[response.NextPageToken] = struct{}{} + pageToken = response.NextPageToken } - return responseToClusterTopology(response, cis), nil + // Phase 2: resolve topology from the complete cross-page switch set. + return buildClusterTopology(allSwitches, want), nil } -// responseToClusterTopology maps switch/node API output to per-instance records for ToGraph. -func responseToClusterTopology(response *TopologyResponse, cis []topology.ComputeInstances) *topology.ClusterTopology { - want := make(map[string]struct{}) - for _, ci := range cis { - for instanceID := range ci.Instances { - want[instanceID] = struct{}{} +// buildClusterTopology translates the complete ordered switch list into per-instance +// topology records. It must be called with the full set of switches across all pages +// so that cross-page parent-child relationships are resolved correctly. +func buildClusterTopology(switches []map[string]SwitchAdjacency, want map[string]struct{}) *topology.ClusterTopology { + // sortedEntryKeys returns the switch names in an entry sorted so that + // multi-key entries (malformed per spec) are processed deterministically. + sortedEntryKeys := func(entry map[string]SwitchAdjacency) []string { + names := make([]string, 0, len(entry)) + for k := range entry { + names = append(names, k) } + sort.Strings(names) + return names } + // First pass: build the parent map from the complete switch list. parentOf := make(map[string]string) - for swName, info := range response.Switches { - for _, child := range info.Switches { - parentOf[child] = swName + for _, entry := range switches { + if len(entry) > 1 { + klog.Warningf("DSX API returned a SwitchEntry with %d keys; expected 1 (ordering non-deterministic)", len(entry)) + } + for _, swName := range sortedEntryKeys(entry) { + for _, child := range entry[swName].Switches { + parentOf[child] = swName + } } } + // Second pass: emit an InstanceTopology for each node attached to a leaf switch. + // emitted guards against duplicate node IDs in malformed API responses. + emitted := make(map[string]struct{}) topo := topology.NewClusterTopology() - for swName, info := range response.Switches { - for _, n := range info.Nodes { - if _, ok := want[n.NodeID]; !ok { - continue - } - leafID := swName - spineID := parentOf[leafID] - coreID := "" - if spineID != "" { - coreID = parentOf[spineID] - } + for _, entry := range switches { + for _, swName := range sortedEntryKeys(entry) { + adj := entry[swName] + for _, n := range adj.Nodes { + if _, ok := want[n.NodeID]; !ok { + continue + } + if _, dup := emitted[n.NodeID]; dup { + klog.Warningf("DSX API returned duplicate node_id %q; ignoring", n.NodeID) + continue + } + emitted[n.NodeID] = struct{}{} - //create the instance topology - inst := &topology.InstanceTopology{ - InstanceID: n.NodeID, - FabricTiers: topology.ClosestFirstFabricTiers(leafID, spineID, coreID), - } - if n.AcceleratedNetworkID != "" { - inst.XclrDomainID = n.AcceleratedNetworkID + // Walk up the parentOf map to collect all fabric tiers closest-first. + // The seen set guards against cycles in malformed API responses. + var tierIDs []string + seen := make(map[string]struct{}) + for id := swName; id != ""; id = parentOf[id] { + if _, cycle := seen[id]; cycle { + klog.Warningf("DSX parentOf map contains a cycle at switch %q; truncating fabric tiers", id) + break + } + seen[id] = struct{}{} + tierIDs = append(tierIDs, id) + } + + inst := &topology.InstanceTopology{ + InstanceID: n.NodeID, + FabricTiers: topology.ClosestFirstFabricTiers(tierIDs...), + } + if n.AcceleratedNetworkID != "" { + inst.XclrDomainID = n.AcceleratedNetworkID + } + klog.V(4).Infof("Adding instance topology %s", inst.String()) + topo.Append(inst) } - klog.V(4).Infof("Adding instance topology %s", inst.String()) - topo.Append(inst) } } diff --git a/pkg/providers/dsx/instance_topology_test.go b/pkg/providers/dsx/instance_topology_test.go new file mode 100644 index 00000000..efda040a --- /dev/null +++ b/pkg/providers/dsx/instance_topology_test.go @@ -0,0 +1,618 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dsx + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/topograph/pkg/topology" +) + +// mockClient is a controllable Client for unit testing pagination and error paths. +type mockClient struct { + responses []*TopologyResponse + errors []error + idx int + pageSizes []int + pageTokens []string + nodeIDsPerCall [][]string +} + +func (m *mockClient) GetTopology(ctx context.Context, _ string, nodeIDs []string, pageSize int, pageToken string) (*TopologyResponse, error) { + cp := make([]string, len(nodeIDs)) + copy(cp, nodeIDs) + m.nodeIDsPerCall = append(m.nodeIDsPerCall, cp) + m.pageSizes = append(m.pageSizes, pageSize) + m.pageTokens = append(m.pageTokens, pageToken) + if err := ctx.Err(); err != nil { + return nil, err + } + i := m.idx + m.idx++ + if i < len(m.errors) && m.errors[i] != nil { + return nil, m.errors[i] + } + if i < len(m.responses) { + return m.responses[i], nil + } + return &TopologyResponse{}, nil +} + +func newProvider(client Client) *baseProvider { + return &baseProvider{ + clientFactory: func() (Client, error) { return client, nil }, + } +} + +// --------------------------------------------------------------------------- +// buildClusterTopology unit tests +// --------------------------------------------------------------------------- + +func TestBuildClusterTopology(t *testing.T) { + tests := []struct { + name string + switches []map[string]SwitchAdjacency + want map[string]struct{} + wantLen int + checkInst func(t *testing.T, topo *topology.ClusterTopology) + }{ + { + name: "empty switches returns empty topology", + switches: nil, + want: map[string]struct{}{"n1": {}}, + wantLen: 0, + }, + { + name: "empty want set returns empty topology", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{}, + wantLen: 0, + }, + { + name: "2-tier: leaf with no parent produces single-tier FabricTiers", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + inst := topo.Instances[0] + require.Equal(t, "n1", inst.InstanceID) + require.Len(t, inst.FabricTiers, 1, "2-tier network: only leaf tier expected") + require.Equal(t, "leaf", inst.FabricTiers[0].ID) + }, + }, + { + name: "3-tier: core→spine→leaf produces 3-element FabricTiers, closest-first", + switches: []map[string]SwitchAdjacency{ + {"core": {Switches: []string{"spine"}}}, + {"spine": {Switches: []string{"leaf"}}}, + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + inst := topo.Instances[0] + require.Len(t, inst.FabricTiers, 3, "3-tier network: leaf, spine, core expected") + require.Equal(t, "leaf", inst.FabricTiers[0].ID) + require.Equal(t, "spine", inst.FabricTiers[1].ID) + require.Equal(t, "core", inst.FabricTiers[2].ID) + }, + }, + { + name: "4-tier: ultra-core→core→spine→leaf produces 4-element FabricTiers, closest-first", + switches: []map[string]SwitchAdjacency{ + {"ultra-core": {Switches: []string{"core"}}}, + {"core": {Switches: []string{"spine"}}}, + {"spine": {Switches: []string{"leaf"}}}, + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + inst := topo.Instances[0] + require.Len(t, inst.FabricTiers, 4, "4-tier network: leaf, spine, core, ultra-core expected") + require.Equal(t, "leaf", inst.FabricTiers[0].ID) + require.Equal(t, "spine", inst.FabricTiers[1].ID) + require.Equal(t, "core", inst.FabricTiers[2].ID) + require.Equal(t, "ultra-core", inst.FabricTiers[3].ID) + }, + }, + { + name: "NVLink accelerated_network_id is propagated to XclrDomainID", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1", AcceleratedNetworkID: "nvl-domain-1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Equal(t, "nvl-domain-1", topo.Instances[0].XclrDomainID) + }, + }, + { + name: "empty accelerated_network_id leaves XclrDomainID unset", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1", AcceleratedNetworkID: ""}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Empty(t, topo.Instances[0].XclrDomainID) + }, + }, + { + name: "nodes not in want set are filtered out", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}, {NodeID: "n2"}, {NodeID: "n3"}}}}, + }, + want: map[string]struct{}{"n1": {}, "n3": {}}, + wantLen: 2, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + ids := make(map[string]struct{}, 2) + for _, inst := range topo.Instances { + ids[inst.InstanceID] = struct{}{} + } + require.Contains(t, ids, "n1") + require.Contains(t, ids, "n3") + require.NotContains(t, ids, "n2") + }, + }, + { + name: "node in want but absent from response is not emitted", + switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}, "n2": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Equal(t, "n1", topo.Instances[0].InstanceID) + }, + }, + { + name: "multiple leaves under one spine: each node gets correct leaf and shared spine", + switches: []map[string]SwitchAdjacency{ + {"spine": {Switches: []string{"leaf1", "leaf2"}}}, + {"leaf1": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + {"leaf2": {Nodes: []NodeInfo{{NodeID: "n2"}}}}, + }, + want: map[string]struct{}{"n1": {}, "n2": {}}, + wantLen: 2, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + byID := make(map[string]*topology.InstanceTopology, 2) + for _, inst := range topo.Instances { + byID[inst.InstanceID] = inst + } + n1 := byID["n1"] + require.Len(t, n1.FabricTiers, 2) + require.Equal(t, "leaf1", n1.FabricTiers[0].ID) + require.Equal(t, "spine", n1.FabricTiers[1].ID) + + n2 := byID["n2"] + require.Len(t, n2.FabricTiers, 2) + require.Equal(t, "leaf2", n2.FabricTiers[0].ID) + require.Equal(t, "spine", n2.FabricTiers[1].ID) + }, + }, + { + name: "switch entry with only downstream switches and no nodes emits nothing", + switches: []map[string]SwitchAdjacency{ + {"core": {Switches: []string{"spine"}}}, + {"spine": {Switches: []string{"leaf"}}}, + // leaf is referenced but never defined with nodes + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 0, + }, + { + name: "multi-key SwitchEntry is processed deterministically", + // The API spec requires single-key entries, but a malformed response may + // carry multiple keys in one entry. Both switches must be processed and + // the result must be idempotent across repeated calls (no map-order + // dependence on swName selection). + switches: []map[string]SwitchAdjacency{ + { + "leaf-b": {Nodes: []NodeInfo{{NodeID: "n2"}}}, + "leaf-a": {Nodes: []NodeInfo{{NodeID: "n1"}}}, + }, + }, + want: map[string]struct{}{"n1": {}, "n2": {}}, + wantLen: 2, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + // sortedEntryKeys processes "leaf-a" before "leaf-b", so n1 must + // appear before n2. An order-insensitive check would pass even if + // sortedEntryKeys were removed and map iteration changed the order. + require.Equal(t, "n1", topo.Instances[0].InstanceID) + require.Equal(t, "leaf-a", topo.Instances[0].FabricTiers[0].ID) + require.Equal(t, "n2", topo.Instances[1].InstanceID) + require.Equal(t, "leaf-b", topo.Instances[1].FabricTiers[0].ID) + }, + }, + { + name: "SwitchEntry keyed by empty string produces no empty FabricTier IDs", + // A malformed entry with key "" would make leafID="" and unconditionally + // add it to tierIDs, producing FabricTier{ID:""}. The guard must skip it. + switches: []map[string]SwitchAdjacency{ + {"": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Empty(t, topo.Instances[0].FabricTiers, + "an entry keyed by \"\" must produce no fabric tiers") + for _, tier := range topo.Instances[0].FabricTiers { + require.NotEmpty(t, tier.ID, "FabricTier must not have an empty ID") + } + }, + }, + { + name: "cyclic switch relationship terminates and emits unique tier IDs", + // sw-a lists sw-b as child; sw-b lists sw-a as child — a direct 2-cycle. + // parentOf["sw-a"] = "sw-b" and parentOf["sw-b"] = "sw-a", so without + // the duplicate-tier guard coreID would equal leafID ("sw-a") and the + // FabricTiers slice would contain "sw-a" twice. + switches: []map[string]SwitchAdjacency{ + {"sw-a": {Switches: []string{"sw-b"}, Nodes: []NodeInfo{{NodeID: "n1"}}}}, + {"sw-b": {Switches: []string{"sw-a"}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Equal(t, "n1", topo.Instances[0].InstanceID) + tiers := topo.Instances[0].FabricTiers + require.NotEmpty(t, tiers) + seen := make(map[string]struct{}, len(tiers)) + for _, tier := range tiers { + _, dup := seen[tier.ID] + require.False(t, dup, "duplicate tier ID %q in FabricTiers", tier.ID) + seen[tier.ID] = struct{}{} + } + }, + }, + { + name: "repeated NodeID across switch entries is emitted only once", + // A malformed response lists n1 under two different leaf switches. + // The deduplication guard must suppress the second occurrence. + switches: []map[string]SwitchAdjacency{ + {"leaf1": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + {"leaf2": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + want: map[string]struct{}{"n1": {}}, + wantLen: 1, + checkInst: func(t *testing.T, topo *topology.ClusterTopology) { + require.Equal(t, "n1", topo.Instances[0].InstanceID) + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + topo := buildClusterTopology(tc.switches, tc.want) + require.Len(t, topo.Instances, tc.wantLen) + if tc.checkInst != nil { + tc.checkInst(t, topo) + } + }) + } +} + +// --------------------------------------------------------------------------- +// generateInstanceTopology — pagination and error paths via mock client +// --------------------------------------------------------------------------- + +func TestGenerateInstanceTopologyCrossPageAncestry(t *testing.T) { + // Spine and core arrive on page 1; their leaf child and node arrive on page 2. + // Without the two-phase accumulation fix, n1 would get a 1-tier topology because + // parentOf would only be built from page 2's switches. + mc := &mockClient{ + responses: []*TopologyResponse{ + { + Switches: []map[string]SwitchAdjacency{ + {"core": {Switches: []string{"spine"}}}, + {"spine": {Switches: []string{"leaf"}}}, + }, + NextPageToken: "page2", + }, + { + Switches: []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + }, + NextPageToken: "", + }, + }, + } + + ps := 50 + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + topo, err := newProvider(mc).generateInstanceTopology(context.Background(), &ps, cis) + + require.Nil(t, err) + require.Len(t, topo.Instances, 1) + inst := topo.Instances[0] + require.Len(t, inst.FabricTiers, 3, "cross-page ancestry must produce full 3-tier hierarchy") + require.Equal(t, "leaf", inst.FabricTiers[0].ID) + require.Equal(t, "spine", inst.FabricTiers[1].ID) + require.Equal(t, "core", inst.FabricTiers[2].ID) + + // Verify token propagation and page size forwarding on every call. + require.Equal(t, []string{"", "page2"}, mc.pageTokens) + require.Equal(t, []int{50, 50}, mc.pageSizes) +} + +func TestGenerateInstanceTopologyMultiPageNVLink(t *testing.T) { + // Verify that NVLink domain IDs are correctly propagated through pagination. + mc := &mockClient{ + responses: []*TopologyResponse{ + { + Switches: []map[string]SwitchAdjacency{ + {"leaf1": {Nodes: []NodeInfo{{NodeID: "n1", AcceleratedNetworkID: "nvl-a"}}}}, + }, + NextPageToken: "p2", + }, + { + Switches: []map[string]SwitchAdjacency{ + {"leaf2": {Nodes: []NodeInfo{{NodeID: "n2", AcceleratedNetworkID: "nvl-b"}}}}, + }, + NextPageToken: "", + }, + }, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1", "n2": "node2"}}} + topo, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.Nil(t, err) + require.Len(t, topo.Instances, 2) + byID := make(map[string]*topology.InstanceTopology) + for _, inst := range topo.Instances { + byID[inst.InstanceID] = inst + } + require.Equal(t, "nvl-a", byID["n1"].XclrDomainID) + require.Equal(t, "nvl-b", byID["n2"].XclrDomainID) +} + +func TestGenerateInstanceTopologyAPIErrorOnSecondPage(t *testing.T) { + mc := &mockClient{ + responses: []*TopologyResponse{ + { + Switches: []map[string]SwitchAdjacency{{"spine": {Switches: []string{"leaf"}}}}, + NextPageToken: "p2", + }, + }, + errors: []error{nil, errors.New("backend unavailable")}, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusBadGateway, err.Code()) + require.Contains(t, err.Error(), "backend unavailable") +} + +func TestGenerateInstanceTopologyDirectCyclePageToken(t *testing.T) { + // A→A: the API echoes the same token on the very next response. + mc := &mockClient{ + responses: []*TopologyResponse{ + {NextPageToken: "stuck"}, + {NextPageToken: "stuck"}, + }, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusUnprocessableEntity, err.Code()) + require.Contains(t, err.Error(), "page token cycle") + require.Equal(t, 2, mc.idx, "pagination must stop immediately after the repeated token is received") +} + +func TestGenerateInstanceTopologyNonConsecutiveCyclePageToken(t *testing.T) { + // A→B→A: the cycle skips one hop, so comparing only against the previous + // token would miss it and loop indefinitely. + mc := &mockClient{ + responses: []*TopologyResponse{ + {NextPageToken: "A"}, + {NextPageToken: "B"}, + {NextPageToken: "A"}, // revisits "A" — cycle detected here + }, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusUnprocessableEntity, err.Code()) + require.Contains(t, err.Error(), "page token cycle") + require.Equal(t, 3, mc.idx, "pagination must stop immediately after the revisited token is received") +} + +func TestGenerateInstanceTopologyContextCancelled(t *testing.T) { + // A cancelled context must surface as an error rather than hanging. + // This covers the stall scenario: the real HTTP client honours ctx.Done() + // during dial, header wait, and body read; the mock checks it explicitly. + ctx, cancel := context.WithCancel(context.Background()) + cancel() // cancel before the first call + + mc := &mockClient{ + responses: []*TopologyResponse{{NextPageToken: ""}}, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(mc).generateInstanceTopology(ctx, nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusUnprocessableEntity, err.Code()) + require.Contains(t, err.Error(), "context cancelled") +} + +func TestGenerateInstanceTopologyTotalTimeout(t *testing.T) { + // Lower the total deadline so the test completes quickly. Page 1 returns + // immediately; page 2 blocks on ctx.Done(), simulating a stalled server. + // The total deadline fires and the call must return 422 Unprocessable Entity. + oldTimeout := totalGenerationTimeout + totalGenerationTimeout = 50 * time.Millisecond + defer func() { totalGenerationTimeout = oldTimeout }() + + stallOnPage2 := &stallingMockClient{ + firstResponse: &TopologyResponse{NextPageToken: "p2"}, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(stallingPage2Client(stallOnPage2)).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusUnprocessableEntity, err.Code()) + require.Contains(t, err.Error(), "deadline exceeded") +} + +// stallingMockClient returns firstResponse on the first call, then blocks +// until its context is cancelled on every subsequent call. +type stallingMockClient struct { + firstResponse *TopologyResponse + called bool +} + +func (s *stallingMockClient) GetTopology(ctx context.Context, _ string, _ []string, _ int, _ string) (*TopologyResponse, error) { + if !s.called { + s.called = true + return s.firstResponse, nil + } + <-ctx.Done() + return nil, ctx.Err() +} + +func stallingPage2Client(s *stallingMockClient) Client { return s } + +func TestGenerateInstanceTopologyPageLimit(t *testing.T) { + // Lower the cap so the test doesn't need thousands of mock responses. + old := maxPaginationPages + maxPaginationPages = 3 + defer func() { maxPaginationPages = old }() + + // Four pages of unique tokens — exceeds the cap of 3. + mc := &mockClient{ + responses: []*TopologyResponse{ + {NextPageToken: "t1"}, + {NextPageToken: "t2"}, + {NextPageToken: "t3"}, + {NextPageToken: "t4"}, + }, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusUnprocessableEntity, err.Code()) + require.Contains(t, err.Error(), "maximum page limit") + require.Equal(t, 3, mc.idx, "guard fires before the 4th GetTopology call") +} + +func TestGenerateInstanceTopologyEmptyInstanceID(t *testing.T) { + // An empty-key entry in Instances must be skipped; the API must not be called. + mc := &mockClient{} + cis := []topology.ComputeInstances{{Instances: map[string]string{"": "node0"}}} + + topo, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + require.Nil(t, err) + require.Equal(t, 0, topo.Len()) + require.Equal(t, 0, mc.idx, "GetTopology must not be called for an all-empty instance ID list") +} + +func TestGenerateInstanceTopologyEmptyInstances(t *testing.T) { + // Empty cis must short-circuit before any API call is made. + mc := &mockClient{} + + topo, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, nil) + require.Nil(t, err) + require.Equal(t, 0, topo.Len()) + require.Equal(t, 0, mc.idx, "no API calls should be made when nodeIDs is empty") +} + +func TestGenerateInstanceTopologyNilResponse(t *testing.T) { + // A client that returns (nil, nil) must not panic; expect a 502. + nilClient := &nilResponseClient{} + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1"}}} + _, err := newProvider(nilClient).generateInstanceTopology(context.Background(), nil, cis) + + require.NotNil(t, err) + require.Equal(t, http.StatusBadGateway, err.Code()) + require.Contains(t, err.Error(), "nil response") +} + +type nilResponseClient struct{} + +func (nilResponseClient) GetTopology(_ context.Context, _ string, _ []string, _ int, _ string) (*TopologyResponse, error) { + return nil, nil +} + +func TestGenerateInstanceTopologySinglePage(t *testing.T) { + // Happy path with a single complete response page. + mc := &mockClient{ + responses: []*TopologyResponse{ + { + Switches: []map[string]SwitchAdjacency{ + {"core": {Switches: []string{"spine"}}}, + {"spine": {Switches: []string{"leaf"}}}, + {"leaf": {Nodes: []NodeInfo{ + {NodeID: "n1", AcceleratedNetworkID: "nvl1"}, + {NodeID: "n2", AcceleratedNetworkID: "nvl1"}, + }}}, + }, + }, + }, + } + + cis := []topology.ComputeInstances{{Instances: map[string]string{"n1": "node1", "n2": "node2"}}} + topo, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + + require.Nil(t, err) + require.Equal(t, 2, topo.Len()) + for _, inst := range topo.Instances { + require.Len(t, inst.FabricTiers, 3) + require.Equal(t, "nvl1", inst.XclrDomainID) + } +} + +func TestGenerateInstanceTopologyDeduplicatesAndSortsNodeIDs(t *testing.T) { + // "n2" appears in both ComputeInstances groups; only one copy should reach + // the API. The three unique IDs must arrive sorted so requests are + // deterministic regardless of map-iteration order. + mc := &mockClient{ + responses: []*TopologyResponse{{}}, + } + cis := []topology.ComputeInstances{ + {Instances: map[string]string{"n2": "node2a", "n3": "node3"}}, + {Instances: map[string]string{"n1": "node1", "n2": "node2b"}}, + } + _, err := newProvider(mc).generateInstanceTopology(context.Background(), nil, cis) + require.Nil(t, err) + + require.Len(t, mc.nodeIDsPerCall, 1, "exactly one API call expected") + // Fails if deduplication is removed (n2 appears twice) or sorting is + // removed (order becomes non-deterministic). + require.Equal(t, []string{"n1", "n2", "n3"}, mc.nodeIDsPerCall[0]) +} diff --git a/pkg/providers/dsx/provider.go b/pkg/providers/dsx/provider.go index a46db1b7..bf25a99f 100644 --- a/pkg/providers/dsx/provider.go +++ b/pkg/providers/dsx/provider.go @@ -1,14 +1,27 @@ /* - * Copyright 2026, NVIDIA CORPORATION - * SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. */ package dsx import ( "context" + "fmt" "net/http" + "github.com/mitchellh/mapstructure" "k8s.io/klog/v2" "github.com/NVIDIA/topograph/internal/httperr" @@ -16,7 +29,12 @@ import ( "github.com/NVIDIA/topograph/pkg/topology" ) -const NAME = "dsx" +const ( + NAME = "dsx" + + paramBaseURL = "base_url" + credToken = "token" +) type baseProvider struct { clientFactory ClientFactory @@ -29,27 +47,63 @@ type Client interface { GetTopology(ctx context.Context, vpcID string, nodeIDs []string, pageSize int, pageToken string) (*TopologyResponse, error) } +// TopologyResponse mirrors the models.TopologyResponse envelope from the DSX API spec. +// Switches is an ordered list of single-key entries (switch name → its adjacency). type TopologyResponse struct { - Switches map[string]SwitchInfo `json:"switches"` + Switches []map[string]SwitchAdjacency `json:"switches"` + NextPageToken string `json:"next_page_token,omitempty"` } -type SwitchInfo struct { +// SwitchAdjacency mirrors models.SwitchAdjacency from the DSX API spec. +type SwitchAdjacency struct { Switches []string `json:"switches,omitempty"` Nodes []NodeInfo `json:"nodes,omitempty"` } +// NodeInfo mirrors models.Node from the DSX API spec. type NodeInfo struct { NodeID string `json:"node_id"` AcceleratedNetworkID string `json:"accelerated_network_id,omitempty"` } +type paramsConfig struct { + BaseURL string `mapstructure:"base_url"` +} + func NamedLoader() (string, providers.Loader) { return NAME, Loader } func Loader(ctx context.Context, config providers.Config) (providers.Provider, *httperr.Error) { - // TODO: Implement real loader with authentication - return nil, httperr.NewError(http.StatusNotImplemented, "dsx provider not implemented") + var params paramsConfig + if err := mapstructure.Decode(config.Params, ¶ms); err != nil { + return nil, httperr.NewError(http.StatusBadRequest, "parameters error: "+err.Error()) + } + if params.BaseURL == "" { + return nil, httperr.NewError(http.StatusBadRequest, fmt.Sprintf("parameters error: missing '%s'", paramBaseURL)) + } + + trimTiers, err := providers.GetTrimTiers(config.Params) + if err != nil { + return nil, httperr.NewError(http.StatusBadRequest, "parameters error: "+err.Error()) + } + + var token string + if t, ok := config.Creds[credToken]; ok { + s, ok := t.(string) + if !ok { + return nil, httperr.NewError(http.StatusBadRequest, + fmt.Sprintf("credentials error: '%s' must be a string", credToken)) + } + token = s + } + + klog.InfoS("Loaded DSX provider", "base_url", params.BaseURL, "auth", map[bool]string{true: "bearer-token", false: "ambient-svid"}[token != ""]) + + factory := func() (Client, error) { + return NewHTTPClient(params.BaseURL, token), nil + } + return New(factory, trimTiers), nil } func (p *baseProvider) GenerateTopologyConfig(ctx context.Context, pageSize *int, instances []topology.ComputeInstances) (*topology.Graph, *httperr.Error) { diff --git a/pkg/providers/dsx/provider_sim.go b/pkg/providers/dsx/provider_sim.go index f68b8163..65e70cef 100644 --- a/pkg/providers/dsx/provider_sim.go +++ b/pkg/providers/dsx/provider_sim.go @@ -1,6 +1,17 @@ /* - * Copyright 2026, NVIDIA CORPORATION - * SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. */ package dsx @@ -34,44 +45,33 @@ func (client *simClient) GetTopology(ctx context.Context, _ string, nodeIDs []st return nil, providers.ErrAPIError } - // For simulation, generate topology from model - response := &TopologyResponse{ - Switches: make(map[string]SwitchInfo), - } - want := make(map[string]struct{}) for _, nodeID := range nodeIDs { want[nodeID] = struct{}{} } - //Iterate over the switches from the model and add them to the switch map + // Build the ordered list of single-key entries matching the API wire format. + var switches []map[string]SwitchAdjacency for _, sw := range client.model.Switches { - swInfo := SwitchInfo{ - Switches: make([]string, 0), - Nodes: make([]NodeInfo, 0), - } - - if len(sw.Nodes) > 0 { - //If it is a leaf switch, add the nodes to the switch info - for _, nodeName := range sw.Nodes { - if _, exists := want[nodeName]; !exists { - continue - } - node, exists := client.model.Nodes[nodeName] - if !exists { - continue - } - swInfo.Nodes = append(swInfo.Nodes, NodeInfo{NodeID: nodeName, AcceleratedNetworkID: node.AcceleratorDomain()}) + adj := SwitchAdjacency{} + for _, nodeName := range sw.Nodes { + if _, exists := want[nodeName]; !exists { + continue + } + node, exists := client.model.Nodes[nodeName] + if !exists { + continue } - } else { - //If it is not a leaf switch, add the child switches to the switch info - swInfo.Switches = append(swInfo.Switches, sw.Switches...) + adj.Nodes = append(adj.Nodes, NodeInfo{NodeID: nodeName, AcceleratedNetworkID: node.AcceleratorDomain()}) + } + adj.Switches = append(adj.Switches, sw.Switches...) + if len(adj.Nodes) == 0 && len(adj.Switches) == 0 { + continue } - response.Switches[sw.Name] = swInfo + switches = append(switches, map[string]SwitchAdjacency{sw.Name: adj}) } - //Return the response - return response, nil + return &TopologyResponse{Switches: switches}, nil } func NamedLoaderSim() (string, providers.Loader) { diff --git a/pkg/providers/dsx/provider_sim_test.go b/pkg/providers/dsx/provider_sim_test.go index d83eb770..f449517a 100644 --- a/pkg/providers/dsx/provider_sim_test.go +++ b/pkg/providers/dsx/provider_sim_test.go @@ -1,6 +1,17 @@ /* - * Copyright 2026, NVIDIA CORPORATION - * SPDX-License-Identifier: Apache-2.0 + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. */ package dsx @@ -368,14 +379,12 @@ func TestProviderSimWithNVLink(t *testing.T) { } func TestResponseToClusterTopologyOmitsEmptyAccelerator(t *testing.T) { - response := &TopologyResponse{Switches: map[string]SwitchInfo{ - "leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}, - }} - instances := []topology.ComputeInstances{{ - Instances: map[string]string{"n1": "node1"}, - }} - - topo := responseToClusterTopology(response, instances) + switches := []map[string]SwitchAdjacency{ + {"leaf": {Nodes: []NodeInfo{{NodeID: "n1"}}}}, + } + want := map[string]struct{}{"n1": {}} + + topo := buildClusterTopology(switches, want) require.Len(t, topo.Instances, 1) require.Empty(t, topo.Instances[0].XclrDomainID) diff --git a/pkg/providers/dsx/provider_test.go b/pkg/providers/dsx/provider_test.go new file mode 100644 index 00000000..587b4f88 --- /dev/null +++ b/pkg/providers/dsx/provider_test.go @@ -0,0 +1,91 @@ +/* + * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package dsx + +import ( + "context" + "net/http" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/topograph/pkg/providers" +) + +func TestLoader(t *testing.T) { + ctx := context.Background() + + tests := []struct { + name string + params map[string]any + creds map[string]any + wantErrCode int + wantErrMsg string + }{ + { + name: "missing base_url returns 400", + params: map[string]any{}, + creds: map[string]any{}, + wantErrCode: http.StatusBadRequest, + wantErrMsg: "missing 'base_url'", + }, + { + name: "non-string token returns 400", + params: map[string]any{"base_url": "https://topology.example.com"}, + creds: map[string]any{"token": 12345}, + wantErrCode: http.StatusBadRequest, + wantErrMsg: "credentials error: 'token' must be a string", + }, + { + name: "valid config without token succeeds", + params: map[string]any{"base_url": "https://topology.example.com"}, + creds: map[string]any{}, + }, + { + name: "valid config with string token succeeds", + params: map[string]any{"base_url": "https://topology.example.com"}, + creds: map[string]any{"token": "my-bearer-token"}, + }, + { + name: "trimTiers param is accepted", + params: map[string]any{"base_url": "https://topology.example.com", "trimTiers": 1}, + creds: map[string]any{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg := providers.Config{Params: tc.params, Creds: tc.creds} + p, err := Loader(ctx, cfg) + if tc.wantErrCode != 0 { + require.NotNil(t, err) + require.Equal(t, tc.wantErrCode, err.Code()) + require.Contains(t, err.Error(), tc.wantErrMsg) + require.Nil(t, p) + } else { + require.Nil(t, err) + require.NotNil(t, p) + } + }) + } +} + +func TestNamedLoader(t *testing.T) { + name, loader := NamedLoader() + require.Equal(t, NAME, name) + require.NotNil(t, loader) +} diff --git a/pkg/registry/registry.go b/pkg/registry/registry.go index d0ad0f18..53f16ad6 100644 --- a/pkg/registry/registry.go +++ b/pkg/registry/registry.go @@ -53,6 +53,7 @@ var Providers = providers.NewRegistry( netq.NamedLoader, lambdai.NamedLoader, lambdai.NamedLoaderSim, + dsx.NamedLoader, dsx.NamedLoaderSim, nscale.NamedLoader, nscale.NamedLoaderSim,