From d4525f872e2b43220b5487b474eb79c41bb2d9a3 Mon Sep 17 00:00:00 2001 From: ravjotb Date: Wed, 9 Sep 2026 15:30:58 -0700 Subject: [PATCH 1/7] Fix K8s sidecar registration and stale cluster topology broadcast Two coupled defects broke the Kubernetes metrics path and left the Cluster Topology view stale: 1. Sidecar registration (regressed in #379): the register handler required a pre-existing metricsServerMap entry, but in K8s the orchestrator never spawns collectors, so externally-managed sidecars were always rejected (401/404). Registration is now allowed when the nodeId belongs to the discovered cluster topology, and the sidecar's entry is created on first register. Scoped to K8s so the strict no-entry gate is preserved for spawned (Web/Electron) collectors. Sidecars authenticate with a shared ORCHESTRATOR_KEY provisioned to both the orchestrator and every sidecar via a Kubernetes Secret; resolveCollectorKey falls back to that key only in K8s mode. 2. Stale topology broadcast (regressed in #279): refreshAllClusterRegistries broadcast the registry every 30s but no longer re-discovered topology, so clients kept the connect-time snapshot. It now re-discovers each tracked cluster through a live client belonging to that cluster before broadcasting. Adds tests covering registry refresh on topology change and that reconcile acts on the refreshed topology. Updates the K8s example manifests (Secret + env wiring) and deployment docs. Signed-off-by: ravjotb --- .../__tests__/metrics-orchestrator.test.ts | 72 +++++++++++++++++++ apps/server/src/index.ts | 21 +++++- apps/server/src/metrics-orchestrator.ts | 37 ++++++++-- .../src/content/docs/deployment/kubernetes.md | 26 ++++++- examples/k8s/app.yaml | 25 +++++++ .../k8s/valkey-statefulset-sidecar-patch.yaml | 5 ++ 6 files changed, 176 insertions(+), 10 deletions(-) diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts index a2b615ad..77f04932 100644 --- a/apps/server/src/__tests__/metrics-orchestrator.test.ts +++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts @@ -6,6 +6,7 @@ import { metricsServerMap, stopAllMetricsServers, reconcileClusterMetricsServers, + updateClusterNodeRegistry, clients, clusterNodesRegistry, __test__, @@ -271,4 +272,75 @@ describe("metrics-orchestrator", () => { assert.strictEqual(updateMetricsServers.mock.callCount(), 0) }) }) + + describe("topology refresh", () => { + afterEach(() => { + mock.restoreAll() + clusterNodesRegistry.clear() + metricsServerMap.clear() + }) + + it("updateClusterNodeRegistry replaces stale topology with the freshly discovered one", async () => { + // Seed a stale registry: cluster-1 knows only node1. + clusterNodesRegistry.set("cluster-1", { + node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, + }) + + // Discovery now returns an added node (node2) — i.e. the topology changed. + mock.method(__test__, "getClusterTopology", async () => ({ + clusterId: "cluster-1", + discoveredClusterNodes: { + node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, + node2: { host: "10.0.0.2", port: 6379, tls: false, verifyTlsCertificate: false }, + }, + })) + + await updateClusterNodeRegistry({} as never) + + assert.deepStrictEqual( + Object.keys(clusterNodesRegistry.get("cluster-1") ?? {}).sort(), + ["node1", "node2"], + "registry should reflect the newly discovered topology, not the stale snapshot", + ) + }) + + it("reconcile acts on refreshed topology: a node added by discovery is passed to updateMetricsServers", async () => { + mock.restoreAll() + clients.clear() + + // Start with a registry + metrics map in sync on node1 only. + clusterNodesRegistry.set("cluster-1", { + node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, + }) + metricsServerMap.set("node1", { metricsURI: "http://10.0.0.1:3000", pid: 123, lastSeen: Date.now() }) + + // Discovery reveals a new node2 → refresh the registry. + mock.method(__test__, "getClusterTopology", async () => ({ + clusterId: "cluster-1", + discoveredClusterNodes: { + node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, + node2: { host: "10.0.0.2", port: 6379, tls: false, verifyTlsCertificate: false }, + }, + })) + await updateClusterNodeRegistry({} as never) + + // findDiff derives adds/removes from whatever topology the registry now + // holds — so reconcile operates on the refreshed set, not the stale one. + mock.method(__test__, "findDiff", async (map: MetricsServerMap, nodes: ClusterNodeMap) => ({ + nodesToAdd: Object.fromEntries(Object.entries(nodes).filter(([id]) => !map.has(id))), + nodesToRemove: [] as string[], + })) + const updateMetricsServers = mock.method(__test__, "updateMetricsServers", async () => {}) + + await reconcileClusterMetricsServers(metricsServerMap) + + assert.strictEqual(updateMetricsServers.mock.callCount(), 1, "reconcile should act on the topology change") + const [nodesToAdd] = updateMetricsServers.mock.calls[0].arguments as [Record, string[], string] + assert.deepStrictEqual( + Object.keys(nodesToAdd), + ["node2"], + "the node added by the refreshed topology should be reconciled", + ) + }) + }) }) diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index b12006ae..f1b09842 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -3,6 +3,7 @@ import express from "express" import helmet from "helmet" import path from "path" import http from "http" +import { GlideClient, GlideClusterClient } from "@valkey/valkey-glide" import { VALKEY, CONNECTION_TEARDOWN_DELAY_MS, @@ -143,7 +144,23 @@ const wss = new WebSocketServer({ noServer: true }) const delay = (ms: number) => new Promise((res) => setTimeout(res, ms)) -function refreshAllClusterRegistries() { +async function refreshAllClusterRegistries() { + // Re-discover each cluster's topology before broadcasting, so clients receive + // the current cluster nodes rather than a snapshot taken once at connect/startup. + // Each cluster is refreshed with a live client that belongs to it; a cluster we + // hold no client for is left as-is (nothing to rediscover through). + const clientByCluster = new Map() + for (const { client, clusterId } of clients.values()) { + if (clusterId && !clientByCluster.has(clusterId)) clientByCluster.set(clusterId, client) + } + + await Promise.all( + [...clusterNodesRegistry.keys()] + .map((clusterId) => clientByCluster.get(clusterId)) + .filter((client): client is GlideClient | GlideClusterClient => client != null) + .map((client) => updateClusterNodeRegistry(client)), + ) + const connectionIdsByCluster = new Map() for (const [connectionId, entry] of clients) { if (!entry.clusterId) continue @@ -172,7 +189,7 @@ function refreshAllClusterRegistries() { async function refreshAllClusterRegistriesLoop() { while (true) { try { - refreshAllClusterRegistries() + await refreshAllClusterRegistries() } catch (err) { console.warn("Unable to refresh cluster topologies. ", err) } finally { diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts index 4d88a414..c1b2783a 100644 --- a/apps/server/src/metrics-orchestrator.ts +++ b/apps/server/src/metrics-orchestrator.ts @@ -79,7 +79,14 @@ const collectorKeys: Map = new Map() * replayed against another. */ export function resolveCollectorKey(nodeId: string): string | undefined { - return collectorKeys.get(nodeId) + const spawnedKey = collectorKeys.get(nodeId) + if (spawnedKey) return spawnedKey + // K8s collectors are external sidecars the orchestrator never spawns, so there + // is no per-node minted key. They authenticate with a shared key provisioned to + // both the orchestrator and every sidecar via a Kubernetes Secret. Scoped to K8s + // so the per-node property is preserved for spawned (Web/Electron) collectors. + if (isKubernetes) return process.env[ORCHESTRATOR_AUTH_KEY_ENV] + return undefined } /** @@ -225,6 +232,18 @@ function isPinnedMetricsHost(uri: string): boolean { } } +/** + * True when `nodeId` is part of a discovered cluster topology. In K8s the + * orchestrator only tracks topology (it never spawns collectors), so this is + * how a sidecar's registration is authorized as belonging to the cluster. + */ +function isKnownClusterNode(nodeId: string): boolean { + for (const clusterNodes of clusterNodesRegistry.values()) { + if (flattenClusterNodeMap(clusterNodes)[nodeId]) return true + } + return false +} + /** * `POST /orchestrator/register` — a collector advertising where it can be * reached. @@ -269,14 +288,22 @@ function handleRegister(req: Request, res: Response): void { return } - const entry = metricsServerMap.get(nodeId) - if (!entry) { - // Key material without an entry means the spawn did not complete. + // Allowed if we already have an entry (orchestrator-spawned) or, in K8s, + // the node is part of the discovered cluster (sidecars are external, so + // they create their entry on first register). + const allowed = metricsServerMap.has(nodeId) || (isKubernetes && isKnownClusterNode(nodeId)) + if (!allowed) { console.warn(`Rejected metrics registration for ${nodeId}: no metrics server entry`) res.status(401).send("Unauthorized") return } + let entry = metricsServerMap.get(nodeId) + if (!entry) { + entry = { metricsURI: "", pid: undefined, lastSeen: Date.now() } + metricsServerMap.set(nodeId, entry) + } + entry.metricsURI = metricsServerUri entry.lastSeen = Date.now() console.log(`Metrics server registered for ${nodeId} at ${metricsServerUri}`) @@ -369,7 +396,7 @@ async function getClusterTopology(client: GlideClusterClient | GlideClient | nul export async function updateClusterNodeRegistry(client: GlideClusterClient | GlideClient | null, connectionDetails = initialConnectionDetails) { try { - const { discoveredClusterNodes, clusterId } = await getClusterTopology(client, connectionDetails) + const { discoveredClusterNodes, clusterId } = await internals.getClusterTopology(client, connectionDetails) if (clusterId && discoveredClusterNodes) clusterNodesRegistry.set(clusterId, discoveredClusterNodes) } catch (err) { diff --git a/docs-site/src/content/docs/deployment/kubernetes.md b/docs-site/src/content/docs/deployment/kubernetes.md index 74444ca5..333a2a07 100644 --- a/docs-site/src/content/docs/deployment/kubernetes.md +++ b/docs-site/src/content/docs/deployment/kubernetes.md @@ -55,6 +55,20 @@ For a non-local deployment, use published images from a container registry and u ### 3. Deploy the app server +The metrics sidecars authenticate to the orchestrator's `/orchestrator/register` +and `/orchestrator/ping` endpoints with a shared key. Because sidecars are external +(the orchestrator does not spawn them in Kubernetes), the same key must be +provisioned to both the app server and every sidecar via a Secret. Create it first +with your own random value: + +```bash +kubectl create secret generic valkey-admin-orchestrator-key -n valkey \ + --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)" +``` + +`app.yaml` also ships a placeholder Secret for GitOps workflows — replace its value +rather than committing a real key. Then deploy the app server: + ```bash kubectl apply -f examples/k8s/app.yaml kubectl rollout status deployment/valkey-admin-app -n valkey @@ -181,15 +195,21 @@ kubectl logs -n valkey valkey-0 -c metrics You want to see `Register success` in the metrics sidecar log. -:::caution[Sidecar registration is not yet functional] -Sidecar registration does not currently succeed in `DEPLOYMENT_MODE=K8`. The orchestrator only tracks collectors it spawned itself, and it does not spawn sidecars, so a sidecar's registration is rejected and the sidecar exits after 30 attempts: +:::caution[Sidecar registration requires the shared key] +Registration and ping are authenticated. Each sidecar signs its requests with the +shared `ORCHESTRATOR_KEY`, and the orchestrator verifies against the same key, so +the Secret from step 3 must be present on **both** the app Deployment and the +sidecar. If the key is missing or does not match, the sidecar exits after 30 +attempts: ```text Register failed: 401 Unauthorized Failed to register with server after 30 attempts. Shutting down. ``` -Support is in progress. Until then the Kubernetes sidecar topology on this page will deploy but will not populate metrics panels. Use the [Docker deployment](/deployment/docker/) for a working metrics path. +If you see this, confirm the `valkey-admin-orchestrator-key` Secret exists in the +`valkey` namespace and that both pods reference it (`kubectl get pod ... -o yaml | +grep -A3 ORCHESTRATOR_KEY`). ::: ### Charts Empty in the UI diff --git a/examples/k8s/app.yaml b/examples/k8s/app.yaml index f06dc692..655e2b5d 100644 --- a/examples/k8s/app.yaml +++ b/examples/k8s/app.yaml @@ -1,3 +1,23 @@ +# Shared authentication key for the metrics control plane (/orchestrator/register, +# /orchestrator/ping). In K8s the metrics collectors are external sidecars, so the +# orchestrator cannot mint a per-node key at spawn time as it does for Web/Electron. +# A single shared key is provisioned to BOTH this Deployment and every metrics +# sidecar (see valkey-statefulset-sidecar-patch.yaml) and used to sign/verify +# register and ping requests. +# +# Generate your own key instead of committing one; e.g.: +# kubectl create secret generic valkey-admin-orchestrator-key -n valkey \ +# --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)" +apiVersion: v1 +kind: Secret +metadata: + name: valkey-admin-orchestrator-key + namespace: valkey +type: Opaque +stringData: + # openssl rand -hex 32 + ORCHESTRATOR_KEY: "replace-me-with-a-32-byte-random-hex-string" +--- apiVersion: apps/v1 kind: Deployment metadata: @@ -28,6 +48,11 @@ spec: value: valkey-0.valkey-headless.valkey.svc.cluster.local - name: VALKEY_PORT value: "6379" + - name: ORCHESTRATOR_KEY + valueFrom: + secretKeyRef: + name: valkey-admin-orchestrator-key + key: ORCHESTRATOR_KEY ports: - containerPort: 8080 name: http diff --git a/examples/k8s/valkey-statefulset-sidecar-patch.yaml b/examples/k8s/valkey-statefulset-sidecar-patch.yaml index ef574190..22104c95 100644 --- a/examples/k8s/valkey-statefulset-sidecar-patch.yaml +++ b/examples/k8s/valkey-statefulset-sidecar-patch.yaml @@ -34,6 +34,11 @@ spec: fieldPath: status.podIP - name: METRICS_ADVERTISE_PORT value: "3000" + - name: ORCHESTRATOR_KEY + valueFrom: + secretKeyRef: + name: valkey-admin-orchestrator-key + key: ORCHESTRATOR_KEY ports: - containerPort: 3000 name: metrics From fdc87ac24bab39c39e3f5d8114fa9634941b810a Mon Sep 17 00:00:00 2001 From: ravjotb Date: Thu, 10 Sep 2026 11:37:29 -0700 Subject: [PATCH 2/7] Reuse the per-cluster connection grouping in refreshAllClusterRegistries Build connectionIdsByCluster once and use it both to pick a client for topology re-discovery and to target the broadcast, instead of making a second pass over `clients` for a separate clientByCluster map. The client is reached via clients.get(connectionId).client, keyed by the same connectionIds already grouped for the broadcast. No behavior change. Signed-off-by: ravjotb --- apps/server/src/index.ts | 31 +++++++++++++++---------------- 1 file changed, 15 insertions(+), 16 deletions(-) diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index f1b09842..30860615 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -145,22 +145,8 @@ const wss = new WebSocketServer({ noServer: true }) const delay = (ms: number) => new Promise((res) => setTimeout(res, ms)) async function refreshAllClusterRegistries() { - // Re-discover each cluster's topology before broadcasting, so clients receive - // the current cluster nodes rather than a snapshot taken once at connect/startup. - // Each cluster is refreshed with a live client that belongs to it; a cluster we - // hold no client for is left as-is (nothing to rediscover through). - const clientByCluster = new Map() - for (const { client, clusterId } of clients.values()) { - if (clusterId && !clientByCluster.has(clusterId)) clientByCluster.set(clusterId, client) - } - - await Promise.all( - [...clusterNodesRegistry.keys()] - .map((clusterId) => clientByCluster.get(clusterId)) - .filter((client): client is GlideClient | GlideClusterClient => client != null) - .map((client) => updateClusterNodeRegistry(client)), - ) - + // Group live connections by cluster once; used both to pick a client to + // re-discover each cluster's topology and to target the broadcast below. const connectionIdsByCluster = new Map() for (const [connectionId, entry] of clients) { if (!entry.clusterId) continue @@ -169,6 +155,19 @@ async function refreshAllClusterRegistries() { connectionIdsByCluster.set(entry.clusterId, ids) } + // Re-discover each tracked cluster's topology before broadcasting, so clients + // receive the current cluster nodes rather than the connect-time snapshot. + // Each cluster is refreshed through one of its own live clients (topology + // discovery keys off that client's CLUSTER SLOTS); a cluster with no live + // client is left as-is, since there is nothing to rediscover through. + await Promise.all( + [...clusterNodesRegistry.keys()] + .map((clusterId) => connectionIdsByCluster.get(clusterId)?.[0]) + .map((connectionId) => (connectionId ? clients.get(connectionId)?.client : undefined)) + .filter((client): client is GlideClient | GlideClusterClient => client != null) + .map((client) => updateClusterNodeRegistry(client)), + ) + for (const [clusterId, clusterNodes] of clusterNodesRegistry) { const connectionIds = connectionIdsByCluster.get(clusterId) if (!connectionIds) continue From ef1acb0db2cc65503e8556e444f6455d24615125 Mon Sep 17 00:00:00 2001 From: ravjotb Date: Thu, 10 Sep 2026 11:41:50 -0700 Subject: [PATCH 3/7] Don't ship a placeholder orchestrator Secret in app.yaml MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit app.yaml defined a Secret named valkey-admin-orchestrator-key with a committed placeholder value. Applying it after creating a real random Secret of the same name would overwrite the real key with the public placeholder, leaving the orchestrator and every sidecar signing with a known key (CWE-798). An attacker able to reach the orchestrator could then register a known cluster node with an attacker-controlled metrics URI. Remove the Secret definition from app.yaml — the manifest now only references the Secret by name. The key is created out-of-band with a user-supplied random value (kubectl create secret ...), so apply can never clobber it. Docs updated to state the Secret is created first and is intentionally not defined in the manifest. Signed-off-by: ravjotb --- .../src/content/docs/deployment/kubernetes.md | 5 ++-- examples/k8s/app.yaml | 26 +++++++------------ 2 files changed, 12 insertions(+), 19 deletions(-) diff --git a/docs-site/src/content/docs/deployment/kubernetes.md b/docs-site/src/content/docs/deployment/kubernetes.md index 333a2a07..a728b119 100644 --- a/docs-site/src/content/docs/deployment/kubernetes.md +++ b/docs-site/src/content/docs/deployment/kubernetes.md @@ -66,8 +66,9 @@ kubectl create secret generic valkey-admin-orchestrator-key -n valkey \ --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)" ``` -`app.yaml` also ships a placeholder Secret for GitOps workflows — replace its value -rather than committing a real key. Then deploy the app server: +`app.yaml` intentionally does **not** define this Secret — it only references it — +so `kubectl apply` can never overwrite your generated key with a committed +placeholder. Create the Secret first (above), then deploy the app server: ```bash kubectl apply -f examples/k8s/app.yaml diff --git a/examples/k8s/app.yaml b/examples/k8s/app.yaml index 655e2b5d..a3718145 100644 --- a/examples/k8s/app.yaml +++ b/examples/k8s/app.yaml @@ -1,23 +1,15 @@ -# Shared authentication key for the metrics control plane (/orchestrator/register, -# /orchestrator/ping). In K8s the metrics collectors are external sidecars, so the -# orchestrator cannot mint a per-node key at spawn time as it does for Web/Electron. -# A single shared key is provisioned to BOTH this Deployment and every metrics -# sidecar (see valkey-statefulset-sidecar-patch.yaml) and used to sign/verify -# register and ping requests. +# NOTE: the metrics control plane (/orchestrator/register, /orchestrator/ping) +# authenticates sidecars with a shared key read from the ORCHESTRATOR_KEY env +# var below, sourced from the `valkey-admin-orchestrator-key` Secret. Create that +# Secret with your OWN random value before applying this manifest — it is +# intentionally NOT defined here so `kubectl apply` can never overwrite a real +# key with a committed placeholder: # -# Generate your own key instead of committing one; e.g.: # kubectl create secret generic valkey-admin-orchestrator-key -n valkey \ # --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)" -apiVersion: v1 -kind: Secret -metadata: - name: valkey-admin-orchestrator-key - namespace: valkey -type: Opaque -stringData: - # openssl rand -hex 32 - ORCHESTRATOR_KEY: "replace-me-with-a-32-byte-random-hex-string" ---- +# +# The same Secret is referenced by every metrics sidecar +# (see valkey-statefulset-sidecar-patch.yaml). apiVersion: apps/v1 kind: Deployment metadata: From 1b18dfba163e63e9ae87c8fae7319f401793da7d Mon Sep 17 00:00:00 2001 From: ravjotb Date: Wed, 16 Sep 2026 16:16:56 -0700 Subject: [PATCH 4/7] Address review feedback on topology refresh and K8s registration - Carry per-cluster connection metadata forward on refresh instead of reverting to initialConnectionDetails: updateClusterNodeRegistry now takes a required NodeInfo, and the refresh loop sources it from an existing node of that cluster. Narrow discoverCluster to NodeInfo and drop the redundant getClusterTopology wrapper. - Re-discover preconfigured clusters (K8s sidecars, headless Web) through the initial client each cycle, since they have no entry in `clients`; otherwise a node scaled in after boot never registers until a UI session opens. Await the boot-time K8s discovery. - Bound each cluster's re-discovery with a timeout so one hung CLUSTER SLOTS can't stall the whole broadcast loop. - examples/k8s/valkey-statefulset.yaml: give the inline metrics sidecar the ORCHESTRATOR_KEY it needs, and document that the Secret must exist and the cluster must be formed before the sidecars start. - Document that the shared orchestrator key is a cluster-scoped credential and that the K8s metrics host is not pinned to loopback. - Rework the topology-refresh test to exercise real discovery via a fake CLUSTER SLOTS client; drop a tautological reconcile test. Signed-off-by: ravjotb --- .../__tests__/metrics-orchestrator.test.ts | 89 ++++++------------- apps/server/src/connection.ts | 5 +- apps/server/src/index.ts | 38 +++++--- apps/server/src/metrics-orchestrator.ts | 17 +--- .../src/content/docs/deployment/kubernetes.md | 9 ++ examples/k8s/valkey-statefulset.yaml | 20 +++++ 6 files changed, 90 insertions(+), 88 deletions(-) diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts index 77f04932..056bdb6d 100644 --- a/apps/server/src/__tests__/metrics-orchestrator.test.ts +++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts @@ -230,12 +230,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: [] })) @@ -274,73 +268,46 @@ describe("metrics-orchestrator", () => { }) describe("topology refresh", () => { - afterEach(() => { + 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-1 knows only node1. - clusterNodesRegistry.set("cluster-1", { - node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, + // 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 }, }) - // Discovery now returns an added node (node2) — i.e. the topology changed. - mock.method(__test__, "getClusterTopology", async () => ({ - clusterId: "cluster-1", - discoveredClusterNodes: { - node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, - node2: { host: "10.0.0.2", port: 6379, tls: false, verifyTlsCertificate: false }, - }, - })) - - await updateClusterNodeRegistry({} as never) + // 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("cluster-1") ?? {}).sort(), - ["node1", "node2"], + 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("reconcile acts on refreshed topology: a node added by discovery is passed to updateMetricsServers", async () => { - mock.restoreAll() - clients.clear() - - // Start with a registry + metrics map in sync on node1 only. - clusterNodesRegistry.set("cluster-1", { - node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, - }) - metricsServerMap.set("node1", { metricsURI: "http://10.0.0.1:3000", pid: 123, lastSeen: Date.now() }) - - // Discovery reveals a new node2 → refresh the registry. - mock.method(__test__, "getClusterTopology", async () => ({ - clusterId: "cluster-1", - discoveredClusterNodes: { - node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false }, - node2: { host: "10.0.0.2", port: 6379, tls: false, verifyTlsCertificate: false }, - }, - })) - await updateClusterNodeRegistry({} as never) - - // findDiff derives adds/removes from whatever topology the registry now - // holds — so reconcile operates on the refreshed set, not the stale one. - mock.method(__test__, "findDiff", async (map: MetricsServerMap, nodes: ClusterNodeMap) => ({ - nodesToAdd: Object.fromEntries(Object.entries(nodes).filter(([id]) => !map.has(id))), - nodesToRemove: [] as string[], - })) - const updateMetricsServers = mock.method(__test__, "updateMetricsServers", async () => {}) - - await reconcileClusterMetricsServers(metricsServerMap) - - assert.strictEqual(updateMetricsServers.mock.callCount(), 1, "reconcile should act on the topology change") - const [nodesToAdd] = updateMetricsServers.mock.calls[0].arguments as [Record, string[], string] - assert.deepStrictEqual( - Object.keys(nodesToAdd), - ["node2"], - "the node added by the refreshed topology should be reconciled", - ) - }) }) }) diff --git a/apps/server/src/connection.ts b/apps/server/src/connection.ts index 191853e9..7587490d 100644 --- a/apps/server/src/connection.ts +++ b/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 diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 30860615..e41d1185 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -3,7 +3,6 @@ import express from "express" import helmet from "helmet" import path from "path" import http from "http" -import { GlideClient, GlideClusterClient } from "@valkey/valkey-glide" import { VALKEY, CONNECTION_TEARDOWN_DELAY_MS, @@ -144,6 +143,11 @@ const wss = new WebSocketServer({ noServer: true }) const delay = (ms: number) => new Promise((res) => setTimeout(res, ms)) +// Upper bound on a single cluster's topology re-discovery. Without it, one hung +// CLUSTER SLOTS would leave the awaited refresh unsettled and stall the broadcast +// for every cluster, not just the unresponsive one. +const TOPOLOGY_REDISCOVERY_TIMEOUT_MS = 10_000 + async function refreshAllClusterRegistries() { // Group live connections by cluster once; used both to pick a client to // re-discover each cluster's topology and to target the broadcast below. @@ -155,17 +159,27 @@ async function refreshAllClusterRegistries() { connectionIdsByCluster.set(entry.clusterId, ids) } - // Re-discover each tracked cluster's topology before broadcasting, so clients - // receive the current cluster nodes rather than the connect-time snapshot. - // Each cluster is refreshed through one of its own live clients (topology - // discovery keys off that client's CLUSTER SLOTS); a cluster with no live - // client is left as-is, since there is nothing to rediscover through. + // Re-discover each tracked cluster before broadcasting. User-connected clusters + // use their own client + node metadata; a preconfigured cluster (K8s / headless + // Web) has no entry in `clients`, so it falls back to the initial client so + // scaled-in nodes are picked up without a UI session. Time-bounded so one hung + // node can't stall the whole loop. await Promise.all( - [...clusterNodesRegistry.keys()] - .map((clusterId) => connectionIdsByCluster.get(clusterId)?.[0]) - .map((connectionId) => (connectionId ? clients.get(connectionId)?.client : undefined)) - .filter((client): client is GlideClient | GlideClusterClient => client != null) - .map((client) => updateClusterNodeRegistry(client)), + [...clusterNodesRegistry.entries()].map(async ([clusterId, clusterNodes]) => { + const connectionId = connectionIdsByCluster.get(clusterId)?.[0] + const userClient = connectionId ? clients.get(connectionId)?.client : undefined + + const client = userClient ?? (preConfiguredConnection ? await getInitialClient() : undefined) + const nodeInfo = userClient ? Object.values(clusterNodes)[0] : initialConnectionDetails + if (!client || !nodeInfo) return + + await Promise.race([ + updateClusterNodeRegistry(client, nodeInfo), + delay(TOPOLOGY_REDISCOVERY_TIMEOUT_MS).then(() => + console.warn(`Topology re-discovery for cluster ${clusterId} timed out; broadcasting last known nodes.`), + ), + ]) + }), ) for (const [clusterId, clusterNodes] of clusterNodesRegistry) { @@ -200,7 +214,7 @@ async function refreshAllClusterRegistriesLoop() { async function updateRegistryforK8() { const client = await getInitialClient() - updateClusterNodeRegistry(client, initialConnectionDetails) + await updateClusterNodeRegistry(client, initialConnectionDetails) } // Electron: bind to localhost only — Origin headers are forgeable by non-browser clients, diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts index c1b2783a..48dbf222 100644 --- a/apps/server/src/metrics-orchestrator.ts +++ b/apps/server/src/metrics-orchestrator.ts @@ -43,7 +43,7 @@ export type MetricsServerMap = Map -type NodeInfo = { +export type NodeInfo = { host: string; port: number | string; username?: string; @@ -386,17 +386,9 @@ async function createClient(connectionDetails: ConnectionDetails) { return await createOrchestratorValkeyClient({ addresses, credentials, useTLS: tls, verifyTlsCertificate, databaseId: db }) } -async function getClusterTopology(client: GlideClusterClient | GlideClient | null, node: ConnectionDetails) { - if (!client) client = await createClient(node) - - const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: node }) - - return { discoveredClusterNodes, clusterId } -} - -export async function updateClusterNodeRegistry(client: GlideClusterClient | GlideClient | null, connectionDetails = initialConnectionDetails) { +export async function updateClusterNodeRegistry(client: GlideClusterClient | GlideClient, nodeInfo: NodeInfo) { try { - const { discoveredClusterNodes, clusterId } = await internals.getClusterTopology(client, connectionDetails) + const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: nodeInfo }) if (clusterId && discoveredClusterNodes) clusterNodesRegistry.set(clusterId, discoveredClusterNodes) } catch (err) { @@ -609,7 +601,7 @@ export async function startPreconfiguredMetricsServers() { const client = await getInitialClient() if (await belongsToCluster(client)) { if (isWebMode) { - const { discoveredClusterNodes, clusterId } = await internals.getClusterTopology(client, initialConnectionDetails) + const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: initialConnectionDetails }) if (clusterId && discoveredClusterNodes) { clusterNodesRegistry.set(clusterId, discoveredClusterNodes) if (!clusterCredentials.has(clusterId)) clusterCredentials.set(clusterId, initialConnectionDetails.password) @@ -643,7 +635,6 @@ export function cleanupOrchestratorResources() { const internals = { startMetricsServers, createClient, - getClusterTopology, updateClusterNodeRegistry, findDiff, flattenClusterNodeMap, diff --git a/docs-site/src/content/docs/deployment/kubernetes.md b/docs-site/src/content/docs/deployment/kubernetes.md index a728b119..4e503e90 100644 --- a/docs-site/src/content/docs/deployment/kubernetes.md +++ b/docs-site/src/content/docs/deployment/kubernetes.md @@ -211,6 +211,15 @@ Failed to register with server after 30 attempts. Shutting down. If you see this, confirm the `valkey-admin-orchestrator-key` Secret exists in the `valkey` namespace and that both pods reference it (`kubectl get pod ... -o yaml | grep -A3 ORCHESTRATOR_KEY`). + +The key is shared cluster-wide rather than per-node, so it authenticates a sidecar +as *a* member of the cluster, not as one specific node. Treat it as a +cluster-scoped credential: restrict read access on the Secret, and rotate it by +recreating the Secret and restarting the app Deployment and the sidecars. Note that +in `DEPLOYMENT_MODE=K8` the registered metrics host is not pinned to loopback, so +any holder of the key can point a node's metrics URI at an arbitrary host that the +server will then fetch from — keep the key tightly scoped and the namespace's +network egress constrained accordingly. ::: ### Charts Empty in the UI diff --git a/examples/k8s/valkey-statefulset.yaml b/examples/k8s/valkey-statefulset.yaml index 5aaf2e66..cc5442fb 100644 --- a/examples/k8s/valkey-statefulset.yaml +++ b/examples/k8s/valkey-statefulset.yaml @@ -1,3 +1,18 @@ +# Example Valkey cluster StatefulSet with an inline metrics sidecar. +# +# PREREQUISITES — the metrics sidecars will crash-loop (and, because StatefulSet +# rollout is ordered, block the whole rollout at valkey-0) unless both are met: +# +# 1. Create the shared orchestrator key Secret first (the sidecar exits with +# "Missing ORCHESTRATOR_KEY ... Shutting down" without it): +# kubectl create secret generic valkey-admin-orchestrator-key -n valkey \ +# --from-literal=ORCHESTRATOR_KEY="$(openssl rand -hex 32)" +# +# 2. Form the cluster after the pods are up. These nodes start with +# --cluster-enabled yes but do NOT self-form, so CLUSTER SLOTS is empty and +# the sidecar dies on "Cluster(No topology views found)". Assign slots once: +# valkey-cli --cluster create :6379 --cluster-replicas +# apiVersion: v1 kind: Service metadata: @@ -105,6 +120,11 @@ spec: fieldPath: status.podIP - name: METRICS_ADVERTISE_PORT value: "3000" + - name: ORCHESTRATOR_KEY + valueFrom: + secretKeyRef: + name: valkey-admin-orchestrator-key + key: ORCHESTRATOR_KEY ports: - containerPort: 3000 name: metrics From 4a434b5495753dcdb5c081a8f0df0f6928d5b0d9 Mon Sep 17 00:00:00 2001 From: ravjotb Date: Wed, 16 Sep 2026 16:23:37 -0700 Subject: [PATCH 5/7] Overwrite the known cluster entry on refresh instead of orphaning it updateClusterNodeRegistry keyed the registry on the clusterId derived from CLUSTER SLOTS (the first slot range's primary). That id changes on failover/resharding, so re-discovering an existing cluster could write a new entry under a different id and leave the old one orphaned in the map. Add an optional clusterId override: the refresh loop passes the cluster's existing id so the entry is overwritten in place, while boot-time first discovery still falls back to the derived id. Adds a test covering the changed-derived-id case. Signed-off-by: ravjotb --- .../__tests__/metrics-orchestrator.test.ts | 30 +++++++++++++++++++ apps/server/src/index.ts | 2 +- apps/server/src/metrics-orchestrator.ts | 14 +++++++-- 3 files changed, 42 insertions(+), 4 deletions(-) diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts index 056bdb6d..b60664d4 100644 --- a/apps/server/src/__tests__/metrics-orchestrator.test.ts +++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts @@ -309,5 +309,35 @@ describe("metrics-orchestrator", () => { "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", + ) + }) }) }) diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index e41d1185..077d1d7d 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -174,7 +174,7 @@ async function refreshAllClusterRegistries() { if (!client || !nodeInfo) return await Promise.race([ - updateClusterNodeRegistry(client, nodeInfo), + updateClusterNodeRegistry(client, nodeInfo, clusterId), delay(TOPOLOGY_REDISCOVERY_TIMEOUT_MS).then(() => console.warn(`Topology re-discovery for cluster ${clusterId} timed out; broadcasting last known nodes.`), ), diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts index 48dbf222..c11c43da 100644 --- a/apps/server/src/metrics-orchestrator.ts +++ b/apps/server/src/metrics-orchestrator.ts @@ -386,10 +386,18 @@ async function createClient(connectionDetails: ConnectionDetails) { return await createOrchestratorValkeyClient({ addresses, credentials, useTLS: tls, verifyTlsCertificate, databaseId: db }) } -export async function updateClusterNodeRegistry(client: GlideClusterClient | GlideClient, nodeInfo: NodeInfo) { +export async function updateClusterNodeRegistry( + client: GlideClusterClient | GlideClient, + nodeInfo: NodeInfo, + clusterId?: string, +) { try { - const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: nodeInfo }) - if (clusterId && discoveredClusterNodes) clusterNodesRegistry.set(clusterId, discoveredClusterNodes) + const { discoveredClusterNodes, clusterId: discoveredClusterId } = await discoverCluster(client, { connectionDetails: nodeInfo }) + // Prefer the caller's known clusterId when refreshing an existing cluster: the + // discovered id is derived from the first primary in CLUSTER SLOTS, which can + // change on failover/resharding and would otherwise orphan the old entry. + const key = clusterId ?? discoveredClusterId + if (key && discoveredClusterNodes) clusterNodesRegistry.set(key, discoveredClusterNodes) } catch (err) { if (err instanceof ConnectionError) { From 3679ca5a9f70d2a4fe36ba6a0854e614f33a531a Mon Sep 17 00:00:00 2001 From: ravjotb Date: Wed, 16 Sep 2026 17:00:31 -0700 Subject: [PATCH 6/7] Add coverage for the K8s register and topology-refresh fixes The branch's new behavior was largely untested, and the K8s-gated paths are unreachable from the default test suite because isKubernetes is a module-load constant. - orchestrator-k8s-register.test.ts sets DEPLOYMENT_MODE=K8 before import to cover the shared-key path: resolveCollectorKey falls back to ORCHESTRATOR_KEY (and still prefers a minted per-node key), and handleRegister admits a signed sidecar for a known cluster node with no pre-existing entry while rejecting unknown nodes and wrong-key signatures. - Extract resolveClusterRefreshTarget (the refresh client/nodeInfo choice) so it is unit-testable: a user-connected cluster carries its own node metadata forward; a preconfigured/headless cluster falls back to the initial client + initialConnectionDetails. Covered in metrics-orchestrator (user path / skip) and topology-refresh-preconfigured (fallback path). getInitialClient is routed via internals so the fallback is mockable; no behavior change. Signed-off-by: ravjotb --- .../__tests__/metrics-orchestrator.test.ts | 25 ++++ .../orchestrator-k8s-register.test.ts | 114 ++++++++++++++++++ .../topology-refresh-preconfigured.test.ts | 43 +++++++ apps/server/src/index.ts | 10 +- apps/server/src/metrics-orchestrator.ts | 19 +++ 5 files changed, 206 insertions(+), 5 deletions(-) create mode 100644 apps/server/src/__tests__/orchestrator-k8s-register.test.ts create mode 100644 apps/server/src/__tests__/topology-refresh-preconfigured.test.ts diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts index b60664d4..f07fad38 100644 --- a/apps/server/src/__tests__/metrics-orchestrator.test.ts +++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts @@ -7,6 +7,7 @@ import { stopAllMetricsServers, reconcileClusterMetricsServers, updateClusterNodeRegistry, + resolveClusterRefreshTarget, clients, clusterNodesRegistry, __test__, @@ -339,5 +340,29 @@ describe("metrics-orchestrator", () => { "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(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( + { 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") + }) }) }) diff --git a/apps/server/src/__tests__/orchestrator-k8s-register.test.ts b/apps/server/src/__tests__/orchestrator-k8s-register.test.ts new file mode 100644 index 00000000..3512fabf --- /dev/null +++ b/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) + }) + }) +}) diff --git a/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts new file mode 100644 index 00000000..ece9c0df --- /dev/null +++ b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts @@ -0,0 +1,43 @@ +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, initialConnectionDetails, __test__ } = await import("../metrics-orchestrator") + +describe("resolveClusterRefreshTarget (preconfigured / headless)", () => { + afterEach(() => mock.restoreAll()) + + it("falls back to the initial client and initialConnectionDetails when there is no user client", async () => { + const initialClient = { id: "initial-client" } as never + mock.method(__test__, "getInitialClient", async () => initialClient) + + const target = await resolveClusterRefreshTarget( + { node1: { host: "10.0.0.1", port: 6379, tls: false, verifyTlsCertificate: false } }, + undefined, + ) + + assert.strictEqual(target?.client, initialClient, "headless refresh should use the orchestrator's initial client") + assert.strictEqual( + target?.nodeInfo, + initialConnectionDetails, + "headless refresh decorates with initialConnectionDetails, the authoritative preconfigured config", + ) + }) + + it("still prefers a user client over the initial client when one is present", async () => { + const userClient = { id: "user-client" } as never + const initialClient = { id: "initial-client" } as never + mock.method(__test__, "getInitialClient", async () => initialClient) + + const clusterNodes = { node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false } } + const target = await resolveClusterRefreshTarget(clusterNodes, userClient) + + assert.strictEqual(target?.client, userClient) + assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1) + }) +}) diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 077d1d7d..91120598 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -47,7 +47,8 @@ import { isKubernetes, preConfiguredConnection, getInitialClient, - updateClusterNodeRegistry + updateClusterNodeRegistry, + resolveClusterRefreshTarget } from "./metrics-orchestrator" import { isAllowedWebSocketOrigin } from "./websocket-origin" import { ensureSession, hasAuthorizedSession, isConnectionAuthorized, setSessionExpiryListener } from "./session" @@ -169,12 +170,11 @@ async function refreshAllClusterRegistries() { const connectionId = connectionIdsByCluster.get(clusterId)?.[0] const userClient = connectionId ? clients.get(connectionId)?.client : undefined - const client = userClient ?? (preConfiguredConnection ? await getInitialClient() : undefined) - const nodeInfo = userClient ? Object.values(clusterNodes)[0] : initialConnectionDetails - if (!client || !nodeInfo) return + const target = await resolveClusterRefreshTarget(clusterNodes, userClient) + if (!target) return await Promise.race([ - updateClusterNodeRegistry(client, nodeInfo, clusterId), + updateClusterNodeRegistry(target.client, target.nodeInfo, clusterId), delay(TOPOLOGY_REDISCOVERY_TIMEOUT_MS).then(() => console.warn(`Topology re-discovery for cluster ${clusterId} timed out; broadcasting last known nodes.`), ), diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts index c11c43da..f456ae99 100644 --- a/apps/server/src/metrics-orchestrator.ts +++ b/apps/server/src/metrics-orchestrator.ts @@ -408,6 +408,24 @@ export async function updateClusterNodeRegistry( return clusterNodesRegistry } +/** + * Decide how to refresh one tracked cluster. A user-connected cluster is refreshed + * with its own client and metadata carried forward from an existing node (so a + * refresh doesn't revert TLS/auth to server defaults). A preconfigured cluster + * (K8s / headless Web) has no live user client, so it falls back to the initial + * client with initialConnectionDetails. Returns undefined when the cluster can't + * be refreshed (no client available). + */ +export async function resolveClusterRefreshTarget( + clusterNodes: ClusterNodeMap, + userClient: GlideClusterClient | GlideClient | undefined, +): Promise<{ client: GlideClusterClient | GlideClient; nodeInfo: NodeInfo } | undefined> { + const client = userClient ?? (preConfiguredConnection ? await internals.getInitialClient() : undefined) + const nodeInfo = userClient ? Object.values(clusterNodes)[0] : initialConnectionDetails + if (!client || !nodeInfo) return undefined + return { client, nodeInfo } +} + async function findDiff(metricsServerMap: MetricsServerMap, clusterNodeMap: ClusterNodeMap) { const clusterNodes = isKubernetes ? flattenClusterNodeMap(clusterNodeMap) : clusterNodeMap // These are nodes that are in the clusterMap but not metricsMap @@ -643,6 +661,7 @@ export function cleanupOrchestratorResources() { const internals = { startMetricsServers, createClient, + getInitialClient, updateClusterNodeRegistry, findDiff, flattenClusterNodeMap, From 694681a705a4de38a0cc1a3dfc738cc062cab7dd Mon Sep 17 00:00:00 2001 From: ravjotb Date: Thu, 17 Sep 2026 10:56:37 -0700 Subject: [PATCH 7/7] Use the initial client only for the preconfigured cluster on refresh resolveClusterRefreshTarget fell back to the initial client for any client-less cluster when preConfiguredConnection was set. In a mixed deployment, an inactive cluster could then be rediscovered through a different preconfigured cluster's client, storing that topology under the inactive cluster's id. Track the preconfigured clusterId (recorded at boot from the Web and K8s discovery paths; updateClusterNodeRegistry now returns the written id) and only use the initial client when the cluster being refreshed matches it. Other client-less clusters are left unchanged. A user-connected cluster still refreshes through its own client, so the preconfigured cluster is covered by that path when a session is open. Signed-off-by: ravjotb --- .../__tests__/metrics-orchestrator.test.ts | 3 +- .../topology-refresh-preconfigured.test.ts | 43 +++++++++++++++---- apps/server/src/index.ts | 8 ++-- apps/server/src/metrics-orchestrator.ts | 25 ++++++++--- 4 files changed, 61 insertions(+), 18 deletions(-) diff --git a/apps/server/src/__tests__/metrics-orchestrator.test.ts b/apps/server/src/__tests__/metrics-orchestrator.test.ts index f07fad38..3cc6e4a0 100644 --- a/apps/server/src/__tests__/metrics-orchestrator.test.ts +++ b/apps/server/src/__tests__/metrics-orchestrator.test.ts @@ -347,7 +347,7 @@ describe("metrics-orchestrator", () => { node1: { host: "10.0.0.1", port: 6379, tls: true, verifyTlsCertificate: false, username: "admin", authType: "iam" as const }, } - const target = await resolveClusterRefreshTarget(clusterNodes, userClient) + 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), @@ -359,6 +359,7 @@ describe("metrics-orchestrator", () => { // 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, ) diff --git a/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts index ece9c0df..83ee5fa0 100644 --- a/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts +++ b/apps/server/src/__tests__/topology-refresh-preconfigured.test.ts @@ -7,35 +7,62 @@ import assert from "node:assert" process.env.VALKEY_HOST = "valkey-0.example" process.env.VALKEY_PORT = "6379" -const { resolveClusterRefreshTarget, initialConnectionDetails, __test__ } = await import("../metrics-orchestrator") +const { + resolveClusterRefreshTarget, + setPreconfiguredClusterId, + initialConnectionDetails, + __test__, +} = await import("../metrics-orchestrator") + +const PRECONFIGURED_ID = "preconfigured-cluster" describe("resolveClusterRefreshTarget (preconfigured / headless)", () => { - afterEach(() => mock.restoreAll()) + afterEach(() => { + mock.restoreAll() + setPreconfiguredClusterId(undefined) + }) - it("falls back to the initial client and initialConnectionDetails when there is no user client", async () => { + 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, "headless refresh should use the orchestrator's initial client") + assert.strictEqual(target?.client, initialClient, "the preconfigured cluster should refresh via the initial client") assert.strictEqual( target?.nodeInfo, initialConnectionDetails, - "headless refresh decorates with initialConnectionDetails, the authoritative preconfigured config", + "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 - const initialClient = { id: "initial-client" } as never - mock.method(__test__, "getInitialClient", async () => initialClient) + 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(clusterNodes, userClient) + const target = await resolveClusterRefreshTarget("any-cluster", clusterNodes, userClient) assert.strictEqual(target?.client, userClient) assert.deepStrictEqual(target?.nodeInfo, clusterNodes.node1) diff --git a/apps/server/src/index.ts b/apps/server/src/index.ts index 91120598..149cdb9f 100644 --- a/apps/server/src/index.ts +++ b/apps/server/src/index.ts @@ -48,7 +48,8 @@ import { preConfiguredConnection, getInitialClient, updateClusterNodeRegistry, - resolveClusterRefreshTarget + resolveClusterRefreshTarget, + setPreconfiguredClusterId } from "./metrics-orchestrator" import { isAllowedWebSocketOrigin } from "./websocket-origin" import { ensureSession, hasAuthorizedSession, isConnectionAuthorized, setSessionExpiryListener } from "./session" @@ -170,7 +171,7 @@ async function refreshAllClusterRegistries() { const connectionId = connectionIdsByCluster.get(clusterId)?.[0] const userClient = connectionId ? clients.get(connectionId)?.client : undefined - const target = await resolveClusterRefreshTarget(clusterNodes, userClient) + const target = await resolveClusterRefreshTarget(clusterId, clusterNodes, userClient) if (!target) return await Promise.race([ @@ -214,7 +215,8 @@ async function refreshAllClusterRegistriesLoop() { async function updateRegistryforK8() { const client = await getInitialClient() - await updateClusterNodeRegistry(client, initialConnectionDetails) + const clusterId = await updateClusterNodeRegistry(client, initialConnectionDetails) + setPreconfiguredClusterId(clusterId) } // Electron: bind to localhost only — Origin headers are forgeable by non-browser clients, diff --git a/apps/server/src/metrics-orchestrator.ts b/apps/server/src/metrics-orchestrator.ts index f456ae99..b71792cb 100644 --- a/apps/server/src/metrics-orchestrator.ts +++ b/apps/server/src/metrics-orchestrator.ts @@ -64,6 +64,11 @@ export const clusterNodesRegistry: Map = new Map() export const clusterCredentials: Map = new Map() +let preconfiguredClusterId: string | undefined +export function setPreconfiguredClusterId(clusterId: string | undefined) { + preconfiguredClusterId = clusterId +} + export const metricsServerMap: MetricsServerMap = new Map() /** @@ -397,7 +402,10 @@ export async function updateClusterNodeRegistry( // discovered id is derived from the first primary in CLUSTER SLOTS, which can // change on failover/resharding and would otherwise orphan the old entry. const key = clusterId ?? discoveredClusterId - if (key && discoveredClusterNodes) clusterNodesRegistry.set(key, discoveredClusterNodes) + if (key && discoveredClusterNodes) { + clusterNodesRegistry.set(key, discoveredClusterNodes) + return key + } } catch (err) { if (err instanceof ConnectionError) { @@ -405,7 +413,7 @@ export async function updateClusterNodeRegistry( } console.error(err) } - return clusterNodesRegistry + return undefined } /** @@ -417,13 +425,17 @@ export async function updateClusterNodeRegistry( * be refreshed (no client available). */ export async function resolveClusterRefreshTarget( + clusterId: string, clusterNodes: ClusterNodeMap, userClient: GlideClusterClient | GlideClient | undefined, ): Promise<{ client: GlideClusterClient | GlideClient; nodeInfo: NodeInfo } | undefined> { - const client = userClient ?? (preConfiguredConnection ? await internals.getInitialClient() : undefined) - const nodeInfo = userClient ? Object.values(clusterNodes)[0] : initialConnectionDetails - if (!client || !nodeInfo) return undefined - return { client, nodeInfo } + if (userClient) return { client: userClient, nodeInfo: Object.values(clusterNodes)[0] } + + // Only the preconfigured cluster may be refreshed via the initial client. + if (preConfiguredConnection && clusterId === preconfiguredClusterId) { + return { client: await internals.getInitialClient(), nodeInfo: initialConnectionDetails } + } + return undefined } async function findDiff(metricsServerMap: MetricsServerMap, clusterNodeMap: ClusterNodeMap) { @@ -630,6 +642,7 @@ export async function startPreconfiguredMetricsServers() { const { discoveredClusterNodes, clusterId } = await discoverCluster(client, { connectionDetails: initialConnectionDetails }) if (clusterId && discoveredClusterNodes) { clusterNodesRegistry.set(clusterId, discoveredClusterNodes) + setPreconfiguredClusterId(clusterId) if (!clusterCredentials.has(clusterId)) clusterCredentials.set(clusterId, initialConnectionDetails.password) } runReconcileLoop()