This is an automated email from the ASF dual-hosted git repository. wu-sheng pushed a commit to branch feat/infra-3d-live-pipeline in repository https://gitbox.apache.org/repos/asf/skywalking-horizon-ui.git
commit 5c89a67eb1317ce9b40e14b25b62ca3601294762 Author: Wu Sheng <[email protected]> AuthorDate: Fri May 29 22:44:34 2026 +0800 feat(infra-3d): live topology loader behind ?live=1 (phase 1) Add a sequential, concurrency-1 live loader that assembles a DemoTopology from per-layer OAP reads (bff.layer.services + bff.layer.topology for the serviceMap-capable layers), byte-compatible with the committed snapshot so buildSceneGraph consumes either source unchanged. It is wired as a parallel set of pipeline impls selected by the `?live=1` query param; with no param (default) the scene renders from the snapshot exactly as before, so the live wire is opt-in and cannot break the map. Phase 1 stops at assembling the topology + reporting live counts to the timeline. It does NOT yet swap the scene's data source (the assembly is held in liveTopo as a Phase-2 seam), does NOT build the cross-layer hierarchy (stays []), and does NOT pre-compute cross-layer edges (buildSceneGraph derives those from cross-layer calls). Validated against the demo OAP: 73 services / 17 layers, 3 topology maps (k8s_service / general / mesh), 21 sequential calls in ~6s, zero errors; service ids (base64(name).1) and call shapes match the snapshot contract. --- apps/ui/src/features/infra-3d/Infra3DView.vue | 93 +++++++++- .../infra-3d/composables/useLiveTopology.ts | 196 +++++++++++++++++++++ 2 files changed, 286 insertions(+), 3 deletions(-) diff --git a/apps/ui/src/features/infra-3d/Infra3DView.vue b/apps/ui/src/features/infra-3d/Infra3DView.vue index 4a83a10..0b8e686 100644 --- a/apps/ui/src/features/infra-3d/Infra3DView.vue +++ b/apps/ui/src/features/infra-3d/Infra3DView.vue @@ -22,7 +22,7 @@ operator can toggle whole tiers or individual zones. --> <script setup lang="ts"> -import { computed, onMounted, onUnmounted, ref, watch } from 'vue'; +import { computed, onMounted, onUnmounted, ref, shallowRef, watch } from 'vue'; import { useRoute } from 'vue-router'; import { useLayers } from '@/shell/useLayers'; import type { ServiceNamingRule } from '@skywalking-horizon-ui/api-client'; @@ -31,6 +31,7 @@ import PipelineTimeline from './PipelineTimeline.vue'; import { buildSceneGraph, loadDemoTopology, + type DemoTopology, type SceneServiceNode, } from './composables/useDemoTopology'; import { @@ -43,6 +44,14 @@ import logoSw from '@/assets/icons/logo-sw.svg?raw'; import { useInfra3dConfig } from './composables/useInfra3dConfig'; import { useInfra3dPipeline, type PipelineStageId, type StageImpl } from './composables/useInfra3dPipeline'; import { setValues as setMetricValues, setUnitForLayer, reset as resetMetrics } from './composables/useInfra3dMetrics'; +import { + isTopologyBearing, + liveRoster, + liveSkeleton, + loadLiveServices, + loadLiveTopologies, + type LiveWindow, +} from './composables/useLiveTopology'; import { bff, type Infra3dConfig } from '@/api/client'; /** Imperative handle on the scene's camera-control methods. The @@ -139,7 +148,22 @@ const { stages, stageOrder, running: pipelineRunning, run: runPipelineState } = interface PipelineCtx { servicesByLayer: Record<string, SceneServiceNode[]>; + /** Live path only: the DemoTopology assembled as stages land. The + * snapshot impls leave it null and read loadDemoTopology() directly. */ + topo: DemoTopology | null; } + +// Phase-1 seam for live data. Opt-in per visit with `?live=1`; with no +// param (default) the scene renders from the committed snapshot exactly +// as before, so the live wire can't break the map. `liveTopo` holds the +// latest assembly so a later phase can swap the scene's data source. +const liveTopologyEnabled = computed(() => route.query.live === '1'); +const liveTopo = shallowRef<DemoTopology | null>(null); +const liveWindow = (): LiveWindow => ({ + startMs: Date.now() - 2 * 3600_000, + endMs: Date.now(), + step: 'HOUR', +}); const pipelineImpls: Record<PipelineStageId, StageImpl<PipelineCtx>> = { services: async (rep, ctx) => { rep.start(); @@ -341,9 +365,72 @@ const pipelineImpls: Record<PipelineStageId, StageImpl<PipelineCtx>> = { }, }; +// Live variant of the pipeline: same five stages, but `services` / +// `templates` / `topologies` / `layout` read from sequential per-layer OAP +// queries (assembled into a DemoTopology) instead of the snapshot. The +// `metrics` stage is identical — it consumes ctx.servicesByLayer either way. +const livePipelineImpls: Record<PipelineStageId, StageImpl<PipelineCtx>> = { + services: async (rep, ctx) => { + const roster = liveRoster(menuLayers.value); + const topo = liveSkeleton(roster); + await loadLiveServices(rep, roster, topo); + ctx.topo = topo; + liveTopo.value = topo; + const byLayer: Record<string, SceneServiceNode[]> = {}; + for (const [key, refs] of Object.entries(topo.servicesByLayer)) { + byLayer[key] = refs.map((s) => ({ + nodeId: `${key.toUpperCase()}::${s.id}`, + layerKey: key, + serviceId: s.id, + name: s.name, + shortName: s.name.split('::').slice(-1)[0]!.split('.')[0]!, + normal: s.normal, + })); + } + ctx.servicesByLayer = byLayer; + }, + templates: async (rep, ctx) => { + rep.start(); + const rosterKeys = new Set((ctx.topo?.layers ?? []).map((l) => l.key)); + const withTopology = menuLayers.value + .filter((L) => rosterKeys.has(L.key) && isTopologyBearing(L)) + .map((L) => L.key); + const withoutTopology = [...rosterKeys].filter((k) => !withTopology.includes(k)); + rep.ok(`${withTopology.length} topology-bearing`, { + kind: 'templates', + layersWithTopology: withTopology, + layersWithoutTopology: withoutTopology, + }); + }, + topologies: async (rep, ctx) => { + const topo = ctx.topo; + if (!topo) { + rep.ok('no topology', { kind: 'topologies', probes: [] }); + return; + } + const rosterKeys = new Set(topo.layers.map((l) => l.key)); + const bearing = menuLayers.value.filter((L) => rosterKeys.has(L.key) && isTopologyBearing(L)); + await loadLiveTopologies(rep, bearing, topo, liveWindow()); + liveTopo.value = topo; + }, + layout: async (rep, ctx) => { + rep.start(); + const t0 = performance.now(); + const g = buildSceneGraph(ctx.topo ?? loadDemoTopology(), levelForLayer); + const p = computePlacement(g, planeOrder.value, infraGroups.value); + const ms = Math.round(performance.now() - t0); + rep.ok(`${p.zones.length} zones laid in ${ms} ms`, { + kind: 'layout', + layersReLaid: p.zones.length, + ms, + }); + }, + metrics: pipelineImpls.metrics, +}; + async function runPipeline(): Promise<void> { - const ctx: PipelineCtx = { servicesByLayer: {} }; - await runPipelineState(ctx, pipelineImpls); + const ctx: PipelineCtx = { servicesByLayer: {}, topo: null }; + await runPipelineState(ctx, liveTopologyEnabled.value ? livePipelineImpls : pipelineImpls); } /** Arm (or re-arm) the auto-refresh timer + countdown anchor. */ diff --git a/apps/ui/src/features/infra-3d/composables/useLiveTopology.ts b/apps/ui/src/features/infra-3d/composables/useLiveTopology.ts new file mode 100644 index 0000000..3a9d12c --- /dev/null +++ b/apps/ui/src/features/infra-3d/composables/useLiveTopology.ts @@ -0,0 +1,196 @@ +/* + * 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. + */ + +/** + * Live assembly of the 3D-map topology — the sequential, low-concurrency + * counterpart to the committed `demo-topology.json` snapshot. + * + * The output is a `DemoTopology` byte-compatible with `loadDemoTopology()`, + * so `buildSceneGraph` consumes either source unchanged. It is built one + * layer at a time (concurrency 1) from OAP: per-layer service rosters + * (stage `services`) and per-layer service maps (stage `topologies`). + * + * Phase-1 scope: cross-layer call edges fall out of `buildSceneGraph` + * itself (it derives them from cross-layer entries in `topologies[].calls`), + * so nothing here pre-computes them; `hierarchy` stays `[]` (the Smartscape + * peers are a later phase, fetched per-service). The reporter wiring is + * split across the two stages by the caller so each owns its own + * `StageReporter` — see Infra3DView's live pipeline impls. + */ + +import type { LayerDef } from '@skywalking-horizon-ui/api-client'; +import { bff } from '@/api/client'; +import type { + DemoLayer, + DemoLayerTopology, + DemoServiceRef, + DemoTopology, + DemoTopologyCall, +} from './useDemoTopology'; +import type { StageReporter, TopologyProbe } from './useInfra3dPipeline'; + +export interface LiveWindow { + startMs: number; + endMs: number; + step: 'MINUTE' | 'HOUR' | 'DAY'; +} + +/** A layer ships a service-map iff OAP advertises the `serviceMap` cap — + * the same signal that gates the 2D Service Map page and the only layers + * `bff.layer.topology` returns edges for. */ +export function isTopologyBearing(L: LayerDef): boolean { + return L.caps.serviceMap === true; +} + +/** Active layers with at least one reporting service, tier-ordered (level + * asc, then key) to match the snapshot's ordering. `serviceCount` of -1 + * (OAP unreachable) or 0 is dropped so the scene never gains empty zones. */ +export function liveRoster(layers: LayerDef[]): LayerDef[] { + return layers + .filter((L) => L.active && L.serviceCount > 0) + .slice() + .sort( + (a, b) => + (a.level ?? Number.MAX_SAFE_INTEGER) - (b.level ?? Number.MAX_SAFE_INTEGER) || + a.key.localeCompare(b.key), + ); +} + +function toDemoLayer(L: LayerDef): DemoLayer { + return { + key: L.key, + name: L.name, + level: L.level, + group: L.group ?? null, + serviceCount: L.serviceCount, + color: L.color, + }; +} + +/** Layers-only skeleton; `servicesByLayer` / `topologies` fill in as the + * sequential stages land. `hierarchy` stays `[]` in Phase 1. */ +export function liveSkeleton(roster: LayerDef[]): DemoTopology { + return { + capturedAt: new Date().toISOString(), + oapDemo: 'live', + layers: roster.map(toDemoLayer), + servicesByLayer: {}, + hierarchy: [], + topologies: {}, + }; +} + +/** Stage `services` — read each layer's roster from OAP one at a time, + * reporting per-layer progress. Mutates `topo.servicesByLayer` and returns + * the running service total. An unreachable layer contributes nothing + * (never a fabricated row) and is logged, not thrown — a single failure + * must not abort the pipeline. */ +export async function loadLiveServices( + rep: StageReporter, + roster: LayerDef[], + topo: DemoTopology, +): Promise<number> { + rep.start(); + let total = 0; + for (let i = 0; i < roster.length; i++) { + const L = roster[i]!; + rep.progress(`reading ${L.key} services (${i + 1}/${roster.length})`, { + kind: 'services', + servicesTotal: total, + layersTotal: Object.keys(topo.servicesByLayer).length, + addedSince: null, + removedSince: null, + }); + try { + const resp = await bff.layer.services(L.key); + if (resp.reachable && resp.services.length > 0) { + const refs: DemoServiceRef[] = resp.services.map((s) => ({ + id: s.id, + name: s.name, + normal: s.normal ?? true, + })); + topo.servicesByLayer[L.key] = refs; + total += refs.length; + } + } catch (err) { + console.warn(`[infra-3d] live services failed for ${L.key}:`, err); + } + } + rep.ok(`${total} services / ${Object.keys(topo.servicesByLayer).length} layers`, { + kind: 'services', + servicesTotal: total, + layersTotal: Object.keys(topo.servicesByLayer).length, + addedSince: null, + removedSince: null, + }); + return total; +} + +/** Stage `topologies` — pull each topology-bearing layer's service map one + * at a time. Builds a `DemoLayerTopology` (calls + faithful-but-inert nodes) + * and a per-layer `TopologyProbe`. Per-layer try/catch so one failure + * records a `failed` probe and continues. */ +export async function loadLiveTopologies( + rep: StageReporter, + topoLayers: LayerDef[], + topo: DemoTopology, + window: LiveWindow, +): Promise<TopologyProbe[]> { + rep.start(); + const probes: TopologyProbe[] = []; + for (let i = 0; i < topoLayers.length; i++) { + const L = topoLayers[i]!; + rep.progress(`fetching ${L.key} topology (${i + 1}/${topoLayers.length})`, { + kind: 'topologies', + probes, + }); + const t0 = performance.now(); + try { + const resp = await bff.layer.topology(L.key, undefined, 1, window); + const map: DemoLayerTopology = { + nodes: resp.nodes.map((n) => ({ id: n.id, name: n.name, layer: L.key })), + calls: resp.calls.map<DemoTopologyCall>((c) => ({ + source: c.source, + target: c.target, + detectPoints: c.detectPoints, + })), + }; + topo.topologies[L.key] = map; + probes.push({ + layerKey: L.key, + status: resp.reachable ? (map.calls.length > 0 ? 'ok' : 'empty') : 'failed', + ms: Math.round(performance.now() - t0), + nodeCount: map.nodes.length, + edgeCount: map.calls.length, + error: resp.reachable ? undefined : resp.error, + }); + } catch (err) { + probes.push({ + layerKey: L.key, + status: 'failed', + ms: Math.round(performance.now() - t0), + error: err instanceof Error ? err.message : String(err), + }); + } + } + const okCount = probes.filter((p) => p.status === 'ok').length; + const failed = probes.filter((p) => p.status === 'failed').length; + const detail = { kind: 'topologies' as const, probes }; + if (failed > 0) rep.warn(`${okCount} maps with edges · ${failed} failed`, detail); + else rep.ok(`${okCount} maps with edges`, detail); + return probes; +}
