From 66efc56ab86f8f78202dcc61c201d033b66d40d8 Mon Sep 17 00:00:00 2001 From: Dillon Lillehaug Date: Sun, 13 Sep 2026 15:20:45 +0000 Subject: [PATCH] feat(core): show live per-DSP resource and backup diagnostics --- core/core/api/access-http.js | 2 +- .../examples/independent-updates-preview.js | 4 +- .../frontend/src/pages/Diagnostics.tsx | 33 +------ .../frontend/src/pages/DspResources.tsx | 86 ++++++++++++++++ core/dashboard/playwright.updates.config.cjs | 2 +- .../browser/diagnostics-resources.spec.cjs | 36 +++++++ core/host/capacity/README.md | 20 ++++ core/host/capacity/monitor.js | 82 +++++++-------- core/host/capacity/resources.js | 53 ++++++++++ core/host/capacity/storage.js | 99 +++++++++++++++++++ core/host/capacity/tests/resources.test.js | 44 +++++++++ core/host/capacity/tests/storage.test.js | 27 +++++ core/tooling/tests.json | 1 + 13 files changed, 410 insertions(+), 79 deletions(-) create mode 100644 core/dashboard/frontend/src/pages/DspResources.tsx create mode 100644 core/dashboard/tests/browser/diagnostics-resources.spec.cjs create mode 100644 core/host/capacity/README.md create mode 100644 core/host/capacity/resources.js create mode 100644 core/host/capacity/storage.js create mode 100644 core/host/capacity/tests/resources.test.js create mode 100644 core/host/capacity/tests/storage.test.js diff --git a/core/core/api/access-http.js b/core/core/api/access-http.js index bcd9e2d..2662cf9 100644 --- a/core/core/api/access-http.js +++ b/core/core/api/access-http.js @@ -370,7 +370,7 @@ function createAccessHttp({ const current = session(request); requireNoQuery(url); access.requirePlatform(current, 'platform.installations.manage'); - if (current.user.platformRole !== 'owner') throw new AccessError('platform_forbidden', 403); + if (current.user.platformRole !== 'owner' || current.dspView) throw new AccessError('platform_forbidden', 403); sendJson(response, 200, { ok: true, status: 'found', data: platformRuntime?.() || { enabled: false, storageAvailableBytes: null, runtimes: [] }, error: null }); return true; diff --git a/core/dashboard/examples/independent-updates-preview.js b/core/dashboard/examples/independent-updates-preview.js index b526594..0b2d05d 100644 --- a/core/dashboard/examples/independent-updates-preview.js +++ b/core/dashboard/examples/independent-updates-preview.js @@ -10,7 +10,7 @@ const { UpdateWorker } = require('../../core/updates/worker'); const { createUpdatesService } = require('../../core/updates/service'); const { hash, inventory } = require('../../shared/releases/package'); const { createDashboardServer } = require('../server/server'); -async function createPreview({ port = 0, automatic = true, versionedDashboards = false } = {}) { +async function createPreview({ port = 0, automatic = true, versionedDashboards = false, platformRuntime = null } = {}) { const root = fs.mkdtempSync(path.join(os.tmpdir(), 'dispatch-independent-updates-')); const store = new AccessStore({ databaseRoot: path.join(root, 'access'), database: path.join(root, 'access/control.sqlite3') }); const access = new AccessControlService(store, { installationOperatorEnabled: true, installationBackend: 'directory_service_v1' }); @@ -80,7 +80,7 @@ async function createPreview({ port = 0, automatic = true, versionedDashboards = await worker.initialize(); const updates = createUpdatesService({ releases, commands, store, devDspId: dsps[0] }); const unavailable = async () => ({ ok: false, status: 'installation_not_ready', data: null, error: { code: 'installation_not_ready' } }); - const server = createDashboardServer({ access, updates, ...(versionedDashboards ? {dashboards:require('../../core/updates/dashboard').dashboardProvider({paths:{local:path.join(root,'local')},store})} : {}), plugins: { catalog: () => ({ items: [] }) }, + const server = createDashboardServer({ access, updates, platformRuntime, ...(versionedDashboards ? {dashboards:require('../../core/updates/dashboard').dashboardProvider({paths:{local:path.join(root,'local')},store})} : {}), plugins: { catalog: () => ({ items: [] }) }, client: { workforce: { day: unavailable }, sync: { status: unavailable, runNow: unavailable }, system: { status: unavailable } } }); const original = server.listeners('request')[0]; server.removeAllListeners('request'); server.on('request', async (request, response) => { diff --git a/core/dashboard/frontend/src/pages/Diagnostics.tsx b/core/dashboard/frontend/src/pages/Diagnostics.tsx index 98989c8..6385ec5 100644 --- a/core/dashboard/frontend/src/pages/Diagnostics.tsx +++ b/core/dashboard/frontend/src/pages/Diagnostics.tsx @@ -5,6 +5,8 @@ import { idempotent, queryClient, request } from "@/lib/api"; import { Button } from "@/components/ui/button"; import { PageHeading, ErrorNotice, Loading, Notice } from "@/components/shared"; +import { DspResources } from "./DspResources"; + type DiagnosticsView = { enabled: boolean; dsps: { @@ -23,14 +25,6 @@ export function Diagnostics() { queryFn: () => request("/api/platform/diagnostics"), refetchInterval: 5000, }); - const runtime = useQuery({ - queryKey: ["platform-runtime"], - queryFn: () => request<{ enabled: boolean; storageAvailableBytes: number | null; - runtimes: { reference: string; name: string; status: string; memoryBytes: number | null; memoryLimitBytes: number | null; tasks: number | null; - storage: { limited: boolean | null; capacityBytes: number | null; availableBytes: number | null } }[] - }>("/api/platform/runtime"), - refetchInterval: 5000, - }); async function deploy() { if (busy) return; setBusy(true); @@ -53,27 +47,10 @@ export function Diagnostics() { <> - - {runtime.data?.enabled && ( -
-

Runtime health

-

- Available storage: {((runtime.data.storageAvailableBytes ?? 0) / 1024 ** 3).toFixed(1)} GiB -

- {runtime.data.runtimes.map((item) => ( -
- {item.name} - {item.status} · {item.memoryBytes === null ? "—" : `${Math.round(item.memoryBytes / 1024 ** 2)} MiB`} · {item.tasks ?? 0} tasks - {item.storage?.limited ? ` · ${((item.storage.availableBytes ?? 0) / 1024 ** 3).toFixed(1)} GiB storage free` - : item.storage?.limited === false ? " · Storage limit pending migration" : " · Storage unavailable"} - -
- ))} - {!runtime.data.runtimes.length &&

No DSP runtimes yet.

} -
- )} + + {data.isPending ? ( ) : data.data ? ( diff --git a/core/dashboard/frontend/src/pages/DspResources.tsx b/core/dashboard/frontend/src/pages/DspResources.tsx new file mode 100644 index 0000000..8976d67 --- /dev/null +++ b/core/dashboard/frontend/src/pages/DspResources.tsx @@ -0,0 +1,86 @@ +import { useEffect, useState } from "react"; +import { useQuery } from "@tanstack/react-query"; +import { request } from "@/lib/api"; +import { ErrorNotice, Loading } from "@/components/shared"; + +type Storage = { + status: "measuring" | "ready" | "stale" | "unavailable"; + sampledAt?: number | null; + refreshing?: boolean; + limited?: boolean; + usedBytes?: number; + capacityBytes?: number | null; + availableBytes?: number | null; + runtimeBytes?: number; + dataBytes?: number; + pluginBytes?: number; + logBytes?: number; + localBackupBytes?: number; + backups?: { available: boolean; count: number; bytes: number; manual: number; updates: number; plugins: number; lastAt: string | null }; +}; +type RuntimeView = { + enabled: boolean; + sampledAt: number; + storageAvailableBytes: number | null; + runtimes: { reference: string; name: string; status: string; cpuPercent: number | null; memoryBytes: number | null; + tasks: number | null; activeWorkers: number | null; storage: Storage }[]; +}; +function bytes(value?: number | null) { + if (value == null) return "Unavailable"; + if (value < 1024) return `${value} B`; + const power = Math.min(4, Math.floor(Math.log(value) / Math.log(1024))); + return `${(value / 1024 ** power).toFixed(1)} ${["B", "KiB", "MiB", "GiB", "TiB"][power]}`; +} +function age(at: number | null | undefined, now: number) { + if (at == null) return "Waiting for measurement"; + const seconds = Math.max(0, Math.floor((now - at) / 1000)); + return seconds < 60 ? `${seconds}s ago` : `${Math.floor(seconds / 60)}m ago`; +} +function Metric({ label, value, children }: { label: string; value: string; children?: React.ReactNode }) { + return
{label}
+
{value}
+ {children &&
{children}
} +
; +} +export function DspResources() { + const [now, setNow] = useState(Date.now); + useEffect(() => { const timer = window.setInterval(() => setNow(Date.now()), 1000); return () => window.clearInterval(timer); }, []); + const runtime = useQuery({ queryKey: ["platform-runtime"], queryFn: () => request("/api/platform/runtime"), refetchInterval: 2000 }); + const data = runtime.data; + const stale = Boolean(runtime.error || (data && now - data.sampledAt > 10000)); + return
+
+

DSP resources

CPU and RAM refresh every 2 seconds. Storage and backups refresh every minute.

+

+
+ + {runtime.isPending && } + {data?.enabled === false &&

Resource monitoring is unavailable on this installation.

} + {data?.enabled && <> +

Measured {age(data.sampledAt, now)} · Host storage available: {bytes(data.storageAvailableBytes)}{stale ? " · Showing last received measurements" : ""}

+ {data.runtimes.map(item => { + const storage = item.storage, backup = storage?.backups; + const storageStale = storage?.status === "stale" || (storage?.sampledAt != null && now - storage.sampledAt > 120000); + return
+

{item.name}

{item.status}
+
+

100% equals one CPU core

 +

{item.tasks ?? "—"} tasks · {item.activeWorkers ?? "—"} plugin / browser workers

 + + {storage?.limited &&

{bytes(storage.capacityBytes)} capacity · {bytes(storage.availableBytes)} free

} +

Data {bytes(storage?.dataBytes)} · Plugins {bytes(storage?.pluginBytes)}

+

Logs {bytes(storage?.logBytes)} · Local backups {bytes(storage?.localBackupBytes)}

+

Runtime code: {bytes(storage?.runtimeBytes)} separately

+ + + {backup?.available && <>

{bytes(backup.bytes)} of backup data

{backup.manual} manual · {backup.updates} update · {backup.plugins} plugin rollback

{backup.lastAt ? `Latest: ${new Date(backup.lastAt).toLocaleString()}` : "No completed backups"}

} + +
+

Storage and backups: {storage?.status === "measuring" ? "Measuring…" : storage?.status === "unavailable" ? "Measurement unavailable" : `${age(storage?.sampledAt, now)}${storageStale ? " · Stale measurement" : ""}${storage?.refreshing ? " · Refreshing…" : ""}`}

+
; + })} + {!data.runtimes.length &&

No DSP runtimes yet.

} +

CPU and RAM include each DSP’s runtime and isolated plugin and browser workers. Shared Core services and centrally stored dashboards are excluded. Backup totals include manual backups, update snapshots and plugin rollback copies; these can overlap local storage usage.

+ } +
; +} diff --git a/core/dashboard/playwright.updates.config.cjs b/core/dashboard/playwright.updates.config.cjs index 2d0a006..2957693 100644 --- a/core/dashboard/playwright.updates.config.cjs +++ b/core/dashboard/playwright.updates.config.cjs @@ -1,6 +1,6 @@ const { defineConfig } = require('@playwright/test'); module.exports = defineConfig({ - testDir: './tests/browser', testMatch: ['independent-updates.spec.cjs', 'update-progress.spec.cjs', 'updates-workspace.spec.cjs', 'dashboard-rollout.spec.cjs'], + testDir: './tests/browser', testMatch: ['diagnostics-resources.spec.cjs', 'independent-updates.spec.cjs', 'update-progress.spec.cjs', 'updates-workspace.spec.cjs', 'dashboard-rollout.spec.cjs'], workers: 1, timeout: 60000, outputDir: process.env.DISPATCH_UI_ARTIFACTS || '/tmp/dispatch-updates-browser', use: { viewport: { width: 1440, height: 1000 }, reducedMotion: 'reduce', diff --git a/core/dashboard/tests/browser/diagnostics-resources.spec.cjs b/core/dashboard/tests/browser/diagnostics-resources.spec.cjs new file mode 100644 index 0000000..f31006b --- /dev/null +++ b/core/dashboard/tests/browser/diagnostics-resources.spec.cjs @@ -0,0 +1,36 @@ +const {test,expect}=require('@playwright/test'); +const {createPreview}=require('../../examples/independent-updates-preview'); +let app,view; +test.beforeAll(async()=>{ + view={enabled:true,storageAvailableBytes:100*1024**3,runtimes:[{reference:'synthetic-dev',name:'Dev DSP',status:'connected',cpuPercent:12.5,memoryBytes:256*1024**2,tasks:12,activeWorkers:1, + storage:{status:'ready',sampledAt:Date.now(),limited:true,usedBytes:2*1024**3,capacityBytes:10*1024**3,availableBytes:8*1024**3,runtimeBytes:20*1024**2,dataBytes:1024**3,pluginBytes:40*1024**2,logBytes:1024,localBackupBytes:1024**3, + backups:{available:true,count:6,bytes:1024**3,manual:2,updates:3,plugins:1,lastAt:'2026-01-01T00:00:00Z'}}}]}; + app=await createPreview({automatic:false,platformRuntime:()=>({...view,sampledAt:Date.now()})}); +}); +test.afterAll(async()=>{await app?.close();}); +test('owner sees live DSP resources, backup breakdown and stale storage on desktop and mobile',async({page},info)=>{ + const errors=[];page.on('pageerror',error=>errors.push(error.message));page.on('console',message=>{if(message.type()==='error')errors.push(message.text());}); + await page.goto(`${app.url}/#/diagnostics`);await page.getByLabel('Email address').fill('platform@example.test');await page.getByLabel('Password',{exact:true}).fill('synthetic preview password');await page.getByRole('button',{name:'Sign in',exact:true}).click(); + await page.locator('.desktop-sidebar').getByRole('link',{name:'Diagnostics',exact:true}).click(); + const card=page.getByRole('article',{name:'Dev DSP resources'}); + await expect(card).toContainText('12.5%');await expect(card).toContainText('256.0 MiB');await expect(card).toContainText('2 manual · 3 update · 1 plugin rollback'); + await page.screenshot({path:info.outputPath('resources-desktop.png'),fullPage:true}); + view.runtimes[0].cpuPercent=87.3;view.runtimes[0].memoryBytes=512*1024**2; + await expect(card).toContainText('87.3%',{timeout:8000});await expect(card).toContainText('512.0 MiB'); + view.runtimes[0].storage.status='stale';view.runtimes[0].cpuPercent=null;view.runtimes[0].memoryBytes=null; + await expect(card).toContainText('Stale measurement',{timeout:8000});await expect(card).toContainText('Measuring / unavailable');await expect(card).toContainText('Unavailable'); + await page.setViewportSize({width:390,height:844});await expect.poll(()=>page.evaluate(()=>document.documentElement.scrollWidth<=innerWidth)).toBe(true); + await page.screenshot({path:info.outputPath('resources-mobile.png'),fullPage:true}); + await page.route('**/api/platform/runtime',route=>route.fulfill({json:{ok:true,status:'found',data:{...view,sampledAt:Date.now()-20000},error:null}})); + await expect(page.getByRole('status').filter({hasText:'Live updates interrupted'})).toBeVisible({timeout:8000});expect(errors).toEqual([]); +}); + +test('runtime endpoint rejects DSP members and the owner’s scoped DSP view',async({request})=>{ + const endpoint=app.url+'/api/platform/runtime'; + expect((await request.get(endpoint)).status()).toBe(401); + expect((await request.get(endpoint,{headers:{Cookie:'dispatch_session='+app.owners[0].token}})).status()).toBe(403); + const organization=app.access.platformOrganizations(app.owner.session)[0]; + const scoped=app.access.beginDspView(app.owner.session,{controlRef:organization.controlRef}); + expect((await request.get(endpoint,{headers:{Cookie:'dispatch_session='+app.owner.token,'X-Dispatch-DSP-View':scoped.dspView.viewRef}})).status()).toBe(403); + expect((await request.get(endpoint,{headers:{Cookie:'dispatch_session='+app.owner.token}})).status()).toBe(200); +}); diff --git a/core/host/capacity/README.md b/core/host/capacity/README.md new file mode 100644 index 0000000..36c4b46 --- /dev/null +++ b/core/host/capacity/README.md @@ -0,0 +1,20 @@ +# DSP resource diagnostics + +The owner-only `/api/platform/runtime` endpoint caches one fleet sample for two +seconds across dashboard clients. CPU is a monotonic cgroup CPU-time delta: +100% means one logical CPU core. The first sample after startup or a cgroup +restart has no CPU delta. Missing counters remain unavailable. + +RAM, CPU and tasks include the DSP runtime plus isolated plugin and authentication +browser workers identified by Core's registries. Shared Core processes and shared +dashboard assets are excluded; worker memory is not compared to the runtime limit. + +Storage scans run asynchronously at most once per minute while requested. They +read file metadata and completed backup manifests, skip symlinks, and have a shared +entry/time budget. Managed volume usage comes from statfs; legacy layouts use +allocated file bytes. Runtime code outside the volume is reported separately. +Backup counts distinguish manual backups, DSP update snapshots and plugin rollback +copies. Backup bytes represent logical backup data, not additional volume usage +or a restore-integrity guarantee. Shared platform backup overhead is excluded. +Failed scans retain timestamped stale values, or report unavailable without a +prior successful sample. No host paths or runtime identifiers enter the response. diff --git a/core/host/capacity/monitor.js b/core/host/capacity/monitor.js index 8ee7a9b..4c35a8d 100644 --- a/core/host/capacity/monitor.js +++ b/core/host/capacity/monitor.js @@ -1,49 +1,37 @@ 'use strict'; - -const fs = require('node:fs'); -const path = require('node:path'); -const crypto = require('node:crypto'); -const { unitName } = require('../services/host'); -const { assertVolumeMounted } = require('../storage/volume-state'); - -function counter(root, name) { - try { - const value = fs.readFileSync(path.join(root, name), 'utf8').trim(); - return /^\d+$/.test(value) && Number.isSafeInteger(Number(value)) ? Number(value) : null; - } catch { return null; } +const fs=require('node:fs'),path=require('node:path'),crypto=require('node:crypto'); +const {unitName}=require('../services/host'); +const {createResourceSampler,workerGroups}=require('./resources'); +const {createStorageSampler}=require('./storage'); +const sum=(values,key)=>values.some(value=>value[key]===null)?null:values.reduce((n,value)=>n+value[key],0); +const zero={memoryBytes:0,memoryLimitBytes:null,tasks:0,cpuPercent:0}; +function createDirectoryMonitor({store,manager,paths,execution=null,clock=Date.now,cgroupRoot='/sys/fs/cgroup/system.slice', + sampleResources=createResourceSampler(),readWorkers=workerGroups,storageSampler=createStorageSampler({paths}),disk=()=>fs.statfsSync(paths.dsps)}={}){ + let cached=null; + return ()=>{ + const now=clock();if(cached&&now-cached.sampledAt<2000)return cached; + const rows=store.db.prepare(`SELECT i.runtime_key,i.status installation_status,o.name FROM installations i JOIN organizations o ON o.id=i.organization_id + WHERE i.backend='directory_service_v1' AND i.status<>'decommissioned' ORDER BY o.created_at,o.id`).all(); + const ids=rows.map(row=>row.runtime_key),workers=readWorkers(paths,ids),storage=storageSampler.read(ids); + const groups=ids.flatMap(id=>[unitName(id),...(workers.groups.get(id)||[])]); + const resources=sampleResources(groups.map(name=>path.join(cgroupRoot,name))); + let storageAvailableBytes=null;try{const value=disk();storageAvailableBytes=value.bavail*value.bsize;}catch{} + cached={enabled:true,sampledAt:now,refreshIntervalMs:2000,storageRefreshIntervalMs:60000,storageAvailableBytes, + resourceScope:'DSP runtime, isolated plugin jobs and browser workers. Shared Core services are excluded.', + runtimes:rows.map(row=>{ + const id=row.runtime_key,record=manager.journal.record(id),runtime=resources.get(path.join(cgroupRoot,unitName(id))); + const worker=execution?.store?.get(id),tasks=runtime?.tasks; + const asleep=worker?.state==='sleeping'&&!(tasks>0)&&['ready','waiting_for_owner','waiting_for_provider_auth'].includes(row.installation_status); + const stopped=asleep||record?.desiredState!=='running'; + const values=[runtime||(stopped?zero:{memoryBytes:null,memoryLimitBytes:null,tasks:null,cpuPercent:null})]; + let activeWorkers=0; + for(const group of workers.groups.get(id)||[]){const value=resources.get(path.join(cgroupRoot,group));if(value){values.push(value);activeWorkers++;}} + return {reference:crypto.createHash('sha256').update(id).digest('hex'),name:row.name, + status:asleep?'sleeping':stopped?tasks>0?'stopping':'stopped':manager.hub.connected(id)?'connected':tasks>0?'starting':'offline', + memoryBytes:workers.available?sum(values,'memoryBytes'):null,memoryLimitBytes:runtime?.memoryLimitBytes??null, + cpuPercent:workers.available?sum(values,'cpuPercent'):null,tasks:workers.available?sum(values,'tasks'):null, + activeWorkers:workers.available?activeWorkers:null,storage:storage.get(id)}; + })};return cached; + }; } - -// Read only the cgroups belonging to Core's directory identities. Public views -// contain resource values and display names, never unit names or filesystem paths. -function createDirectoryMonitor({ store, manager, paths, execution = null }) { - return () => { - const rows = store.db.prepare(`SELECT i.runtime_key,i.status installation_status,o.name FROM installations i JOIN organizations o ON o.id=i.organization_id - WHERE i.backend='directory_service_v1' ORDER BY o.created_at,o.id`).all(); - const disk = fs.statfsSync(paths.dsps); - return { enabled: true, storageAvailableBytes: disk.bavail * disk.bsize, - runtimes: rows.map(row => { - const record = manager.journal.record(row.runtime_key); - const group = path.join('/sys/fs/cgroup/system.slice', unitName(row.runtime_key)); - const tasks = counter(group, 'pids.current'); - const worker = execution?.store?.get(row.runtime_key); - const asleep = worker?.state === 'sleeping' && !(tasks > 0) - && ['ready', 'waiting_for_owner', 'waiting_for_provider_auth'].includes(row.installation_status); - let storage = { limited: false, capacityBytes: null, availableBytes: null }; - try { - const root = path.join(paths.dsps, row.runtime_key), volume = assertVolumeMounted(root); - if (volume) { - const usage = fs.statfsSync(path.join(root, 'data')); - storage = { limited: true, capacityBytes: usage.blocks * usage.bsize, availableBytes: usage.bavail * usage.bsize }; - } - } catch { storage = { limited: null, capacityBytes: null, availableBytes: null }; } - return { reference: crypto.createHash('sha256').update(row.runtime_key).digest('hex'), name: row.name, - status: asleep ? 'sleeping' : record?.desiredState !== 'running' ? tasks > 0 ? 'stopping' : 'stopped' - : manager.hub.connected(row.runtime_key) ? 'connected' : tasks > 0 ? 'starting' : 'offline', - memoryBytes: asleep ? 0 : counter(group, 'memory.current'), memoryLimitBytes: counter(group, 'memory.max'), tasks: asleep ? 0 : tasks, - storage, - }; - }), - }; - }; -} -module.exports = { createDirectoryMonitor }; +module.exports={createDirectoryMonitor}; diff --git a/core/host/capacity/resources.js b/core/host/capacity/resources.js new file mode 100644 index 0000000..74508ea --- /dev/null +++ b/core/host/capacity/resources.js @@ -0,0 +1,53 @@ +'use strict'; +const fs=require('node:fs'),path=require('node:path'),crypto=require('node:crypto'); +const {DatabaseSync}=require('node:sqlite'); +const {privateJson}=require('../../core/installations/src/release-delivery-files'); +const {unitName}=require('../services/scoped-worker'); +function number(value){return /^\d+$/.test(value)&&Number.isSafeInteger(Number(value))?Number(value):null;} +function counter(root,name){try{return number(fs.readFileSync(path.join(root,name),'utf8').trim());}catch{return null;}} +function cpuUsage(root){try{return number(/^usage_usec (\d+)$/m.exec(fs.readFileSync(path.join(root,'cpu.stat'),'utf8'))?.[1]);}catch{return null;}} +function readGroup(root){ + try { + const stat=fs.statSync(root);if(!stat.isDirectory())return null; + return {identity:`${stat.dev}:${stat.ino}`,memoryBytes:counter(root,'memory.current'),memoryLimitBytes:counter(root,'memory.max'),tasks:counter(root,'pids.current'),cpuUsage:cpuUsage(root)}; + }catch(error){if(error.code==='ENOENT')return null;return {identity:null,memoryBytes:null,memoryLimitBytes:null,tasks:null,cpuUsage:null};} +} +// Each sampler is shared by all dashboard clients. CPU is a delta, not lifetime CPU time. +function createResourceSampler({read=readGroup,monotonic=()=>performance.now()}={}){ + const previous=new Map(); + return roots=>{ + const now=monotonic(),values=new Map(); + for(const root of new Set(roots)){ + const value=read(root),before=previous.get(root); + if(!value){previous.delete(root);values.set(root,null);continue;} + const elapsed=before?now-before.at:0; + const cpuPercent=value.identity&&before?.identity===value.identity&&value.cpuUsage!==null&&before.cpuUsage!==null&&value.cpuUsage>=before.cpuUsage&&elapsed>0 + ?(value.cpuUsage-before.cpuUsage)/(elapsed*1000)*100:null; + values.set(root,{...value,cpuPercent});previous.set(root,{...value,at:now}); + } + for(const root of previous.keys())if(!values.has(root))previous.delete(root); + return values; + }; +} +// Use Core's worker registry, never a tenant-supplied list of another DSP's jobs. +function workerGroups(paths,ids){ + const groups=new Map(ids.map(id=>[id,new Set()]));let available=true; + const root=path.join(paths.local,'state/plugin-backend/jobs'); + try{for(const file of fs.readdirSync(root)){ + if(!/^job_[a-f0-9]{32}\.json$/.test(file))continue; + let row;try{row=privateJson(path.join(root,file),process.geteuid());}catch(error){if(error.code==='ENOENT')continue;throw error;} + if(row.schemaVersion!==1||row.jobId+'.json'!==file)throw Error('invalid_worker'); + groups.get(row.dspId)?.add(unitName(row.jobId)); + }}catch(error){if(error.code!=='ENOENT')available=false;} + const file=path.join(paths.local,'state/plugin-backend/browsers.sqlite3');let db; + try{ + const info=fs.lstatSync(file);if(!info.isFile()||info.isSymbolicLink()||info.uid!==process.geteuid()||fs.realpathSync(file)!==file)throw Error('invalid_browser_store'); + db=new DatabaseSync(file,{readOnly:true}); + for(const row of db.prepare("SELECT id,dsp_id FROM browser_leases WHERE state IN ('starting','active','closing')").all()){ + if(!/^browser_[a-f0-9]{48}$/.test(row.id))throw Error('invalid_browser'); + groups.get(row.dsp_id)?.add(unitName('job_'+crypto.createHash('sha256').update(row.id).digest('hex').slice(0,32))); + } + }catch(error){if(error.code!=='ENOENT')available=false;}finally{db?.close();} + return {groups,available}; +} +module.exports={createResourceSampler,workerGroups,readGroup,counter}; diff --git a/core/host/capacity/storage.js b/core/host/capacity/storage.js new file mode 100644 index 0000000..8fcb419 --- /dev/null +++ b/core/host/capacity/storage.js @@ -0,0 +1,99 @@ +'use strict'; +const fs=require('node:fs'),path=require('node:path'); +const fsp=fs.promises; +const {assertVolumeMounted}=require('../storage/volume-state'); +const DSP=/^dsp_[a-f0-9]{32}$/; +async function directory(root){const stat=await fsp.lstat(root);if(!stat.isDirectory()||stat.isSymbolicLink()||await fsp.realpath(root)!==root)throw Error('unsafe_directory');return stat;} +async function names(root){try{await directory(root);return await fsp.readdir(root);}catch(error){if(error.code==='ENOENT')return [];throw error;}} +async function json(file){ + let fd;try{ + await directory(path.dirname(file));fd=await fsp.open(file,fs.constants.O_RDONLY|fs.constants.O_NOFOLLOW); + const stat=await fd.stat();if(!stat.isFile()||stat.size>2*1024*1024)throw Error('invalid_metadata'); + return {value:JSON.parse(await fd.readFile('utf8')),at:stat.mtimeMs}; + }catch(error){if(error.code==='ENOENT')return null;throw error;}finally{await fd?.close();} +} +// Metadata only: never read tenant file contents, follow symlinks, or block the API on a recursive scan. +async function size(root,budget,logical=false){ + let bytes=0;const seen=new Set(); + async function walk(file){ + if(++budget.entries>budget.maximum||Date.now()>budget.deadline)throw Error('scan_limit'); + let stat;try{stat=await fsp.lstat(file);}catch(error){if(error.code==='ENOENT')return;throw error;} + if(stat.isSymbolicLink())return; + const key=`${stat.dev}:${stat.ino}`;if(seen.has(key))return;seen.add(key); + if(stat.isDirectory()){ + await directory(file); + for(const name of await fsp.readdir(file))await walk(path.join(file,name)); + }else if(stat.isFile())bytes+=logical?stat.size:stat.blocks*512; + } + await walk(root);return bytes; +} +const blankBackups=()=>({manual:0,updates:0,plugins:0,count:0,bytes:0,lastAt:null,available:true}); +function addBackup(value,kind,bytes,at){value[kind]++;value.count++;value.bytes+=bytes;value.lastAt=Math.max(value.lastAt||0,at);} +async function backupIndex(paths,ids,budget){ + const result=new Map(ids.map(id=>[id,blankBackups()])); + const manual=path.join(paths.local,'backups/manual'),updates=path.join(paths.local,'backups/updates/dsp'); + for(const [kind,root,pattern,file] of [['manual',manual,/^mbk_[a-f0-9]{32}$/,'manifest.json'],['updates',updates,/^[a-f0-9]{32}$/,'snapshot.json']]){ + try{for(const name of await names(root)){ + if(!pattern.test(name))continue; + if(kind==='manual'&&await json(path.join(root,name,'.erasing.json')))continue; + const record=await json(path.join(root,name,file));if(!record)continue; + const v=record.value; + if(kind==='manual'){ + if(v.id!==name||![1,2].includes(v.version)||!Array.isArray(v.dsps)||!Array.isArray(v.roots)||!Number.isSafeInteger(v.createdAt)||new Set(v.dsps.map(dsp=>dsp.id)).size!==v.dsps.length)throw Error('invalid_backup'); + for(const dsp of v.dsps){if(!result.has(dsp.id))continue; + const rows=v.roots.filter(r=>typeof r.label==='string'&&r.label.startsWith(dsp.id+'_')); + if(rows.some(r=>!Number.isSafeInteger(r.totalBytes)||r.totalBytes<0))throw Error('invalid_backup'); + addBackup(result.get(dsp.id),kind,rows.reduce((n,r)=>n+r.totalBytes,0),v.createdAt); + } + }else if(result.has(v.dspId)){ + if(!Array.isArray(v.roots)||!/^[a-f0-9]{64}$/.test(v.digest))throw Error('invalid_backup'); + addBackup(result.get(v.dspId),kind,await size(path.join(root,name),budget,true),record.at); + } + }}catch{for(const value of result.values())value.available=false;} + } + return result; +} +async function measureDsp(paths,id,backups,budget,volumeCheck){ + if(!DSP.test(id))throw Error('invalid_dsp'); + const root=path.join(paths.dsps,id);await directory(root); + const volume=volumeCheck(root),usage=volume?await fsp.statfs(path.join(root,'data')):null; + const runtimeBytes=await size(path.join(root,'runtime'),budget); + const areas=['config','data','secrets','state','run','staging','logs','backups','browser','plugins']; + const breakdown={}; + for(const name of areas)breakdown[name]=await size(path.join(root,name),budget); + const revisions=path.join(root,'backups/plugin-revisions'); + for(const plugin of await names(revisions)){ + if(!/^[a-z][a-z0-9-]{0,63}$/.test(plugin))continue; + for(const revision of await names(path.join(revisions,plugin))){ + if(!/^[1-9]\d*$/.test(revision))continue; + const record=await json(path.join(revisions,plugin,revision,'snapshot.json'));if(!record)continue; + const value=record.value; + if(value.schemaVersion!==1||value.pluginId!==plugin||String(value.revision)!==revision||!Array.isArray(value.files))throw Error('invalid_snapshot'); + if(value.files.some(f=>!Number.isSafeInteger(f.size)||f.size<0))throw Error('invalid_snapshot'); + addBackup(backups,'plugins',value.files.reduce((n,f)=>n+f.size,0),record.at); + } + } + const usedBytes=usage?(usage.blocks-usage.bfree)*usage.bsize:Object.values(breakdown).reduce((n,v)=>n+v,0); + return {limited:Boolean(volume),capacityBytes:usage?usage.blocks*usage.bsize:null,availableBytes:usage?usage.bavail*usage.bsize:null, + usedBytes,runtimeBytes,dataBytes:breakdown.data,pluginBytes:breakdown.plugins,logBytes:breakdown.logs,localBackupBytes:breakdown.backups, + backups:{...backups,lastAt:backups.lastAt?new Date(backups.lastAt).toISOString():null}}; +} +function createStorageSampler({paths,clock=Date.now,intervalMs=60000,volumeCheck=assertVolumeMounted,maximumEntries=200000}={}){ + let running=null,lastStart=null;const cache=new Map(); + async function refresh(ids){ + const budget={entries:0,maximum:maximumEntries,deadline:Date.now()+30000}; + const index=await backupIndex(paths,ids,budget); + for(const id of ids){ + try{cache.set(id,{...await measureDsp(paths,id,index.get(id),budget,volumeCheck),sampledAt:clock(),status:'ready'});} + catch{const prior=cache.get(id);cache.set(id,{...prior,status:prior?.sampledAt?'stale':'unavailable'});} + } + for(const id of cache.keys())if(!ids.includes(id))cache.delete(id); + } + return {read(ids){ + if(!running&&(lastStart===null||clock()-lastStart>=intervalMs||ids.some(id=>!cache.has(id)))){ + lastStart=clock();running=refresh(ids).catch(()=>{for(const id of ids)cache.set(id,{...cache.get(id),status:cache.get(id)?.sampledAt?'stale':'unavailable'});}).finally(()=>{running=null;}); + } + return new Map(ids.map(id=>[id,{...(cache.get(id)||{status:'measuring',sampledAt:null}),refreshing:Boolean(running)}])); + },settled:()=>running||Promise.resolve()}; +} +module.exports={createStorageSampler,size,backupIndex}; diff --git a/core/host/capacity/tests/resources.test.js b/core/host/capacity/tests/resources.test.js new file mode 100644 index 0000000..8b21550 --- /dev/null +++ b/core/host/capacity/tests/resources.test.js @@ -0,0 +1,44 @@ +'use strict'; +const test=require('node:test'),assert=require('node:assert/strict'),fs=require('node:fs'),path=require('node:path'),os=require('node:os'),crypto=require('node:crypto'); +const {DatabaseSync}=require('node:sqlite'); +const {createResourceSampler,workerGroups}=require('../resources'); +const {createDirectoryMonitor}=require('../monitor'); +const {unitName}=require('../../services/host'); +const workerName=require('../../services/scoped-worker').unitName; +const id='dsp_'+'a'.repeat(32),peer='dsp_'+'b'.repeat(32); +test('CPU uses monotonic deltas and resets on cgroup recreation, counter reset or disappearance',()=>{ + let now=0,value={identity:'one',cpuUsage:1000000,memoryBytes:500,tasks:2}; + const sample=createResourceSampler({read:()=>value,monotonic:()=>now}); + assert.equal(sample(['group']).get('group').cpuPercent,null); + now=2000;value.cpuUsage+=3000000;assert.equal(sample(['group']).get('group').cpuPercent,150); + now=4000;value.identity='two';assert.equal(sample(['group']).get('group').cpuPercent,null); + now=6000;value.cpuUsage=1;assert.equal(sample(['group']).get('group').cpuPercent,null); + value=null;assert.equal(sample(['group']).get('group'),null); + value={identity:'two',cpuUsage:50};now=8000;assert.equal(sample(['group']).get('group').cpuPercent,null); +}); +test('Core registry attributes plugin and authentication browser cgroups only to their DSP',t=>{ + const root=fs.mkdtempSync(path.join(os.tmpdir(),'resource-workers-'));t.after(()=>fs.rmSync(root,{recursive:true,force:true})); + const backend=path.join(root,'state/plugin-backend'),jobs=path.join(backend,'jobs');fs.mkdirSync(jobs,{recursive:true,mode:0o700}); + const jobId='job_'+'c'.repeat(32),browserId='browser_'+'d'.repeat(48); + fs.writeFileSync(path.join(jobs,jobId+'.json'),JSON.stringify({schemaVersion:1,jobId,dspId:id}),{mode:0o600}); + const db=new DatabaseSync(path.join(backend,'browsers.sqlite3')); + db.exec('CREATE TABLE browser_leases(id TEXT,dsp_id TEXT,state TEXT)'); + db.prepare('INSERT INTO browser_leases VALUES(?,?,?)').run(browserId,peer,'active');db.close(); + const view=workerGroups({local:root},[id,peer]);assert.equal(view.available,true); + assert.deepEqual([...view.groups.get(id)],[workerName(jobId)]); + assert.deepEqual([...view.groups.get(peer)],[workerName('job_'+crypto.createHash('sha256').update(browserId).digest('hex').slice(0,32))]); + fs.writeFileSync(path.join(jobs,jobId+'.json'),'invalid');assert.equal(workerGroups({local:root},[id]).available,false); +}); +test('monitor aggregates runtime and isolated workers, caches samples, handles sleeping and hides identities',()=>{ + let now=1000,calls=0,available=true; + const monitor=createDirectoryMonitor({clock:()=>now,paths:{},cgroupRoot:'/groups', + store:{db:{prepare:()=>({all:()=>[{runtime_key:id,name:'Dev',installation_status:'ready'},{runtime_key:peer,name:'Fleet',installation_status:'ready'}]})}}, + manager:{journal:{record:()=>({desiredState:'running'})},hub:{connected:()=>true}},execution:{store:{get:key=>({state:key===peer?'sleeping':'running'})}}, + disk:()=>{throw Error('unavailable');},storageSampler:{read:ids=>new Map(ids.map(id=>[id,{status:'measuring'}]))}, + readWorkers:()=>({available,groups:new Map([[id,new Set(['worker'])]])}), + sampleResources:()=>{calls++;return new Map([[path.join('/groups',unitName(id)),{memoryBytes:100,tasks:2,cpuPercent:10}],['/groups/worker',{memoryBytes:200,tasks:3,cpuPercent:50}]]);}}); + const first=monitor();assert.equal(first.runtimes[0].memoryBytes,300);assert.equal(first.runtimes[0].cpuPercent,60);assert.equal(first.runtimes[0].tasks,5);assert.equal(first.runtimes[0].activeWorkers,1); + assert.equal(first.runtimes[1].memoryBytes,0);assert.equal(first.runtimes[1].status,'sleeping');assert.equal(first.storageAvailableBytes,null); + assert.equal(JSON.stringify(first).includes(id),false);assert.equal(monitor(),first);assert.equal(calls,1); + now+=2000;available=false;assert.equal(monitor().runtimes[0].memoryBytes,null); +}); diff --git a/core/host/capacity/tests/storage.test.js b/core/host/capacity/tests/storage.test.js new file mode 100644 index 0000000..ca8024c --- /dev/null +++ b/core/host/capacity/tests/storage.test.js @@ -0,0 +1,27 @@ +'use strict'; +const test=require('node:test'),assert=require('node:assert/strict'),fs=require('node:fs'),path=require('node:path'),os=require('node:os'); +const {createStorageSampler}=require('../storage'); +const id='dsp_'+'a'.repeat(32),peer='dsp_'+'b'.repeat(32); +function setup(t){const root=fs.mkdtempSync(path.join(os.tmpdir(),'resource-storage-'));t.after(()=>fs.rmSync(root,{recursive:true,force:true})); + const paths={dsps:path.join(root,'dsps'),local:path.join(root,'local')}; + const write=(file,value)=>{fs.mkdirSync(path.dirname(file),{recursive:true});fs.writeFileSync(file,typeof value==='string'?value:JSON.stringify(value));}; + for(const dsp of [id,peer])write(path.join(paths.dsps,dsp,'data/example'),'synthetic');return {root,paths,write};} +test('cached storage counts completed DSP backups, excludes neighbors and symlinks, preserves stale measurements',async t=>{ + const {root,paths,write}=setup(t);let now=1000,fail=false; + const dsp=path.join(paths.dsps,id),manual='mbk_'+'c'.repeat(32),update='d'.repeat(32); + write(path.join(paths.local,'backups/manual',manual,'manifest.json'),{id:manual,version:2,createdAt:100,dsps:[{id}],roots:[{label:id+'_data',totalBytes:7}]}); + write(path.join(paths.local,'backups/updates/dsp',update,'snapshot.json'),{dspId:id,digest:'e'.repeat(64),roots:[]}); + write(path.join(paths.local,'backups/updates/dsp','f'.repeat(32),'partial'),'ignore'); + write(path.join(dsp,'backups/plugin-revisions/paycom/1/snapshot.json'),{schemaVersion:1,pluginId:'paycom',revision:1,files:[{size:9}]}); + write(path.join(root,'outside'),'x'.repeat(100000));fs.symlinkSync(path.join(root,'outside'),path.join(dsp,'data/link')); + const sampler=createStorageSampler({paths,clock:()=>now,volumeCheck:()=>{if(fail)throw Error('mount_missing');return null;}}); + assert.equal(sampler.read([id,peer]).get(id).status,'measuring');await sampler.settled(); + const first=sampler.read([id,peer]).get(id);assert.equal(first.status,'ready');assert.equal(first.backups.count,3);assert.equal(first.backups.manual,1);assert.equal(first.backups.updates,1);assert.equal(first.backups.plugins,1);assert.equal(sampler.read([id,peer]).get(peer).backups.count,0); + assert.equal(first.dataBytes,fs.statSync(path.join(dsp,'data/example')).blocks*512); + write(path.join(dsp,'data/new'),'x'.repeat(10000));assert.equal(sampler.read([id]).get(id).usedBytes,first.usedBytes); + now+=60000;fail=true;sampler.read([id]);await sampler.settled();const stale=sampler.read([id]).get(id);assert.equal(stale.status,'stale');assert.equal(stale.usedBytes,first.usedBytes);assert.equal(stale.sampledAt,1000); + now+=60000;fail=false;sampler.read([id]);await sampler.settled();assert.ok(sampler.read([id]).get(id).usedBytes>first.usedBytes); +}); +test('missing DSP storage and exhausted scan budgets remain unavailable rather than zero',async t=>{ + const {paths}=setup(t);const sampler=createStorageSampler({paths,maximumEntries:0,volumeCheck:()=>null});sampler.read([id]);await sampler.settled();const view=sampler.read([id]).get(id);assert.equal(view.status,'unavailable');assert.equal(view.usedBytes,undefined); +}); diff --git a/core/tooling/tests.json b/core/tooling/tests.json index e427712..a68ff69 100644 --- a/core/tooling/tests.json +++ b/core/tooling/tests.json @@ -1,5 +1,6 @@ { "unit": [ + "host/capacity/tests", "sdk/tests", "core/api/tests", "core/auth-broker/tests",