You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by wu...@apache.org on 2026/06/21 13:58:09 UTC
(skywalking-horizon-ui) branch main updated: perf(bff): parallelize topology node/edge MQE fan-out across the 4 map routes (#64)
This is an automated email from the ASF dual-hosted git repository.
wu-sheng pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-horizon-ui.git
The following commit(s) were added to refs/heads/main by this push:
new 82a0bfa perf(bff): parallelize topology node/edge MQE fan-out across the 4 map routes (#64)
82a0bfa is described below
commit 82a0bfab9d79c52677a12ca327eb58819b3c3989
Author: 吴晟 Wu Sheng <wu...@foxmail.com>
AuthorDate: Sun Jun 21 21:58:04 2026 +0800
perf(bff): parallelize topology node/edge MQE fan-out across the 4 map routes (#64)
The service-map, instance-map, deployment, and endpoint-dependency routes each loaded per-node then per-edge metrics via serial chunked loops. They now build both fragment families up front and fan them out concurrently via a shared fetchAliasedChunks helper (bounded-concurrency mapPool), cutting the metric-load wall-clock toward a single round-trip for typical graphs.
Output is identical on the success path — same fragments, same merge, same values. On a partial OAP failure the new per-chunk soft-fail degrades only the failed chunk instead of aborting the rest (break-on-first-error before), so strictly more metrics are retained.
- util/mapPool.ts: bounded-concurrency ordered map (+ unit test)
- client/graphql.ts: fetchAliasedChunks<T> (chunk + pool + per-chunk soft-fail + merge)
- topology / instance-topology / deployment / endpoint-dependency: node+edge fragments built up front, then Promise.all(fetchAliasedChunks node, edge)
---
apps/bff/src/client/graphql.ts | 38 +++++++
apps/bff/src/http/query/deployment.ts | 124 +++++++++++------------
apps/bff/src/http/query/endpoint-dependency.ts | 92 ++++++++---------
apps/bff/src/http/query/instance-topology.ts | 114 ++++++++++-----------
apps/bff/src/http/query/topology.ts | 132 +++++++++++--------------
apps/bff/src/util/mapPool.test.ts | 84 ++++++++++++++++
apps/bff/src/util/mapPool.ts | 46 +++++++++
7 files changed, 370 insertions(+), 260 deletions(-)
diff --git a/apps/bff/src/client/graphql.ts b/apps/bff/src/client/graphql.ts
index d48d959..a9c98b3 100644
--- a/apps/bff/src/client/graphql.ts
+++ b/apps/bff/src/client/graphql.ts
@@ -17,6 +17,7 @@
import type { FetchLike } from '@skywalking-horizon-ui/api-client';
import type { HorizonConfig } from '../config/schema.js';
+import { mapPool } from '../util/mapPool.js';
export interface GraphqlOptions {
queryUrl: string;
@@ -132,3 +133,40 @@ export async function graphqlPost<T>(
}
return body.data;
}
+
+/**
+ * Fan an aliased-fragment metric query out to OAP in bounded-concurrency
+ * chunks, merging every chunk's aliases into one object. `chunkSize` caps
+ * fragments per request (OAP enforces a per-request complexity ceiling);
+ * `concurrency` caps simultaneous chunk requests so a wide topology doesn't
+ * stampede OAP. Each chunk soft-fails independently — a failed chunk simply
+ * contributes no aliases, so one transient error degrades that slice to
+ * null metrics instead of aborting the whole fan-out (the topology routes
+ * keep the graph and emit nulls). Returns `{}` when there are no fragments.
+ */
+export async function fetchAliasedChunks<T>(
+ opts: GraphqlOptions,
+ fragments: readonly string[],
+ chunkSize: number,
+ queryName: string,
+ concurrency = 4,
+): Promise<Record<string, T>> {
+ if (fragments.length === 0) return {};
+ const chunks: string[][] = [];
+ for (let i = 0; i < fragments.length; i += chunkSize) {
+ chunks.push(fragments.slice(i, i + chunkSize));
+ }
+ const envs = await mapPool(chunks, concurrency, async (slice) => {
+ const query = `query ${queryName} {\n ${slice.join('\n ')}\n}`;
+ try {
+ return await graphqlPost<Record<string, T>>(opts, query);
+ } catch {
+ return null;
+ }
+ });
+ const merged: Record<string, T> = {};
+ for (const env of envs) {
+ if (env) Object.assign(merged, env);
+ }
+ return merged;
+}
diff --git a/apps/bff/src/http/query/deployment.ts b/apps/bff/src/http/query/deployment.ts
index f2fc6e1..73427f3 100644
--- a/apps/bff/src/http/query/deployment.ts
+++ b/apps/bff/src/http/query/deployment.ts
@@ -53,7 +53,7 @@ import type {
UITemplateClient,
} from '@skywalking-horizon-ui/api-client';
import { requireAuth } from '../../user/middleware.js';
-import { graphqlPost, buildOapOpts } from '../../client/graphql.js';
+import { graphqlPost, buildOapOpts, fetchAliasedChunks } from '../../client/graphql.js';
import { withColdStage } from '../../util/duration.js';
import {
defaultMinuteWindow,
@@ -412,46 +412,29 @@ export function registerDeploymentRoute(
return rc?.nodeMetrics ?? cfgNN.nodeMetrics ?? [];
}
- // ── Per-node MQE (each node uses its role's defs).
+ // ── Per-node + per-edge MQE. Build both fragment families (each node uses
+ // its role's metric defs; each edge its role-pair defs), then fan them out
+ // concurrently — disjoint OAP entities + disjoint result maps. Each family
+ // chunks internally and soft-fails per chunk, keeping the graph on a hiccup.
const nodeMetricVals = new Map<string, Record<string, number | null>>();
- const realNodes = nodes.filter((n) => n.isReal);
- {
- const fragments: string[] = [];
- const aliasMap = new Map<string, { nodeId: string; metric: DeploymentMetricDef }>();
- realNodes.forEach((n, i) => {
- defsFor(n).forEach((m, j) => {
- const alias = `n${i}_${j}`;
- aliasMap.set(alias, { nodeId: n.id, metric: m });
- fragments.push(nodeFragment(alias, m, n.serviceName, n.name, serviceNormal, window, coldStage));
- });
- });
- const CHUNK = 150;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query DeploymentNodeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break; // soft-fail: keep the graph with null node metrics
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const rec = nodeMetricVals.get(info.nodeId) ?? {};
- rec[info.metric.id] = v;
- nodeMetricVals.set(info.nodeId, rec);
- }
- }
- }
-
- // ── Per-edge MQE (server + client families, per-side gate). Self-loop
- // edges (source === target) are allowed — a node may call itself.
const serverMetricVals = new Map<string, Record<string, number | null>>();
const clientMetricVals = new Map<string, Record<string, number | null>>();
const serverMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
const clientMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+
+ const realNodes = nodes.filter((n) => n.isReal);
+ const nodeAliasMap = new Map<string, { nodeId: string; metric: DeploymentMetricDef }>();
+ const nodeFragments: string[] = [];
+ realNodes.forEach((n, i) => {
+ defsFor(n).forEach((m, j) => {
+ const alias = `n${i}_${j}`;
+ nodeAliasMap.set(alias, { nodeId: n.id, metric: m });
+ nodeFragments.push(nodeFragment(alias, m, n.serviceName, n.name, serviceNormal, window, coldStage));
+ });
+ });
+
+ // Per-edge: server + client families, per-side gate. Self-loop edges
+ // (source === target) are allowed — a node may call itself.
const linkSrv = cfg.linkServerMetrics ?? [];
const linkCli = cfg.linkClientMetrics ?? [];
const roleToRole = cfg.roleToRole ?? [];
@@ -490,12 +473,12 @@ export function registerDeploymentRoute(
const b = nodeById.get(c.target);
return !!a && !!b && !!a.name && !!b.name;
});
+ const edgeAliasMap = new Map<
+ string,
+ { callId: string; metric: DeploymentMetricDef; side: 'server' | 'client' }
+ >();
+ const edgeFragments: string[] = [];
if (candidateEdges.length > 0 && (linkSrv.length > 0 || linkCli.length > 0 || roleToRole.length > 0)) {
- const fragments: string[] = [];
- const aliasMap = new Map<
- string,
- { callId: string; metric: DeploymentMetricDef; side: 'server' | 'client' }
- >();
candidateEdges.forEach((c, i) => {
const src = nodeById.get(c.source)!;
const dst = nodeById.get(c.target)!;
@@ -503,8 +486,8 @@ export function registerDeploymentRoute(
if (dst.isReal) {
server.forEach((m, j) => {
const alias = `s${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
- fragments.push(
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
+ edgeFragments.push(
relationFragment(alias, m, entityServiceName, src.name, dst.name, serviceNormal, window, coldStage),
);
});
@@ -512,37 +495,40 @@ export function registerDeploymentRoute(
if (src.isReal) {
client.forEach((m, j) => {
const alias = `c${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
- fragments.push(
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
+ edgeFragments.push(
relationFragment(alias, m, entityServiceName, src.name, dst.name, serviceNormal, window, coldStage),
);
});
}
});
- const CHUNK = 200;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query DeploymentEdgeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const valBucket = info.side === 'server' ? serverMetricVals : clientMetricVals;
- const seriesBucket = info.side === 'server' ? serverMetricSeries : clientMetricSeries;
- const valRec = valBucket.get(info.callId) ?? {};
- valRec[info.metric.id] = v;
- valBucket.set(info.callId, valRec);
- const sRec = seriesBucket.get(info.callId) ?? {};
- sRec[info.metric.id] = seriesFromMqe(shape);
- seriesBucket.set(info.callId, sRec);
- }
- }
+ }
+
+ const [nodeEnv, edgeEnv] = await Promise.all([
+ fetchAliasedChunks<MqeShape>(opts, nodeFragments, 150, 'DeploymentNodeMetrics'),
+ fetchAliasedChunks<MqeShape>(opts, edgeFragments, 200, 'DeploymentEdgeMetrics'),
+ ]);
+
+ for (const [alias, shape] of Object.entries(nodeEnv)) {
+ const info = nodeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const rec = nodeMetricVals.get(info.nodeId) ?? {};
+ rec[info.metric.id] = v;
+ nodeMetricVals.set(info.nodeId, rec);
+ }
+ for (const [alias, shape] of Object.entries(edgeEnv)) {
+ const info = edgeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const valBucket = info.side === 'server' ? serverMetricVals : clientMetricVals;
+ const seriesBucket = info.side === 'server' ? serverMetricSeries : clientMetricSeries;
+ const valRec = valBucket.get(info.callId) ?? {};
+ valRec[info.metric.id] = v;
+ valBucket.set(info.callId, valRec);
+ const sRec = seriesBucket.get(info.callId) ?? {};
+ sRec[info.metric.id] = seriesFromMqe(shape);
+ seriesBucket.set(info.callId, sRec);
}
// ── Build response. Show EVERY container — both the ones in the call
diff --git a/apps/bff/src/http/query/endpoint-dependency.ts b/apps/bff/src/http/query/endpoint-dependency.ts
index 67bb0e1..38c5b26 100644
--- a/apps/bff/src/http/query/endpoint-dependency.ts
+++ b/apps/bff/src/http/query/endpoint-dependency.ts
@@ -41,7 +41,7 @@ import type {
UITemplateClient,
} from '@skywalking-horizon-ui/api-client';
import { requireAuth } from '../../user/middleware.js';
-import { graphqlPost, buildOapOpts } from '../../client/graphql.js';
+import { graphqlPost, buildOapOpts, fetchAliasedChunks } from '../../client/graphql.js';
import { withColdStage } from '../../util/duration.js';
import {
defaultMinuteWindow,
@@ -395,38 +395,25 @@ export function registerEndpointDependencyRoute(
const realNodes = graph.nodes.filter(
(n) => n.isReal && n.serviceName && n.name && n.name !== 'User',
);
+ // ── Per-node + per-edge MQE. Build both fragment families, then fan them
+ // out concurrently (disjoint OAP entities + result maps); each chunks
+ // internally and soft-fails per chunk, keeping the graph on a hiccup.
const nodeMetricVals = new Map<string, Record<string, number | null>>();
+ const edgeMetricVals = new Map<string, Record<string, number | null>>();
+ const edgeMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+
+ const nodeAliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
+ const nodeFragments: string[] = [];
if (realNodes.length > 0 && epCfg.nodeMetrics.length > 0) {
- const fragments: string[] = [];
- const aliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
realNodes.forEach((n, i) => {
const isFocus = n.serviceId === serviceId;
const useNormal = isFocus ? normal : true;
epCfg.nodeMetrics.forEach((m, j) => {
const alias = `e${i}_${j}`;
- aliasMap.set(alias, { nodeId: n.id, metric: m });
- fragments.push(endpointFragment(alias, m, n.serviceName, n.name, useNormal, window, coldStage));
+ nodeAliasMap.set(alias, { nodeId: n.id, metric: m });
+ nodeFragments.push(endpointFragment(alias, m, n.serviceName, n.name, useNormal, window, coldStage));
});
});
- const CHUNK = 150;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query EndpointMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const rec = nodeMetricVals.get(info.nodeId) ?? {};
- rec[info.metric.id] = v;
- nodeMetricVals.set(info.nodeId, rec);
- }
- }
}
// ── Per-edge MQE under EndpointRelation. We also capture the
@@ -441,17 +428,15 @@ export function registerEndpointDependencyRoute(
// virtual we use a synthetic source service name and
// `sourceNormal: false`, which is what booster does too.
const linkMetrics = epCfg.linkMetrics ?? [];
- const edgeMetricVals = new Map<string, Record<string, number | null>>();
- const edgeMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
const realEndpointMap = new Map(realNodes.map((n) => [n.id, n]));
const nodeById = new Map(graph.nodes.map((n) => [n.id, n]));
const candidateEdges = graph.calls.filter((c) => {
const dst = nodeById.get(c.target);
return !!dst && dst.isReal && !!dst.name && !!dst.serviceName;
});
+ const edgeAliasMap = new Map<string, { callId: string; metric: TopologyMetricDef }>();
+ const edgeFragments: string[] = [];
if (candidateEdges.length > 0 && linkMetrics.length > 0) {
- const fragments: string[] = [];
- const aliasMap = new Map<string, { callId: string; metric: TopologyMetricDef }>();
candidateEdges.forEach((c, i) => {
const dst = realEndpointMap.get(c.target) ?? nodeById.get(c.target)!;
const src = nodeById.get(c.source);
@@ -465,8 +450,8 @@ export function registerEndpointDependencyRoute(
const dstNormal = dst.serviceId === serviceId ? normal : true;
linkMetrics.forEach((m, j) => {
const alias = `r${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m });
- fragments.push(
+ edgeAliasMap.set(alias, { callId: c.id, metric: m });
+ edgeFragments.push(
endpointRelationFragment(
alias,
m,
@@ -482,28 +467,31 @@ export function registerEndpointDependencyRoute(
);
});
});
- const CHUNK = 200;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query EndpointEdgeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const rec = edgeMetricVals.get(info.callId) ?? {};
- rec[info.metric.id] = v;
- edgeMetricVals.set(info.callId, rec);
- const sRec = edgeMetricSeries.get(info.callId) ?? {};
- sRec[info.metric.id] = seriesFromMqe(shape);
- edgeMetricSeries.set(info.callId, sRec);
- }
- }
+ }
+
+ const [nodeEnv, edgeEnv] = await Promise.all([
+ fetchAliasedChunks<MqeShape>(opts, nodeFragments, 150, 'EndpointMetrics'),
+ fetchAliasedChunks<MqeShape>(opts, edgeFragments, 200, 'EndpointEdgeMetrics'),
+ ]);
+
+ for (const [alias, shape] of Object.entries(nodeEnv)) {
+ const info = nodeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const rec = nodeMetricVals.get(info.nodeId) ?? {};
+ rec[info.metric.id] = v;
+ nodeMetricVals.set(info.nodeId, rec);
+ }
+ for (const [alias, shape] of Object.entries(edgeEnv)) {
+ const info = edgeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const rec = edgeMetricVals.get(info.callId) ?? {};
+ rec[info.metric.id] = v;
+ edgeMetricVals.set(info.callId, rec);
+ const sRec = edgeMetricSeries.get(info.callId) ?? {};
+ sRec[info.metric.id] = seriesFromMqe(shape);
+ edgeMetricSeries.set(info.callId, sRec);
}
// ── Build response — drop nodes without any metric values, then
diff --git a/apps/bff/src/http/query/instance-topology.ts b/apps/bff/src/http/query/instance-topology.ts
index a5b0136..af21e23 100644
--- a/apps/bff/src/http/query/instance-topology.ts
+++ b/apps/bff/src/http/query/instance-topology.ts
@@ -50,7 +50,7 @@ import type {
UITemplateClient,
} from '@skywalking-horizon-ui/api-client';
import { requireAuth } from '../../user/middleware.js';
-import { graphqlPost, buildOapOpts } from '../../client/graphql.js';
+import { graphqlPost, buildOapOpts, fetchAliasedChunks } from '../../client/graphql.js';
import { withColdStage } from '../../util/duration.js';
import {
defaultMinuteWindow,
@@ -320,45 +320,30 @@ export function registerInstanceTopologyRoute(
return knownServices.get(n.serviceId)?.normal ?? n.isReal;
}
- // ── Per-node MQE.
+ // ── Per-node + per-edge MQE. Build both fragment families, then fan them
+ // out concurrently (disjoint OAP entities + disjoint result maps); each
+ // chunks internally and soft-fails per chunk, keeping the graph on a hiccup.
const nodeMetricVals = new Map<string, Record<string, number | null>>();
+ const serverMetricVals = new Map<string, Record<string, number | null>>();
+ const clientMetricVals = new Map<string, Record<string, number | null>>();
+ const serverMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+ const clientMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+
const realNodes = nodes.filter((n) => n.isReal);
+ const nodeAliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
+ const nodeFragments: string[] = [];
if (realNodes.length > 0 && instCfg.nodeMetrics.length > 0) {
- const fragments: string[] = [];
- const aliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
realNodes.forEach((n, i) => {
instCfg.nodeMetrics.forEach((m, j) => {
const alias = `n${i}_${j}`;
- aliasMap.set(alias, { nodeId: n.id, metric: m });
- fragments.push(nodeFragment(alias, m, n.serviceName, n.name, normalFor(n), window, coldStage));
+ nodeAliasMap.set(alias, { nodeId: n.id, metric: m });
+ nodeFragments.push(nodeFragment(alias, m, n.serviceName, n.name, normalFor(n), window, coldStage));
});
});
- const CHUNK = 150;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query InstanceNodeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break; // soft-fail: keep the graph with null node metrics
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const rec = nodeMetricVals.get(info.nodeId) ?? {};
- rec[info.metric.id] = v;
- nodeMetricVals.set(info.nodeId, rec);
- }
- }
}
- // ── Per-edge MQE (server + client families, per-side gate).
- const serverMetricVals = new Map<string, Record<string, number | null>>();
- const clientMetricVals = new Map<string, Record<string, number | null>>();
- const serverMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
- const clientMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+ // Per-edge: server + client families, per-side gate (server needs a real
+ // DEST, client a real SOURCE).
const linkSrv = instCfg.linkServerMetrics ?? [];
const linkCli = instCfg.linkClientMetrics ?? [];
const candidateEdges = calls.filter((c) => {
@@ -366,12 +351,12 @@ export function registerInstanceTopologyRoute(
const b = nodeById.get(c.target);
return !!a && !!b && !!a.name && !!b.name;
});
+ const edgeAliasMap = new Map<
+ string,
+ { callId: string; metric: TopologyMetricDef; side: 'server' | 'client' }
+ >();
+ const edgeFragments: string[] = [];
if (candidateEdges.length > 0 && (linkSrv.length > 0 || linkCli.length > 0)) {
- const fragments: string[] = [];
- const aliasMap = new Map<
- string,
- { callId: string; metric: TopologyMetricDef; side: 'server' | 'client' }
- >();
candidateEdges.forEach((c, i) => {
const src = nodeById.get(c.source)!;
const dst = nodeById.get(c.target)!;
@@ -380,8 +365,8 @@ export function registerInstanceTopologyRoute(
if (dst.isReal) {
linkSrv.forEach((m, j) => {
const alias = `s${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
- fragments.push(
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
+ edgeFragments.push(
relationFragment(alias, m, src.serviceName, src.name, srcNormal, dst.serviceName, dst.name, dstNormal, window, coldStage),
);
});
@@ -389,37 +374,40 @@ export function registerInstanceTopologyRoute(
if (src.isReal) {
linkCli.forEach((m, j) => {
const alias = `c${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
- fragments.push(
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
+ edgeFragments.push(
relationFragment(alias, m, src.serviceName, src.name, srcNormal, dst.serviceName, dst.name, dstNormal, window, coldStage),
);
});
}
});
- const CHUNK = 200;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query InstanceEdgeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const valBucket = info.side === 'server' ? serverMetricVals : clientMetricVals;
- const seriesBucket = info.side === 'server' ? serverMetricSeries : clientMetricSeries;
- const valRec = valBucket.get(info.callId) ?? {};
- valRec[info.metric.id] = v;
- valBucket.set(info.callId, valRec);
- const sRec = seriesBucket.get(info.callId) ?? {};
- sRec[info.metric.id] = seriesFromMqe(shape);
- seriesBucket.set(info.callId, sRec);
- }
- }
+ }
+
+ const [nodeEnv, edgeEnv] = await Promise.all([
+ fetchAliasedChunks<MqeShape>(opts, nodeFragments, 150, 'InstanceNodeMetrics'),
+ fetchAliasedChunks<MqeShape>(opts, edgeFragments, 200, 'InstanceEdgeMetrics'),
+ ]);
+
+ for (const [alias, shape] of Object.entries(nodeEnv)) {
+ const info = nodeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const rec = nodeMetricVals.get(info.nodeId) ?? {};
+ rec[info.metric.id] = v;
+ nodeMetricVals.set(info.nodeId, rec);
+ }
+ for (const [alias, shape] of Object.entries(edgeEnv)) {
+ const info = edgeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const valBucket = info.side === 'server' ? serverMetricVals : clientMetricVals;
+ const seriesBucket = info.side === 'server' ? serverMetricSeries : clientMetricSeries;
+ const valRec = valBucket.get(info.callId) ?? {};
+ valRec[info.metric.id] = v;
+ valBucket.set(info.callId, valRec);
+ const sRec = seriesBucket.get(info.callId) ?? {};
+ sRec[info.metric.id] = seriesFromMqe(shape);
+ seriesBucket.set(info.callId, sRec);
}
// ── Build response. Connected instances only — an instance with no
diff --git a/apps/bff/src/http/query/topology.ts b/apps/bff/src/http/query/topology.ts
index 37f7dfe..977f4b3 100644
--- a/apps/bff/src/http/query/topology.ts
+++ b/apps/bff/src/http/query/topology.ts
@@ -46,7 +46,7 @@ import type {
UITemplateClient,
} from '@skywalking-horizon-ui/api-client';
import { requireAuth } from '../../user/middleware.js';
-import { graphqlPost, buildOapOpts } from '../../client/graphql.js';
+import { graphqlPost, buildOapOpts, fetchAliasedChunks } from '../../client/graphql.js';
import { withColdStage } from '../../util/duration.js';
import {
defaultMinuteWindow,
@@ -428,51 +428,34 @@ export function registerTopologyRoute(app: FastifyInstance, deps: TopologyRouteD
// skipped since OAP has no metrics for them.
const realNodes = [...nodes.values()].filter((n) => n.isReal);
const nodeMetricVals = new Map<string, Record<string, number | null>>();
+ const serverMetricVals = new Map<string, Record<string, number | null>>();
+ const clientMetricVals = new Map<string, Record<string, number | null>>();
+ // Per-edge time series for the right-sidebar line charts.
+ const serverMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+ const clientMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
+
+ // Build the per-node and per-edge MQE fragments, then fan BOTH families
+ // out concurrently: they query disjoint OAP entities (node metrics vs
+ // service-relation metrics) and fill disjoint result maps, so there's no
+ // reason to await one before the other. Each family chunks internally and
+ // soft-fails per chunk, keeping the graph with null metrics on a hiccup.
+ const nodeAliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
+ const nodeFragments: string[] = [];
if (realNodes.length > 0 && topoCfg.nodeMetrics.length > 0) {
- const fragments: string[] = [];
- const aliasMap = new Map<string, { nodeId: string; metric: TopologyMetricDef }>();
realNodes.forEach((n, i) => {
const meta = knownServices.get(n.id);
const normal = meta?.normal ?? true;
topoCfg.nodeMetrics.forEach((m, j) => {
const alias = `n${i}_${j}`;
- aliasMap.set(alias, { nodeId: n.id, metric: m });
- fragments.push(nodeFragment(alias, m, n.name, normal, window, coldStage));
+ nodeAliasMap.set(alias, { nodeId: n.id, metric: m });
+ nodeFragments.push(nodeFragment(alias, m, n.name, normal, window, coldStage));
});
});
- const CHUNK = 150;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query NodeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- // Soft-fail per operator direction: keep the graph and
- // emit null metrics rather than wiping the response.
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const rec = nodeMetricVals.get(info.nodeId) ?? {};
- rec[info.metric.id] = v;
- nodeMetricVals.set(info.nodeId, rec);
- }
- }
}
// ── Per-edge MQE. Only real → real edges have a relation entity
- // in OAP. We split into server / client families and fan out in a
- // single combined aliased query.
- const serverMetricVals = new Map<string, Record<string, number | null>>();
- const clientMetricVals = new Map<string, Record<string, number | null>>();
- // Per-edge time series for the right-sidebar line charts.
- const serverMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
- const clientMetricSeries = new Map<string, Record<string, Array<number | null> | null>>();
- // Per booster: an edge with at least one named endpoint is
- // queryable. We split per side:
+ // in OAP. We split into server / client families and fan both out
+ // concurrently in chunked aliased queries (see the Promise.all below).
// - server metrics need a real DEST (the callee that records
// the incoming call) — `User → consumer` works on the
// server side, `provider → localhost:-1` does not.
@@ -487,12 +470,12 @@ export function registerTopologyRoute(app: FastifyInstance, deps: TopologyRouteD
});
const linkSrv = topoCfg.linkServerMetrics ?? [];
const linkCli = topoCfg.linkClientMetrics ?? [];
+ const edgeAliasMap = new Map<
+ string,
+ { callId: string; metric: TopologyMetricDef; side: 'server' | 'client' }
+ >();
+ const edgeFragments: string[] = [];
if (candidateEdges.length > 0 && (linkSrv.length > 0 || linkCli.length > 0)) {
- const fragments: string[] = [];
- const aliasMap = new Map<
- string,
- { callId: string; metric: TopologyMetricDef; side: 'server' | 'client' }
- >();
candidateEdges.forEach((c, i) => {
const src = nodes.get(c.source)!;
const dst = nodes.get(c.target)!;
@@ -503,53 +486,50 @@ export function registerTopologyRoute(app: FastifyInstance, deps: TopologyRouteD
// back to the graph node's `isReal`.
const srcNormal = srcMeta?.normal ?? src.isReal;
const dstNormal = dstMeta?.normal ?? dst.isReal;
- // Server metrics live on the DEST — fetch only when the dest
- // is real. Otherwise OAP returns nothing for the server
- // family and we'd waste the round trip.
+ // Server metrics live on the DEST — fetch only when the dest is real.
if (dst.isReal) {
linkSrv.forEach((m, j) => {
const alias = `s${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
- fragments.push(relationFragment(alias, m, src.name, srcNormal, dst.name, dstNormal, window, coldStage));
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'server' });
+ edgeFragments.push(relationFragment(alias, m, src.name, srcNormal, dst.name, dstNormal, window, coldStage));
});
}
- // Client metrics live on the SOURCE — fetch only when the
- // source is real.
+ // Client metrics live on the SOURCE — fetch only when the source is real.
if (src.isReal) {
linkCli.forEach((m, j) => {
const alias = `c${i}_${j}`;
- aliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
- fragments.push(relationFragment(alias, m, src.name, srcNormal, dst.name, dstNormal, window, coldStage));
+ edgeAliasMap.set(alias, { callId: c.id, metric: m, side: 'client' });
+ edgeFragments.push(relationFragment(alias, m, src.name, srcNormal, dst.name, dstNormal, window, coldStage));
});
}
});
- const CHUNK = 200;
- for (let i = 0; i < fragments.length; i += CHUNK) {
- const slice = fragments.slice(i, i + CHUNK);
- const query = `query EdgeMetrics {\n ${slice.join('\n ')}\n}`;
- let env: Record<string, MqeShape>;
- try {
- env = await graphqlPost<Record<string, MqeShape>>(opts, query);
- } catch {
- // Soft-fail: keep the graph with null edge metrics.
- break;
- }
- for (const [alias, shape] of Object.entries(env)) {
- const info = aliasMap.get(alias);
- if (!info) continue;
- const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
- const seriesBucket =
- info.side === 'server' ? serverMetricSeries : clientMetricSeries;
- const valBucket =
- info.side === 'server' ? serverMetricVals : clientMetricVals;
- const valRec = valBucket.get(info.callId) ?? {};
- valRec[info.metric.id] = v;
- valBucket.set(info.callId, valRec);
- const sRec = seriesBucket.get(info.callId) ?? {};
- sRec[info.metric.id] = seriesFromMqe(shape);
- seriesBucket.set(info.callId, sRec);
- }
- }
+ }
+
+ const [nodeEnv, edgeEnv] = await Promise.all([
+ fetchAliasedChunks<MqeShape>(opts, nodeFragments, 150, 'NodeMetrics'),
+ fetchAliasedChunks<MqeShape>(opts, edgeFragments, 200, 'EdgeMetrics'),
+ ]);
+
+ for (const [alias, shape] of Object.entries(nodeEnv)) {
+ const info = nodeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const rec = nodeMetricVals.get(info.nodeId) ?? {};
+ rec[info.metric.id] = v;
+ nodeMetricVals.set(info.nodeId, rec);
+ }
+ for (const [alias, shape] of Object.entries(edgeEnv)) {
+ const info = edgeAliasMap.get(alias);
+ if (!info) continue;
+ const v = aggregateMqe(shape, info.metric.aggregation ?? 'avg');
+ const seriesBucket = info.side === 'server' ? serverMetricSeries : clientMetricSeries;
+ const valBucket = info.side === 'server' ? serverMetricVals : clientMetricVals;
+ const valRec = valBucket.get(info.callId) ?? {};
+ valRec[info.metric.id] = v;
+ valBucket.set(info.callId, valRec);
+ const sRec = seriesBucket.get(info.callId) ?? {};
+ sRec[info.metric.id] = seriesFromMqe(shape);
+ seriesBucket.set(info.callId, sRec);
}
// ── Build response. Connected nodes only — a service with zero
diff --git a/apps/bff/src/util/mapPool.test.ts b/apps/bff/src/util/mapPool.test.ts
new file mode 100644
index 0000000..fa75460
--- /dev/null
+++ b/apps/bff/src/util/mapPool.test.ts
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+import { describe, it, expect } from 'vitest';
+import { mapPool } from './mapPool.js';
+
+const tick = () => new Promise((r) => setTimeout(r, 0));
+
+describe('mapPool', () => {
+ it('preserves input order regardless of completion order', async () => {
+ const out = await mapPool([1, 2, 3, 4, 5], 2, async (n) => {
+ // Smaller numbers resolve later, so completion order != input order.
+ await new Promise((r) => setTimeout(r, (6 - n) * 2));
+ return n * 10;
+ });
+ expect(out).toEqual([10, 20, 30, 40, 50]);
+ });
+
+ it('never exceeds the concurrency cap in flight', async () => {
+ let inFlight = 0;
+ let peak = 0;
+ await mapPool(Array.from({ length: 12 }, (_, i) => i), 3, async (i) => {
+ inFlight++;
+ peak = Math.max(peak, inFlight);
+ await tick();
+ inFlight--;
+ return i;
+ });
+ expect(peak).toBeLessThanOrEqual(3);
+ expect(peak).toBeGreaterThan(1); // actually ran concurrently
+ });
+
+ it('processes every item exactly once', async () => {
+ const seen: number[] = [];
+ const out = await mapPool([5, 6, 7], 5, async (n) => {
+ seen.push(n);
+ return n;
+ });
+ expect(out).toEqual([5, 6, 7]);
+ expect(seen.sort()).toEqual([5, 6, 7]);
+ });
+
+ it('returns [] for empty input without invoking fn', async () => {
+ let called = false;
+ const out = await mapPool([], 4, async (x) => {
+ called = true;
+ return x;
+ });
+ expect(out).toEqual([]);
+ expect(called).toBe(false);
+ });
+
+ it('caps the worker count at the item count (concurrency > length)', async () => {
+ let inFlight = 0;
+ let peak = 0;
+ await mapPool([1, 2], 10, async (n) => {
+ inFlight++;
+ peak = Math.max(peak, inFlight);
+ await tick();
+ inFlight--;
+ return n;
+ });
+ expect(peak).toBeLessThanOrEqual(2);
+ });
+
+ it('treats a zero / negative concurrency as serial (>= 1 worker)', async () => {
+ const out = await mapPool([1, 2, 3], 0, async (n) => n + 1);
+ expect(out).toEqual([2, 3, 4]);
+ });
+});
diff --git a/apps/bff/src/util/mapPool.ts b/apps/bff/src/util/mapPool.ts
new file mode 100644
index 0000000..31f39f4
--- /dev/null
+++ b/apps/bff/src/util/mapPool.ts
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+/**
+ * Bounded-concurrency `Promise.all`. Runs `fn` over `items` with at most
+ * `concurrency` in flight at once, preserving input order in the result.
+ *
+ * Used by the topology routes to fan their per-node / per-edge MQE chunks
+ * out concurrently instead of one-after-another, while still capping how
+ * many heavy aliased queries hit OAP simultaneously. `fn` is expected to be
+ * self-contained re: failure — it should resolve (e.g. to `null`) rather
+ * than reject if a single unit's work is allowed to soft-fail, since one
+ * rejection aborts the whole pool.
+ */
+export async function mapPool<T, R>(
+ items: readonly T[],
+ concurrency: number,
+ fn: (item: T, index: number) => Promise<R>,
+): Promise<R[]> {
+ const results = new Array<R>(items.length);
+ const width = Math.max(1, Math.min(Math.floor(concurrency) || 1, items.length));
+ let next = 0;
+ const worker = async (): Promise<void> => {
+ for (;;) {
+ const i = next++;
+ if (i >= items.length) return;
+ results[i] = await fn(items[i], i);
+ }
+ };
+ await Promise.all(Array.from({ length: width }, worker));
+ return results;
+}