Fix K8s sidecar registration and stale cluster topology broadcast

apps/server/src/__tests__/metrics-orchestrator.test.ts

@@ -6,6 +6,8 @@ import {
   metricsServerMap,
   stopAllMetricsServers,
   reconcileClusterMetricsServers,
+  updateClusterNodeRegistry,
+  resolveClusterRefreshTarget,
   clients,
   clusterNodesRegistry,
   __test__,
@@ -229,12 +231,6 @@ describe("metrics-orchestrator", () => {
 
       // Mock all side-effectful internal functions
       mock.method(__test__, "createClient", async () => ({}))
-      mock.method(__test__, "getClusterTopology", async () => ({
-        clusterNodes: {
-          node1: { host: "127.0.0.1", port: "6379", tls: false, verifyTlsCertificate: false },
-        },
-        clusterId: "cluster-1",
-      }))
       mock.method(__test__, "updateClusterNodeRegistry", async () => mockClusterNodesRegistry)
       mock.method(__test__, "updateMetricsServers", async () => {})
       mock.method(__test__, "findDiff", async () => ({ nodesToAdd: {}, nodesToRemove: [] }))
@@ -271,4 +267,103 @@ describe("metrics-orchestrator", () => {
       assert.strictEqual(updateMetricsServers.mock.callCount(), 0)
     })
   })
+
+  describe("topology refresh", () => {
+    beforeEach(() => {
+      mock.restoreAll()
+      clients.clear()
+      metricsServerMap.clear()
+      clusterNodesRegistry.clear()
+    })
+    afterEach(() => {
+      mock.restoreAll()
+      clients.clear()
+      metricsServerMap.clear()
+      clusterNodesRegistry.clear()
+    })
+
+    it("updateClusterNodeRegistry replaces stale topology with the freshly discovered one", async () => {
+      // Seed a stale registry: cluster "node-1" knows only one node.
+      clusterNodesRegistry.set("node-1", {
+        "192-168-1-1-6379": { host: "192.168.1.1", port: 6379, tls: false, verifyTlsCertificate: false },
+      })
+
+      // A live client for that cluster whose CLUSTER SLOTS now reports three
+      // primaries. (SLOTS parsing itself is covered in connection.test.ts.)
+      const client = {
+        customCommand: async (args: string[]) =>
+          args[0] === "CLUSTER" && args[1] === "SLOTS"
+            ? [
+              [0, 5460, ["192.168.1.1", 6379, "node-1"]],
+              [5461, 10922, ["192.168.1.3", 6379, "node-2"]],
+              [10923, 16383, ["192.168.1.4", 6379, "node-3"]],
+            ]
+            : [],
+      } as never
+
+      const sampleNode = Object.values(clusterNodesRegistry.get("node-1") ?? {})[0]
+      await updateClusterNodeRegistry(client, sampleNode)
+
+      assert.deepStrictEqual(
+        Object.keys(clusterNodesRegistry.get("node-1") ?? {}).sort(),
+        ["192-168-1-1-6379", "192-168-1-3-6379", "192-168-1-4-6379"],
+        "registry should reflect the newly discovered topology, not the stale snapshot",
+      )
+    })
+
+    it("overwrites the existing entry when a known clusterId is passed, even if the derived id changed", async () => {
+      // Existing cluster tracked under "orig-id".
+      clusterNodesRegistry.set("orig-id", {
+        "192-168-1-1-6379": { host: "192.168.1.1", port: 6379, tls: false, verifyTlsCertificate: false },
+      })
+
+      // After a failover, CLUSTER SLOTS now lists a different first primary, so the
+      // derived clusterId would be "new-first-primary" — which would orphan "orig-id".
+      const client = {
+        customCommand: async (args: string[]) =>
+          args[0] === "CLUSTER" && args[1] === "SLOTS"
+            ? [[0, 16383, ["192.168.1.9", 6379, "new-first-primary"]]]
+            : [],
+      } as never
+
+      const sampleNode = Object.values(clusterNodesRegistry.get("orig-id") ?? {})[0]
+      await updateClusterNodeRegistry(client, sampleNode, "orig-id")
+
+      assert.deepStrictEqual(
+        [...clusterNodesRegistry.keys()],
+        ["orig-id"],
+        "the known clusterId should be overwritten in place, leaving no orphaned entry",
+      )
+      assert.deepStrictEqual(
+        Object.keys(clusterNodesRegistry.get("orig-id") ?? {}),
+        ["192-168-1-9-6379"],
+        "the entry should hold the freshly discovered node",
+      )
+    })
+
+    it("resolveClusterRefreshTarget uses the user client and carries node metadata forward", async () => {
+      const userClient = { id: "user-client" } as never
+      const clusterNodes = {
+        node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false, username: "admin", authType: "iam" as const },
+      }
+
+      const target = await resolveClusterRefreshTarget("cluster-1", clusterNodes, userClient)
+
+      assert.strictEqual(target?.client, userClient, "should refresh with the cluster's own live client")
+      // nodeInfo must be a node from THIS cluster (preserving its tls/username/auth),
+      // not initialConnectionDetails.
+      assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1, "should carry the cluster's own node metadata forward")
+    })
+
+    it("resolveClusterRefreshTarget skips a cluster with no live client when not preconfigured", async () => {
+      // DEPLOYMENT_MODE is unset in this file, so preConfiguredConnection is falsy
+      // and there is no initial client to fall back to.
+      const target = await resolveClusterRefreshTarget(
+        "cluster-1",
+        { node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false } },
+        undefined,
+      )
+      assert.strictEqual(target, undefined, "a cluster with no live client and no preconfigured fallback is skipped")
+    })
+  })
 })

apps/server/src/__tests__/orchestrator-k8s-register.test.ts

@@ -0,0 +1,114 @@
+import { describe, it, afterEach } from "node:test"
+import assert from "node:assert"
+import {
+  ORCHESTRATOR_AUTH_DOMAIN,
+  ORCHESTRATOR_AUTH_HEADER,
+  createOrchestratorAuthCredential
+} from "valkey-common"
+import type { Request, Response } from "express"
+
+// These tests exercise the Kubernetes-only branches of the orchestrator:
+//   - resolveCollectorKey falls back to the shared ORCHESTRATOR_KEY, and
+//   - handleRegister admits a sidecar whose nodeId is part of the discovered
+//     cluster topology even though the orchestrator never spawned it.
+// isKubernetes is a module-load constant, so DEPLOYMENT_MODE and the shared key
+// must be set BEFORE importing the module.
+process.env.DEPLOYMENT_MODE = "K8"
+process.env.ORCHESTRATOR_KEY = "k8s-shared-secret"
+
+const {
+  metricsServerMap,
+  clusterNodesRegistry,
+  resolveCollectorKey,
+  __test__,
+} = await import("../metrics-orchestrator")
+
+const NODE_ID = "valkey-6-valkey-headless-valkey-svc-cluster-local-6379"
+const URI = "http://10.42.0.32:3000"
+const SHARED_KEY = "k8s-shared-secret"
+
+const makeRes = () => {
+  const captured: { statusCode: number; body?: unknown } = { statusCode: 200 }
+  const res = {
+    status(code: number) { captured.statusCode = code; return res },
+    send(body?: unknown) { captured.body = body; return res },
+    sendStatus(code: number) { captured.statusCode = code; return res },
+  }
+  return { res: res as unknown as Response, captured }
+}
+
+const makeReq = (body: unknown, credential?: string) => ({
+  body,
+  headers: credential === undefined ? {} : { [ORCHESTRATOR_AUTH_HEADER]: credential },
+}) as unknown as Request
+
+const signRegister = (fields: { nodeId?: string; metricsServerUri?: string; timestamp?: number } = {}, key = SHARED_KEY) =>
+  createOrchestratorAuthCredential(key, ORCHESTRATOR_AUTH_DOMAIN.REGISTER, {
+    nodeId: NODE_ID,
+    metricsServerUri: URI,
+    timestamp: Date.now(),
+    ...fields,
+  }) as string
+
+// Register a node into the discovered topology so it is a known cluster member.
+const seedTopology = (nodeId = NODE_ID) => {
+  clusterNodesRegistry.set("cluster-1", {
+    [nodeId]: { host: "10.42.0.32", port: 6379, tls: false, verifyTlsCertificate: false },
+  })
+}
+
+describe("K8s shared-key registration", () => {
+  afterEach(() => {
+    metricsServerMap.clear()
+    clusterNodesRegistry.clear()
+    __test__.collectorKeys.clear()
+  })
+
+  describe("resolveCollectorKey", () => {
+    it("falls back to the shared ORCHESTRATOR_KEY for a node the orchestrator never spawned", () => {
+      assert.strictEqual(resolveCollectorKey("never-spawned-node"), SHARED_KEY)
+    })
+
+    it("still prefers a per-node minted key when one exists", () => {
+      __test__.collectorKeys.set("spawned-node", "per-node-key")
+      assert.strictEqual(resolveCollectorKey("spawned-node"), "per-node-key")
+    })
+  })
+
+  describe("handleRegister topology-membership gate", () => {
+    it("admits a correctly signed sidecar for a known cluster node with no pre-existing entry", () => {
+      seedTopology()
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false, "precondition: no orchestrator-spawned entry")
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister()), res)
+
+      assert.strictEqual(captured.statusCode, 200, "a known cluster node should be allowed to register")
+      assert.strictEqual(metricsServerMap.get(NODE_ID)?.metricsURI, URI, "the sidecar's entry should be created on first register")
+    })
+
+    it("rejects a correctly signed sidecar whose nodeId is not in the discovered topology", () => {
+      // Topology known, but for a different node.
+      seedTopology("some-other-node-6379")
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister()), res)
+
+      assert.strictEqual(captured.statusCode, 401, "an unknown node must be rejected even with a valid signature")
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false)
+    })
+
+    it("rejects a known cluster node when the signature uses the wrong key", () => {
+      seedTopology()
+
+      const { res, captured } = makeRes()
+      __test__.handleRegister(
+        makeReq({ nodeId: NODE_ID, metricsServerUri: URI, timestamp: Date.now() }, signRegister({}, "wrong-key")),
+        res,
+      )
+
+      assert.strictEqual(captured.statusCode, 401, "a bad signature must be rejected before the membership check")
+      assert.strictEqual(metricsServerMap.has(NODE_ID), false)
+    })
+  })
+})

apps/server/src/__tests__/topology-refresh-preconfigured.test.ts

@@ -0,0 +1,70 @@
+import { describe, it, afterEach, mock } from "node:test"
+import assert from "node:assert"
+
+// preConfiguredConnection is a module-load constant derived from VALKEY_HOST /
+// VALKEY_PORT, so it must be set BEFORE importing the module to exercise the
+// headless-refresh fallback (no live user client → initial client).
+process.env.VALKEY_HOST = "valkey-0.example"
+process.env.VALKEY_PORT = "6379"
+
+const {
+  resolveClusterRefreshTarget,
+  setPreconfiguredClusterId,
+  initialConnectionDetails,
+  __test__,
+} = await import("../metrics-orchestrator")
+
+const PRECONFIGURED_ID = "preconfigured-cluster"
+
+describe("resolveClusterRefreshTarget (preconfigured / headless)", () => {
+  afterEach(() => {
+    mock.restoreAll()
+    setPreconfiguredClusterId(undefined)
+  })
+
+  it("refreshes the preconfigured cluster via the initial client + initialConnectionDetails", async () => {
+    setPreconfiguredClusterId(PRECONFIGURED_ID)
+    const initialClient = { id: "initial-client" } as never
+    mock.method(__test__, "getInitialClient", async () => initialClient)
+
+    const target = await resolveClusterRefreshTarget(
+      PRECONFIGURED_ID,
+      { node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false } },
+      undefined,
+    )
+
+    assert.strictEqual(target?.client, initialClient, "the preconfigured cluster should refresh via the initial client")
+    assert.strictEqual(
+      target?.nodeInfo,
+      initialConnectionDetails,
+      "the preconfigured cluster decorates with initialConnectionDetails",
+    )
+  })
+
+  it("does NOT use the initial client for a different client-less cluster", async () => {
+    // Preconfigured cluster is A; refreshing a stale, client-less cluster B must
+    // not rediscover A's topology through the initial client.
+    setPreconfiguredClusterId(PRECONFIGURED_ID)
+    const getInitial = mock.method(__test__, "getInitialClient", async () => ({ id: "initial-client" } as never))
+
+    const target = await resolveClusterRefreshTarget(
+      "some-other-cluster",
+      { node1: { host: "10.9.9.9", port: 6379, tls: false, verifyTlsCertificate: false } },
+      undefined,
+    )
+
+    assert.strictEqual(target, undefined, "an unrelated client-less cluster must be left unchanged")
+    assert.strictEqual(getInitial.mock.callCount(), 0, "the initial client must not be consulted for other clusters")
+  })
+
+  it("still prefers a user client over the initial client when one is present", async () => {
+    const userClient = { id: "user-client" } as never
+    mock.method(__test__, "getInitialClient", async () => ({ id: "initial-client" } as never))
+
+    const clusterNodes = { node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false } }
+    const target = await resolveClusterRefreshTarget("any-cluster", clusterNodes, userClient)
+
+    assert.strictEqual(target?.client, userClient)
+    assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1)
+  })
+})

apps/server/src/connection.ts

@@ -22,7 +22,8 @@ import {
   reconcileClusterMetricsServers, 
   isKubernetes, 
   forgetCollectorKey,
-  ClusterNodeMap } from "./metrics-orchestrator"
+  ClusterNodeMap,
+  type NodeInfo } from "./metrics-orchestrator"
 import { subscribe } from "./node-watchers"
 import { clearCpuSamples } from "./node-utilization"
 import { createClusterValkeyClient, createStandaloneValkeyClient } from "./valkey-client"
@@ -595,7 +596,7 @@ function sendStandaloneConnectFulfilled(ws: WebSocket, payload: StandaloneConnec
 
 export async function discoverCluster(
   client: GlideClient | GlideClusterClient, 
-  payload: { connectionDetails: ConnectionDetails, connectionId?: string;},
+  payload: { connectionDetails: NodeInfo, connectionId?: string;},
 )  {
   try {
     // eslint-disable-next-line @typescript-eslint/no-explicit-any