仪表盘重做：ECharts 曲线、常驻遥测、路由行直达详情
采集从仪表盘组件挪进模块级单例（web/telemetry.ts），流量曲线和每路 吞吐的历史跨页面留存——跟着组件走的话切一次页面就清零一次。每路吞吐 从 /connections 快照按连接逐条求差再按 inboundName 聚合：先求和再求差 会在连接关闭时算出负数。 图表换 ECharts（按需注册）：手写 SVG 折线做不出平滑曲线和时间轴， 颜色在运行时从 CSS 变量取，跟随主题切换重取。 路由列表每行带走势、速率和可选的公网 IP 检测，点击跳路由页并选中 那一路（外壳传 focus 给 RouterPage 消费）。
This commit is contained in:
@@ -0,0 +1,191 @@
|
||||
/**
|
||||
* 常驻的遥测采集：流量曲线、内核内存、每路入站的吞吐。
|
||||
*
|
||||
* 住在模块级单例里而不是仪表盘组件里，是因为历史要跨页面留存——采集跟着组件
|
||||
* 走的话，切走一次曲线就清零一次。应用启动后由外壳调一次 startTelemetry()，
|
||||
* 之后跟着 kernel.online 自己起落：上线接上，掉线断开但**不清历史**，断档在
|
||||
* 时间轴上自然呈现为一段空白。
|
||||
*
|
||||
* 数据来源两路：
|
||||
* - /traffic、/memory 流，每秒一行（内核源码 hub/route/server.go:386-414）
|
||||
* - /connections 快照 3 秒一轮询，按连接逐条求差聚合出每路的吞吐
|
||||
* （快照本身没有流式 HTTP 版本，见 hub/route/connections.go:24-28）
|
||||
*/
|
||||
import { ref, watch } from 'vue'
|
||||
import {
|
||||
ROUTE_POLL_MS,
|
||||
ROUTE_WINDOW_MS,
|
||||
TRAFFIC_WINDOW_MS,
|
||||
pushWindow,
|
||||
routeDeltas,
|
||||
} from '@shared/telemetry.ts'
|
||||
import type { ConnTotals, RouteSample, TrafficPoint } from '@shared/telemetry.ts'
|
||||
import type { ExitProbe } from '@shared/host-info.ts'
|
||||
import { api } from './api'
|
||||
import { kernel } from './runtime'
|
||||
import { readLines } from './stream'
|
||||
|
||||
/** 流量曲线，最近 30 分钟。up/down 是内核报的每秒字节数。 */
|
||||
export const traffic = ref<TrafficPoint[]>([])
|
||||
|
||||
/** 内核报的累计上下行字节数。没收到过是 null。 */
|
||||
export const totals = ref<{ up: number; down: number } | null>(null)
|
||||
|
||||
/** 内核自己的内存占用。 */
|
||||
export const kernelMemory = ref<number | null>(null)
|
||||
|
||||
/** 每路入站的吞吐样本，键是内核连接快照里的 inboundName。 */
|
||||
export const routeSamples = ref<Record<string, RouteSample[]>>({})
|
||||
|
||||
/** 当前连接数。 */
|
||||
export const connCount = ref<number | null>(null)
|
||||
|
||||
/** 流当前接着没有。断流期间仪表盘的「当前值」要显示成没数据，而不是冻在最后一秒。 */
|
||||
export const live = ref(false)
|
||||
|
||||
/** 每个端口的公网 IP 检测结果，会话内缓存。键是端口号。 */
|
||||
export const exitProbes = ref<Record<number, { state: 'loading' | 'done'; result?: ExitProbe }>>({})
|
||||
|
||||
/** 采集的时钟，每次轮询走一格。活跃判定这类「相对现在」的计算依赖它保持新鲜。 */
|
||||
export const tick = ref(Date.now())
|
||||
|
||||
const num = (value: unknown): number =>
|
||||
typeof value === 'number' && Number.isFinite(value) ? value : 0
|
||||
|
||||
let abort: AbortController | null = null
|
||||
let pollTimer = 0
|
||||
let retryTimer = 0
|
||||
let prevTotals: ConnTotals = new Map()
|
||||
let prevAt = 0
|
||||
|
||||
async function consume(
|
||||
path: string,
|
||||
ctrl: AbortController,
|
||||
onLine: (payload: Record<string, unknown>) => void,
|
||||
) {
|
||||
const body = await api.openStream(path, ctrl.signal)
|
||||
for await (const line of readLines(body)) {
|
||||
onLine(JSON.parse(line) as Record<string, unknown>)
|
||||
}
|
||||
}
|
||||
|
||||
function startStreams() {
|
||||
const ctrl = new AbortController()
|
||||
abort = ctrl
|
||||
live.value = true
|
||||
const run = (path: string, onLine: (payload: Record<string, unknown>) => void) =>
|
||||
void consume(path, ctrl, onLine)
|
||||
.catch(() => {})
|
||||
// 这两个流本身永不结束，正常返回也意味着断了
|
||||
.finally(() => {
|
||||
if (ctrl.signal.aborted) return
|
||||
// 一条断了把另一条也收掉，然后走统一的重连节奏
|
||||
ctrl.abort()
|
||||
live.value = false
|
||||
scheduleRetry()
|
||||
})
|
||||
|
||||
run('/traffic', (p) => {
|
||||
traffic.value = pushWindow(
|
||||
traffic.value,
|
||||
{ t: Date.now(), up: num(p.up), down: num(p.down) },
|
||||
TRAFFIC_WINDOW_MS,
|
||||
)
|
||||
totals.value = { up: num(p.upTotal), down: num(p.downTotal) }
|
||||
})
|
||||
run('/memory', (p) => {
|
||||
kernelMemory.value = num(p.inuse)
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 断流不等于内核没了：保存配置触发的重载会掐断流，内核几秒内就回来、online 从未
|
||||
* 翻转。5 秒后重试，失败会再次走到这里形成节奏；真掉线交给下面 online 的 watch。
|
||||
*/
|
||||
function scheduleRetry() {
|
||||
window.clearTimeout(retryTimer)
|
||||
retryTimer = window.setTimeout(() => {
|
||||
if (kernel.value.online && !live.value) startStreams()
|
||||
}, 5_000)
|
||||
}
|
||||
|
||||
async function pollConnections() {
|
||||
const now = Date.now()
|
||||
try {
|
||||
const snapshot = await api.connections()
|
||||
const conns = snapshot.connections ?? []
|
||||
connCount.value = conns.length
|
||||
const { deltas, next } = routeDeltas(prevTotals, conns)
|
||||
const ms = prevAt ? now - prevAt : ROUTE_POLL_MS
|
||||
prevTotals = next
|
||||
prevAt = now
|
||||
|
||||
const merged = { ...routeSamples.value }
|
||||
for (const [key, delta] of Object.entries(deltas)) {
|
||||
merged[key] = pushWindow(
|
||||
merged[key] ?? [],
|
||||
{ t: now, ms, up: delta.up, down: delta.down },
|
||||
ROUTE_WINDOW_MS,
|
||||
)
|
||||
}
|
||||
// 有历史但这一轮没有连接的路，补零样本让走势落回去而不是冻住；
|
||||
// 整个窗口都归零之后把这路忘掉，改名、删除的入站才不会在内存里越攒越多
|
||||
for (const key of Object.keys(merged)) {
|
||||
if (key in deltas) continue
|
||||
const samples = pushWindow(merged[key]!, { t: now, ms, up: 0, down: 0 }, ROUTE_WINDOW_MS)
|
||||
if (samples.every((s) => s.up === 0 && s.down === 0)) delete merged[key]
|
||||
else merged[key] = samples
|
||||
}
|
||||
routeSamples.value = merged
|
||||
} catch {
|
||||
// 单轮失败就跳过。持续失败说明内核掉了，online 的 watch 会把轮询停掉
|
||||
} finally {
|
||||
tick.value = now
|
||||
}
|
||||
}
|
||||
|
||||
function attach() {
|
||||
if (!abort) startStreams()
|
||||
if (!pollTimer) {
|
||||
void pollConnections()
|
||||
pollTimer = window.setInterval(() => void pollConnections(), ROUTE_POLL_MS)
|
||||
}
|
||||
}
|
||||
|
||||
function detach() {
|
||||
abort?.abort()
|
||||
abort = null
|
||||
live.value = false
|
||||
window.clearInterval(pollTimer)
|
||||
window.clearTimeout(retryTimer)
|
||||
pollTimer = 0
|
||||
prevTotals = new Map()
|
||||
prevAt = 0
|
||||
}
|
||||
|
||||
let started = false
|
||||
|
||||
/** 外壳登录成功后调一次。之后采集跟着内核在线状态自己起落。 */
|
||||
export function startTelemetry(): void {
|
||||
if (started) return
|
||||
started = true
|
||||
watch(
|
||||
() => kernel.value.online,
|
||||
(online) => {
|
||||
if (online) attach()
|
||||
else detach()
|
||||
},
|
||||
{ immediate: true },
|
||||
)
|
||||
}
|
||||
|
||||
/** 检测某个端口的公网 IP。结果缓存到会话结束，force 重测。 */
|
||||
export async function probeExit(port: number, force = false): Promise<void> {
|
||||
const cached = exitProbes.value[port]
|
||||
if (!force && cached && (cached.state === 'loading' || cached.result?.ok)) return
|
||||
exitProbes.value = { ...exitProbes.value, [port]: { state: 'loading' } }
|
||||
const result = await api
|
||||
.probeExit(port)
|
||||
.catch((error: unknown) => ({ ok: false, message: String(error) }) as ExitProbe)
|
||||
exitProbes.value = { ...exitProbes.value, [port]: { state: 'done', result } }
|
||||
}
|
||||
Reference in New Issue
Block a user