diff --git a/apps/control-station/src/components/system/TelemetrySeries.tsx b/apps/control-station/src/components/system/TelemetrySeries.tsx index 99d86fc..cdf5608 100644 --- a/apps/control-station/src/components/system/TelemetrySeries.tsx +++ b/apps/control-station/src/components/system/TelemetrySeries.tsx @@ -6,19 +6,17 @@ interface TelemetrySeriesProps { ceiling?: number; } -function linePoints(values: Array, ceiling?: number): string { - const finite = values.filter((value): value is number => - value !== null && Number.isFinite(value), - ); - if (finite.length === 0) return ""; - const maximum = Math.max(ceiling ?? 0, ...finite, 1); - const denominator = Math.max(1, values.length - 1); - return values.map((value, index) => { - const x = index / denominator * 100; - const normalized = value === null ? 0 : Math.max(0, Math.min(maximum, value)); - const y = 36 - normalized / maximum * 34; - return `${x.toFixed(2)},${y.toFixed(2)}`; - }).join(" "); +export function lineSegments(values: Array, ceiling?: number): string[] { + const finite = values.filter((value): value is number => value !== null && Number.isFinite(value)); + if (!finite.length) return []; + const maximum = Math.max(ceiling ?? 0, ...finite, 1), denominator = Math.max(1, values.length - 1); + const segments:string[]=[];let segment:string[]=[]; + values.forEach((value,index)=>{ + if(value===null||!Number.isFinite(value)){if(segment.length)segments.push(segment.join(' '));segment=[];return;} + segment.push(`${(index/denominator*100).toFixed(2)},${(36-Math.max(0,Math.min(maximum,value))/maximum*34).toFixed(2)}`); + }); + if(segment.length)segments.push(segment.join(' ')); + return segments; } export function TelemetrySeries({ @@ -28,7 +26,7 @@ export function TelemetrySeries({ resource, ceiling, }: TelemetrySeriesProps) { - const points = linePoints(values, ceiling); + const segments = lineSegments(values, ceiling); return (
@@ -45,7 +43,7 @@ export function TelemetrySeries({ aria-label={`${label}: ${value}`} > - {points ? : null} + {segments.map((points,index)=>)}
); diff --git a/apps/control-station/src/core/fleet/boardMonitor.ts b/apps/control-station/src/core/fleet/boardMonitor.ts new file mode 100644 index 0000000..aa73467 --- /dev/null +++ b/apps/control-station/src/core/fleet/boardMonitor.ts @@ -0,0 +1,49 @@ +import {useEffect,useState} from 'react'; + +export interface MetricDefinition {label:string;group:string;resource:string;unit:string;reason:string|null} +export interface BoardMonitor { + available:boolean;fresh:boolean;storage:string;source_id:string; + latest:{seq:number;at:number;boot_id:string;values:Record}|null; + definitions:Record;received_at:number;backlog:number; + database_bytes:number|null;database_budget_bytes:number;archive_since:number|null; + start:number;end:number;bucket_seconds:number; + series:{at:number;min:number|null;max:number|null;mean:number|null;count:number}[]; + events:{at:number;code:string;kind:string;locations:string[]}[]; +} +export function useBoardMonitor(vehicle:string,metric:string,window:number,end:number|null){ + const [value,setValue]=useState(null),[error,setError]=useState(''); + useEffect(()=>{ + let active=true;let pending:AbortController|undefined;let timer:ReturnType|undefined; + setValue(null);setError(''); + const read=async()=>{ + const controller=new AbortController();pending=controller; + const deadline=setTimeout(()=>controller.abort(),12000); + try { + const query=new URLSearchParams({metric,window:String(window),...(end===null?{}:{end:String(end)})}); + const response=await fetch(`/api/v1/fleet/${encodeURIComponent(vehicle)}/monitor?${query}`,{credentials:'same-origin',cache:'no-store',signal:controller.signal}); + if(!response.ok)throw new Error('Мониторинг БК недоступен.'); + const next=await response.json() as BoardMonitor; + if(active){setValue(next);setError('');} + }catch {if(active)setError('Не удалось обновить мониторинг БК.');} + finally {clearTimeout(deadline);if(active)timer=setTimeout(()=>void read(),5000);} + }; + void read();return()=>{active=false;pending?.abort();clearTimeout(timer);}; + },[vehicle,metric,window,end]); + return {value,error,dismissError:()=>setError('')}; +} +export function monitorSeries(value:BoardMonitor,mode:'min'|'max'|'mean') { + const count=Math.ceil((value.end-value.start)/value.bucket_seconds)+1; + const values:Array=Array(Math.min(count,302)).fill(null); + const first=Math.floor(value.start/value.bucket_seconds)*value.bucket_seconds; + for(const point of value.series){const index=Math.round((point.at-first)/value.bucket_seconds);if(index>=0&&index=1024&&index<3){value/=1024;index++;} + return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})} ${labels[index]}`; + } + return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})}${unit?' '+unit:''}`; +} diff --git a/apps/control-station/src/workspaces/fleet/BoardMonitorView.tsx b/apps/control-station/src/workspaces/fleet/BoardMonitorView.tsx new file mode 100644 index 0000000..2e1589d --- /dev/null +++ b/apps/control-station/src/workspaces/fleet/BoardMonitorView.tsx @@ -0,0 +1,45 @@ +import {useState} from 'react'; +import {ActivityIndicator,Button,Icon,ResourceList,ResourceRow,Select,SettingsCard,StatusBadge,TextField,ToastStack} from '@nodedc/ui-react'; +import {useBoardMonitor,monitorSeries,monitorValue} from '../../core/fleet/boardMonitor'; +import {TelemetrySeries} from '../../components/system/TelemetrySeries'; +import '../../styles/system-telemetry.css'; +import './board-monitor.css'; + +const eventLabels:Record={'ui-error':'Ошибка интерфейса','ui-rejection':'Ошибка асинхронной операции','ui-render-error':'Ошибка представления','system-oom':'Нехватка памяти','gpu-reset':'Сброс GPU','disk-error':'Ошибка диска','usb-disconnected':'Отключение USB','usb-error':'Ошибка USB','service-failed':'Сбой службы','ui-process-exit':'Завершение процесса окна','ui-load-failed':'Ошибка загрузки окна'}; +const time=(value:number)=>new Date(value*1000).toLocaleString('ru-RU'); +export function BoardMonitorView({vehicle,name,back}:{vehicle:string;name:string;back:()=>void}) { + const [metric,setMetric]=useState('cpu.usage'),[group,setGroup]=useState('CPU'),[window,setWindow]=useState(900),[end,setEnd]=useState(null),[mode,setMode]=useState<'min'|'max'|'mean'>('max'); + const {value,error,dismissError}=useBoardMonitor(vehicle,metric,window,end); + const definitions=value?.definitions??{},definition=definitions[metric]; + const groups=[...new Set(Object.values(definitions).map(item=>item.group))]; + const metrics=Object.entries(definitions).filter(([,item])=>item.group===group); + const latest=value?.latest; + return
+
+ {value?.fresh&&!error?'Данные поступают':'Нет свежих данных'}}> + {latest&&

Измерено на БК: {time(latest.at)}. {value?.storage==='ready'?'Локальная история записывается.':'Запись локальной истории недоступна.'} {value?.backlog?'История догружается в Mission Core.':''}

} + {value?.database_bytes!=null&&

Архив на БК: {monitorValue(value.database_bytes,'bytes')} из {monitorValue(value.database_budget_bytes,'bytes')}. {value.archive_since?`Копия в Mission Core с ${time(value.archive_since)}.`:''}

} + + {!value?:!value.available?:<> + +
+ ({value:key,label:`${item.resource} · ${item.label}`}))} onChange={setMetric}/> + setMode(next as typeof mode)}/> +
+
{end!==null&&{const next=new Date(event.target.value).getTime()/1000;if(Number.isFinite(next))setEnd(next);}}/>}
+ +

{time(value.start)} — {time(value.end)} · интервал {value.bucket_seconds} с. Разрывы означают отсутствие измерений.

+ {!value.series.length&&

За выбранный период измерений нет.

} + + + {metrics.map(([key,item])=>
    • )}     +     + + {value.events.length?{value.events.map((event,index)=>
    • )}    :

    Событий не получено.

    } +     + } + +
    ; +} diff --git a/apps/control-station/src/workspaces/fleet/VehiclesWorkspace.tsx b/apps/control-station/src/workspaces/fleet/VehiclesWorkspace.tsx index 1406b4e..b1a3787 100644 --- a/apps/control-station/src/workspaces/fleet/VehiclesWorkspace.tsx +++ b/apps/control-station/src/workspaces/fleet/VehiclesWorkspace.tsx @@ -1,3 +1,4 @@ +import {BoardMonitorView} from './BoardMonitorView'; import { useEffect, useRef, useState } from "react"; import { ActivityIndicator, Button, ConfirmationModal, Icon, IconButton, ResourceList, ResourceRow, Select, SettingsCard, StatusBadge, TextAreaField, TextField, Window, WindowFooterActions } from "@nodedc/ui-react"; import { fleetRequest, useFleet, type FleetPreview, type Vehicle } from "../../core/fleet/useFleet"; @@ -19,6 +20,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe const [error, setError] = useState(""); const [selected, setSelected] = useState(null); const [sensorOpen, setSensorOpen] = useState(false); + const [monitorOpen,setMonitorOpen]=useState(false); const [revoking, setRevoking] = useState(null); const lastCreateRequest = useRef(createRequest); useEffect(() => { if (createRequest !== lastCreateRequest.current) { lastCreateRequest.current = createRequest; setAdding(true); setError(""); } }, [createRequest]); @@ -40,6 +42,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe finally { setPending(false); } } const detail = fleet.items?.find(item => item.id === selected); + if(detail&&monitorOpen)return setMonitorOpen(false)}/>; return
    {fleet.error &&

    {fleet.error}

    } {!adding && error &&

    {error}

    } @@ -51,6 +54,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe
    Последняя связь
    {detail.last_seen ? new Date(detail.last_seen * 1000).toLocaleString("ru-RU") : "Соединение ещё не получено"}
    {detail.host && <>
    Имя БК в системе
    {detail.host.hostname}
    Операционная система
    {detail.host.os}
    Архитектура
    {detail.host.architecture}
    Процессоры
    {detail.host.cpus}
    Память
    {detail.host.memory_kib ? `${(detail.host.memory_kib / 1024 / 1024).toFixed(1)} ГиБ` : "Нет сведений"}
    } + {detail.enrollment !== "revoked" && } } diff --git a/apps/control-station/src/workspaces/fleet/board-monitor.css b/apps/control-station/src/workspaces/fleet/board-monitor.css new file mode 100644 index 0000000..eecd2ac --- /dev/null +++ b/apps/control-station/src/workspaces/fleet/board-monitor.css @@ -0,0 +1,3 @@ +.board-monitor { display: grid; gap: var(--nodedc-space-4); min-width: 0; } +.board-monitor-controls { display: grid; grid-template-columns: repeat(auto-fit, minmax(220px, 1fr)); gap: var(--nodedc-space-3); align-items: end; margin-bottom: var(--nodedc-space-3); } +.board-monitor p { font-size: var(--nodedc-font-size-sm); color: var(--nodedc-text-secondary); } diff --git a/apps/control-station/test/boardMonitor.test.mjs b/apps/control-station/test/boardMonitor.test.mjs new file mode 100644 index 0000000..0a508ea --- /dev/null +++ b/apps/control-station/test/boardMonitor.test.mjs @@ -0,0 +1,52 @@ +import assert from 'node:assert/strict'; +import {before,after,test} from 'node:test'; +import {createServer} from 'vite'; +let server,series,segments,frames,localPreview,MEDIA_PROTOCOL; +before(async()=>{ + server=await createServer({appType:'custom',logLevel:'silent',server:{middlewareMode:true}}); + ({monitorSeries:series}=await server.ssrLoadModule('/src/core/fleet/boardMonitor.ts')); + ({lineSegments:segments}=await server.ssrLoadModule('/src/components/system/TelemetrySeries.tsx')); + ({MEDIA_PROTOCOL}=await server.ssrLoadModule('@xgrids-k1/frontend/sensors/previewFrames.ts')); + ({localPreviewFrames:frames,localPreview}=await server.ssrLoadModule('@xgrids-k1/frontend/sensors/localPreview.ts')); +}); +after(async()=>{await server?.close();}); +test('archive gaps break the plot instead of manufacturing zero or connecting missing time',()=>{ + const values=series({start:0,end:20,bucket_seconds:5,series:[{at:0,max:80},{at:5,max:90},{at:20,max:10}]},'max'); + assert.deepEqual(values,[80,90,null,null,10]); + const result=segments(values,100); + assert.equal(result.length,2); + assert.match(result[0],/^0.00,8.80 25.00,5.40$/); + assert.deepEqual(segments([null,NaN,null]),[]); +}); +test('HTTP media framing survives every byte boundary and refuses oversized frames',()=>{ + const output=[],receive=frames((kind,bytes)=>output.push([kind,[...bytes]])); + const encoded=Uint8Array.of(2,0,0,0,3,4,5,6,3,0,0,0,2,7,8); + for(const byte of encoded)receive(Uint8Array.of(byte)); + assert.deepEqual(output,[[2,[4,5,6]],[3,[7,8]]]); + assert.throws(()=>frames(()=>{})(Uint8Array.of(2,0,1,0,0))); + assert.throws(()=>frames(()=>{})(Uint8Array.of(7,0,0,0,0))); +}); + +test('local carrier delivers registered channels and ACKs through completed responses',async()=>{ + const encode=(kind,data)=>{const bytes=typeof data==='string'?new TextEncoder().encode(data):Uint8Array.from(data);const out=new Uint8Array(bytes.length+5);out[0]=kind;new DataView(out.buffer).setUint32(1,bytes.length);out.set(bytes,5);return out;}; + const peer='peer_'+ 'a'.repeat(32), received=[], closed=[];let preview; + const device={id:'synthetic-k1',snapshot:{context:{session_id:'synthetic-session'}}}; + const transport={ + submit:async request=>{closed.push(request);return {state:'complete'};}, + localPreview:{ + open:async request=>{assert.equal(request.action_id,'offer');return new Response(encode(0,JSON.stringify({peer_id:peer,media_protocol:MEDIA_PROTOCOL})));}, + read:async(id,after)=>{ + assert.equal(id,peer); + if(after===0)return new Response(new Uint8Array([...encode(2,[8,9]),...encode(3,'camera-ready')])); + assert.equal(after,7);preview.close();return new Response(); + } + } + }; + preview=localPreview(device,transport,{},()=>assert.fail('carrier failed')); + preview.rrd.onmessage=event=>{received.push([...new Uint8Array(event.data)]);preview.rrd.send(JSON.stringify({ack:7}));}; + preview.camera.onmessage=event=>received.push(event.data); + await preview.start(); + assert.deepEqual(received,[[8,9],'camera-ready']); + assert.equal(preview.rrd.readyState,'closed');assert.equal(closed.length,1); + assert.equal(closed[0].action_id,'close-peer');assert.equal(closed[0].parameters.peer_id,peer); +}); diff --git a/apps/node-agent/cmd/node-agent/main.go b/apps/node-agent/cmd/node-agent/main.go index 0aa9ac7..c76dfeb 100644 --- a/apps/node-agent/cmd/node-agent/main.go +++ b/apps/node-agent/cmd/node-agent/main.go @@ -69,12 +69,14 @@ func run() error { return err } app := &node.Server{Store: store, Assets: assets, Origin: "http://" + *listen, Version: version, Inventory: func() node.Inventory { return node.Host("/") }} + app.Monitor = node.NewMonitor() app.Presentation = node.NewPresentationStore(*dir) pairing, err := node.OpenPairing(store, *dir, version, app.Inventory) if err != nil { return err } app.Pairing = pairing + pairing.Monitor = app.Monitor nodeID, _ := store.Public() app.Sensors, err = node.OpenSensors(*dir, nodeID) if err != nil { @@ -122,6 +124,7 @@ func run() error { go func() { errs <- private.Serve(unix) }() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() + go app.Monitor.Run(ctx) go app.Sensors.WatchUSB(ctx) go pairing.Run(ctx) log.Print("Mission Core Node " + version + " listening on loopback") diff --git a/apps/node-agent/internal/node/local_preview.go b/apps/node-agent/internal/node/local_preview.go new file mode 100644 index 0000000..0bcda6a --- /dev/null +++ b/apps/node-agent/internal/node/local_preview.go @@ -0,0 +1,73 @@ +package node + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "strings" + "time" +) + +// Local WebKit requires completed bounded HTTP responses. The same plugin +// delivery owns RRD/fMP4 framing, backpressure, resume and acquisition isolation. +func (s *Server) localPreviewRoutes(mux *http.ServeMux) { + capacity := make(chan struct{}, 2) + for _, route := range []string{"/api/devices/preview", "/api/devices/preview/read"} { + mux.HandleFunc("POST "+route, func(w http.ResponseWriter, r *http.Request) { + if !s.authorized(w, r) { + return + } + var value json.RawMessage + if !decode(w, r, &value) { + return + } + if s.DeviceEnrollment == nil { + http.Error(w, "Preview unavailable", 503) + return + } + path := "/local-preview/read" + if r.URL.Path == "/api/devices/preview" { + var command SensorCommand + if json.Unmarshal(value, &command) != nil || command.Action != "offer" || !sensorID.MatchString(command.Session.DeviceID) || !strings.HasPrefix(command.Session.DeviceID, "k1_") { + http.Error(w, "Invalid preview", 400) + return + } + path = "/local-preview" + } + select { + case capacity <- struct{}{}: + defer func() { <-capacity }() + default: + http.Error(w, "Close another viewer", 429) + return + } + request, err := http.NewRequestWithContext(r.Context(), "POST", "http://k1"+path, bytes.NewReader(value)) + if err != nil { + http.Error(w, "Invalid preview", 400) + return + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Node-Id", s.DeviceEnrollment.nodeID) + client := &http.Client{Transport: s.DeviceEnrollment.client.Transport, Timeout: 5 * time.Second} + response, err := client.Do(request) + if err != nil { + http.Error(w, "Preview unavailable", 503) + return + } + defer response.Body.Close() + if response.StatusCode != 200 { + http.Error(w, "Preview unavailable", response.StatusCode) + return + } + data, err := io.ReadAll(io.LimitReader(response.Body, 196609)) + if err != nil || len(data) > 196608 { + http.Error(w, "Preview unavailable", 503) + return + } + w.Header().Set("Content-Type", "application/vnd.missioncore.preview") + w.Header().Set("Cache-Control", "no-store") + _, _ = w.Write(data) + }) + } +} diff --git a/apps/node-agent/internal/node/monitor.go b/apps/node-agent/internal/node/monitor.go new file mode 100644 index 0000000..78684a9 --- /dev/null +++ b/apps/node-agent/internal/node/monitor.go @@ -0,0 +1,159 @@ +package node + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net" + "net/http" + "strconv" + "sync" + "time" +) + +// The collector owns sampling/storage. This bounded cache keeps monitoring I/O +// out of the paired control heartbeat and out of acquisition lifecycles. +type Monitor struct { + mu sync.Mutex + client *http.Client + source string + after int64 + sent int64 + value map[string]any +} + +func NewMonitor() *Monitor { + return &Monitor{client: &http.Client{Timeout: 3 * time.Second, Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, "unix", "/run/mission-core-monitor/monitor.sock") + }}}} +} +func (m *Monitor) read(ctx context.Context, path string, body []byte) (map[string]any, error) { + method := "GET" + if body != nil { + method = "POST" + } + r, e := http.NewRequestWithContext(ctx, method, "http://monitor"+path, bytes.NewReader(body)) + if e != nil { + return nil, e + } + r.Header.Set("Content-Type", "application/json") + response, e := m.client.Do(r) + if e != nil { + return nil, e + } + defer response.Body.Close() + if response.StatusCode != 200 { + return nil, errors.New("monitor unavailable") + } + data, e := io.ReadAll(io.LimitReader(response.Body, 262145)) + if e != nil || len(data) > 262144 { + return nil, errors.New("monitor response budget") + } + var value map[string]any + e = json.Unmarshal(data, &value) + return value, e +} +func (m *Monitor) Run(ctx context.Context) { + defer m.client.CloseIdleConnections() + timer := time.NewTicker(2 * time.Second) + defer timer.Stop() + for { + status, e := m.read(ctx, "/status", nil) + if e == nil { + source, _ := status["source_id"].(string) + m.mu.Lock() + if source != m.source { + m.source = source + m.after = 0 + m.sent = 0 + } + after := m.after + m.mu.Unlock() + batch, be := m.read(ctx, "/batch?after="+strconv.FormatInt(after, 10), nil) + if be == nil && batch["source_id"] == source { + status["batch"] = batch + } + m.mu.Lock() + m.value = status + m.mu.Unlock() + } + select { + case <-ctx.Done(): + return + case <-timer.C: + } + } +} +func (m *Monitor) Snapshot() map[string]any { + m.mu.Lock() + defer m.mu.Unlock() + if m.value == nil { + return map[string]any{"storage": "unavailable"} + } + if batch, ok := m.value["batch"].(map[string]any); ok { + if rows, ok := batch["samples"].([]any); ok { + for _, raw := range rows { + if row, ok := raw.(map[string]any); ok { + if seq, ok := row["seq"].(float64); ok && int64(seq) > m.sent { + m.sent = int64(seq) + } + } + } + } + } + return m.value +} +func (m *Monitor) Acknowledge(raw json.RawMessage) { + var ack struct { + Source string `json:"source_id"` + After int64 `json:"after"` + } + if json.Unmarshal(raw, &ack) != nil { + return + } + m.mu.Lock() + defer m.mu.Unlock() + // A receiver may already have a later cursor from before the Node restarted. + // Its authenticated ACK is bounded by the collector's durable latest record. + latest := m.sent + if row, ok := m.value["latest"].(map[string]any); ok { + if seq, ok := row["seq"].(float64); ok { + latest = int64(seq) + } + } + if ack.Source == m.source && ack.After >= m.after && ack.After <= latest { + m.after = ack.After + } +} +func (s *Server) monitorRoutes(mux *http.ServeMux) { + mux.HandleFunc("GET /api/monitor", func(w http.ResponseWriter, r *http.Request) { + if !s.authorized(w, r) { + return + } + if s.Monitor == nil { + reply(w, 503, map[string]string{"error": "Мониторинг недоступен"}) + return + } + reply(w, 200, s.Monitor.Snapshot()) + }) + mux.HandleFunc("POST /api/monitor/events", func(w http.ResponseWriter, r *http.Request) { + if !s.authorized(w, r) { + return + } + var value json.RawMessage + if !decode(w, r, &value) { + return + } + if len(value) > 2048 || s.Monitor == nil { + http.Error(w, "Monitoring unavailable", 503) + return + } + if _, e := s.Monitor.read(r.Context(), "/events", value); e != nil { + http.Error(w, "Monitoring unavailable", 503) + return + } + reply(w, 200, map[string]bool{"ok": true}) + }) +} diff --git a/apps/node-agent/internal/node/monitor_test.go b/apps/node-agent/internal/node/monitor_test.go new file mode 100644 index 0000000..433eac1 --- /dev/null +++ b/apps/node-agent/internal/node/monitor_test.go @@ -0,0 +1,50 @@ +package node + +import ( + "encoding/json" + "testing" +) + +func TestMonitorAcknowledgementsRequireSameArchiveAndDurableBounds(t *testing.T) { + m := NewMonitor() + m.source = "archive-one" + m.value = map[string]any{"latest": map[string]any{"seq": float64(100)}} + for _, raw := range []string{`{"source_id":"another","after":50}`, `{"source_id":"archive-one","after":101}`, `{"source_id":"archive-one","after":-1}`} { + m.Acknowledge(json.RawMessage(raw)) + if m.after != 0 { + t.Fatal("accepted invalid cursor") + } + } + m.Acknowledge(json.RawMessage(`{"source_id":"archive-one","after":80}`)) + if m.after != 80 { + t.Fatal("did not restore committed receiver cursor after restart") + } + m.Acknowledge(json.RawMessage(`{"source_id":"archive-one","after":10}`)) + if m.after != 80 { + t.Fatal("regressed cursor") + } +} + +func TestMonitorAndPreviewCannotBypassNodeSession(t *testing.T) { + s := newTestServer(t) + for _, path := range []string{"/api/devices/preview", "/api/devices/preview/read", "/api/monitor/events"} { + if got := call(s, "POST", path, `{}`, nil).Code; got != 401 { + t.Fatalf("%s: %d", path, got) + } + } + if got := call(s, "GET", "/api/monitor", "", nil).Code; got != 401 { + t.Fatal(got) + } + cookie := login(t, s) + if got := call(s, "POST", "/api/devices/preview", `{}`, cookie).Code; got != 503 { + t.Fatal(got) + } + s.Monitor = NewMonitor() + if got := call(s, "GET", "/api/monitor", "", cookie).Code; got != 200 { + t.Fatal(got) + } + // An unavailable collector remains independent from the authenticated UI. + if got := call(s, "GET", "/api/status", "", cookie).Code; got != 200 { + t.Fatal(got) + } +} diff --git a/apps/node-agent/internal/node/pairing.go b/apps/node-agent/internal/node/pairing.go index e490a7f..d042fbd 100644 --- a/apps/node-agent/internal/node/pairing.go +++ b/apps/node-agent/internal/node/pairing.go @@ -39,6 +39,7 @@ type PairState struct { Revocations []CoreBinding `json:"revocations,omitempty"` } type Pairing struct { + Monitor *Monitor Sensors *Sensors DeviceEnrollment *DeviceEnrollment mu sync.Mutex diff --git a/apps/node-agent/internal/node/pairing_transport.go b/apps/node-agent/internal/node/pairing_transport.go index f1a09bf..9e7874d 100644 --- a/apps/node-agent/internal/node/pairing_transport.go +++ b/apps/node-agent/internal/node/pairing_transport.go @@ -286,10 +286,16 @@ func (p *Pairing) channel(ctx context.Context) { payload["device_enrollment"] = p.DeviceEnrollment.Status() payload["enrollment_results"] = p.DeviceEnrollment.RemoteResults() } + if p.Monitor != nil { + payload["monitor"] = p.Monitor.Snapshot() + } result, status, e := p.send(ctx, *binding, "/v1/node/heartbeat", payload) p.mu.Lock() if p.state.Phase == "paired" && p.state.Binding != nil && p.state.Binding.BindingID == binding.BindingID { if e == nil && status == 200 { + if p.Monitor != nil { + p.Monitor.Acknowledge(result["monitor_ack"]) + } p.connection = "online" p.lastSeen = p.now().Unix() if p.DeviceEnrollment != nil { diff --git a/apps/node-agent/internal/node/server.go b/apps/node-agent/internal/node/server.go index 8ff2d31..27201eb 100644 --- a/apps/node-agent/internal/node/server.go +++ b/apps/node-agent/internal/node/server.go @@ -13,6 +13,7 @@ import ( ) type Server struct { + Monitor *Monitor Store *Store Pairing *Pairing Sensors *Sensors @@ -81,6 +82,8 @@ func reply(w http.ResponseWriter, status int, v any) { func (s *Server) Handler() http.Handler { mux := http.NewServeMux() + s.localPreviewRoutes(mux) + s.monitorRoutes(mux) if s.Presentation != nil { s.presentationRoutes(mux) } diff --git a/apps/node-agent/monitor/collector.py b/apps/node-agent/monitor/collector.py new file mode 100644 index 0000000..068601c --- /dev/null +++ b/apps/node-agent/monitor/collector.py @@ -0,0 +1,201 @@ +#!/usr/bin/python3 +"""Independent one-second collector, local archive and private Node read adapter.""" + +import json +import logging +import os +import re +import signal +import socketserver +import threading +import time +from collections import deque +from http.server import BaseHTTPRequestHandler +from pathlib import Path +from urllib.parse import parse_qs, urlsplit + +from linux_metrics import LinuxMetrics +from storage import SCHEMA, Archive + +SOCKET = Path("/run/mission-core-monitor/monitor.sock") + + +class Collector: + def __init__(self): + self.stop = threading.Event() + self.lock = threading.Lock() + self.latest = None + self.definitions = {} + self.events = deque(maxlen=64) + self.storage = "starting" + self.database_bytes = None + self.archive = None + self.boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip() + + def event(self, value): + if ( + not isinstance(value, dict) + or set(value) - {"code", "kind", "locations"} + or value.get("code") not in {"ui-error", "ui-rejection", "ui-render-error"} + or not isinstance(value.get("kind"), str) + or not re.fullmatch("[A-Za-z]{1,48}", value["kind"]) + or not isinstance(value.get("locations", []), list) + or len(value.get("locations", [])) > 8 + or any( + not isinstance(v, str) + or not re.fullmatch(r"[A-Za-z0-9_-]{1,96}\.js:[0-9]{1,9}:[0-9]{1,9}", v) + for v in value.get("locations", []) + ) + ): + raise ValueError("invalid-event") + with self.lock: + if len(self.events) < 64: + self.events.append({**value, "at": time.time()}) + + def run(self): + metrics = LinuxMetrics() + maintenance = 0 + while not self.stop.is_set(): + started = time.monotonic() + try: + values = metrics.sample() + with self.lock: + events = list(self.events) + sample = dict( + at=time.time(), boot_id=self.boot, uptime=started, values=values, events=events + ) + if self.archive is None: + self.archive = Archive() + # Maintenance also runs while writes are paused by a disk budget. + if started - maintenance > 60: + self.database_bytes = self.archive.maintain() + maintenance = started + record = self.archive.append(sample, metrics.definitions) + with self.lock: + for _ in events: + self.events.popleft() + self.latest = record + self.definitions = metrics.definitions + self.storage = "ready" + except Exception as error: + logging.warning("monitor sample failed class=%s", type(error).__name__) + with self.lock: + self.storage = ( + str(error) + if str(error) in {"disk-reserve", "archive-budget"} + else "unavailable" + ) + self.stop.wait(max(0.05, 1 - (time.monotonic() - started))) + + def status(self): + with self.lock: + return dict( + schema=SCHEMA, + source_id=self.archive.source if self.archive else None, + latest=self.latest, + definitions=self.definitions, + storage=self.storage, + database_bytes=self.database_bytes, + sample_interval_seconds=1, + retention_days=7, + database_budget_bytes=2 * 1024**3, + free_reserve_bytes=2 * 1024**3, + ) + + +class Server(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): + daemon_threads = True + block_on_close = False + + def __init__(self, collector): + self.collector = collector + self.capacity = threading.BoundedSemaphore(6) + super().__init__(str(SOCKET), Handler) + + def process_request(self, request, address): + if not self.capacity.acquire(False): + request.close() + return + request.settimeout(5) + super().process_request(request, address) + + def process_request_thread(self, request, address): + try: + super().process_request_thread(request, address) + finally: + self.capacity.release() + + +class Handler(BaseHTTPRequestHandler): + def log_message(self, *_): + pass + + def reply(self, value, status=200): + payload = json.dumps(value, allow_nan=False, separators=(",", ":")).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + + def do_GET(self): + path = urlsplit(self.path) + try: + if path.path == "/status": + self.reply(self.server.collector.status()) + elif path.path == "/batch": + after = int(parse_qs(path.query).get("after", ["0"])[0]) + if not 0 <= after <= 2**53 - 1: + raise ValueError() + archive = self.server.collector.archive + if archive is None: + self.reply({"error": "Archive unavailable"}, 503) + else: + self.reply(archive.batch(after)) + else: + self.reply({"error": "Unknown request"}, 404) + except Exception: + self.reply({"error": "Archive unavailable"}, 503) + + def do_POST(self): + try: + size = int(self.headers.get("Content-Length", "0")) + if self.path != "/events" or not 0 < size <= 2048: + raise ValueError() + self.server.collector.event(json.loads(self.rfile.read(size))) + self.reply({"ok": True}) + except (ValueError, TypeError): + self.reply({"error": "Invalid event"}, 400) + + +def main(): + logging.basicConfig(level=logging.WARNING) + os.umask(0o007) + collector = Collector() + from journal_events import watch + + threading.Thread( + target=watch, args=(collector,), name="node-system-events", daemon=True + ).start() + SOCKET.unlink(missing_ok=True) + server = Server(collector) + worker = threading.Thread(target=collector.run, name="node-system-sampler", daemon=True) + worker.start() + + def stop(*_): + collector.stop.set() + threading.Thread(target=server.shutdown, daemon=True).start() + + signal.signal(signal.SIGTERM, stop) + signal.signal(signal.SIGINT, stop) + try: + server.serve_forever(poll_interval=0.5) + finally: + collector.stop.set() + server.server_close() + worker.join(5) + SOCKET.unlink(missing_ok=True) + + +if __name__ == "__main__": + main() diff --git a/apps/node-agent/monitor/journal_events.py b/apps/node-agent/monitor/journal_events.py new file mode 100644 index 0000000..62f1f88 --- /dev/null +++ b/apps/node-agent/monitor/journal_events.py @@ -0,0 +1,103 @@ +"""Classify bounded journal records; never persist free-form messages or payloads.""" + +import json +import re +import subprocess +import threading + +RULES = ( + ("system-oom", r"out of memory|oom-kill|killed process"), + ("gpu-reset", r"GPU HANG|GPU reset|Resetting.*(gpu|chip)|i915.*(hang|reset|error)"), + ("disk-error", r"I/O error|EXT4-fs error|nvme.*(timeout|reset)|ata.*failed command"), + ("usb-disconnected", r"USB disconnect"), + ("usb-error", r"usb.*(error -|unable to enumerate|reset.*device)"), + ("service-failed", r"Failed with result|Main process exited|Failed to start"), + ("ui-process-exit", r"node-ui-process-terminated"), + ("ui-load-failed", r"node-ui-load-failed"), +) + + +def classify(record): + message = record.get("MESSAGE") + if not isinstance(message, str): + return None + code = next((code for code, pattern in RULES if re.search(pattern, message, re.I)), None) + if code is None: + return None + try: + at = int(record["__REALTIME_TIMESTAMP"]) / 1_000_000 + except (KeyError, ValueError, TypeError): + return None + return dict( + code=code, + kind="Kernel" if record.get("_TRANSPORT") == "kernel" else "System", + at=at, + locations=[], + ) + + +def watch(collector): + args = [ + "/usr/bin/journalctl", + "--follow", + "--lines=0", + "--output=json", + "--no-pager", + "_TRANSPORT=kernel", + "+", + "SYSLOG_IDENTIFIER=mission-core-node-ui", + "+", + "_SYSTEMD_UNIT=mission-core-node.service", + "+", + "_SYSTEMD_UNIT=mission-core-k1.service", + "+", + "_SYSTEMD_UNIT=mission-core-node-monitor.service", + "+", + "_SYSTEMD_UNIT=postgresql@16-ndc-monitor.service", + ] + for unit in ( + "mission-core-node.service", + "mission-core-k1.service", + "mission-core-node-monitor.service", + "postgresql@16-ndc-monitor.service", + ): + args.extend(["+", "UNIT=" + unit]) + while not collector.stop.is_set(): + process = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) + + def retire(process=process): + while process.poll() is None: + if collector.stop.wait(1): + if process.poll() is None: + process.terminate() + return + + guard = threading.Thread(target=retire, daemon=True) + guard.start() + try: + while not collector.stop.is_set(): + line = process.stdout.readline(65537) + if not line: + break + if len(line) > 65536: + while line and not line.endswith(b"\n"): + line = process.stdout.readline(65537) + continue + try: + event = classify(json.loads(line)) + except (ValueError, TypeError): + continue + if event: + with collector.lock: + if len(collector.events) < 64: + collector.events.append(event) + finally: + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + process.stdout.close() + collector.stop.wait(10) diff --git a/apps/node-agent/monitor/linux_metrics.py b/apps/node-agent/monitor/linux_metrics.py new file mode 100644 index 0000000..d3efecb --- /dev/null +++ b/apps/node-agent/monitor/linux_metrics.py @@ -0,0 +1,353 @@ +"""Read-only Linux counters. Missing or reset counters never manufacture zero.""" + +import os +import re +import time +from pathlib import Path + + +class LinuxMetrics: + def __init__(self, root=Path("/"), clock=time.monotonic): + self.root, self.clock = Path(root), clock + self.previous = {} + self.previous_time = None + self.definitions = {} + + def text(self, path): + try: + return (self.root / str(path).lstrip("/")).read_text()[:262144].strip() + except (OSError, UnicodeError): + return None + + def number(self, path): + try: + return float(self.text(path)) + except (ValueError, TypeError): + return None + + def paths(self, pattern): + return sorted(self.root.glob(pattern))[:128] + + def sample(self): + now = self.clock() + elapsed = None if self.previous_time is None else now - self.previous_time + self.previous_time = now + values, definitions, counters = {}, {}, {} + + def metric( + key, label, group, resource, unit, value, reason="Счётчик недоступен в этой системе" + ): + values[key] = None if value is None else round(value, 3) + definitions[key] = dict( + label=label, + group=group, + resource=resource, + unit=unit, + reason=reason if value is None else None, + ) + + def delta(key, value): + if value is None: + return None + counters[key] = value + previous = self.previous.get(key) + return ( + None + if previous is None or value < previous or not elapsed or elapsed > 10 + else (value - previous) / elapsed + ) + + stat = self.text("proc/stat") or "" + cpus = 0 + for line in stat.splitlines(): + parts = line.split() + if not re.fullmatch(r"cpu\d*", parts[0]): + continue + if parts[0] != "cpu": + cpus += 1 + data = [int(v) for v in parts[1:9]] + total = delta(parts[0] + ".total", sum(data)) + idle = delta(parts[0] + ".idle", data[3] + data[4]) + wait = delta(parts[0] + ".iowait", data[4]) + for name, label, value in [ + ( + "usage", + "Загрузка", + None if not total or idle is None else 100 * (1 - idle / total), + ), + ( + "iowait", + "Ожидание диска", + None if not total or wait is None else 100 * wait / total, + ), + ]: + metric( + parts[0] + "." + name, + label, + "CPU", + parts[0], + "%", + value, + "Ожидаем два последовательных измерения", + ) + if not cpus: + metric("cpu.usage", "Загрузка", "CPU", "CPU", "%", None) + memory = {} + for line in (self.text("proc/meminfo") or "").splitlines(): + parts = line.replace(":", "").split() + if len(parts) > 1: + memory[parts[0]] = int(parts[1]) * 1024 + for key, label in [ + ("MemTotal", "Всего"), + ("MemAvailable", "Доступно"), + ("SwapTotal", "Swap всего"), + ("SwapFree", "Swap свободно"), + ("Dirty", "Ожидает записи"), + ("Writeback", "Записывается"), + ]: + metric("memory." + key, label, "RAM", "Память", "bytes", memory.get(key)) + total, available = memory.get("MemTotal"), memory.get("MemAvailable") + metric( + "memory.usage", + "Использовано", + "RAM", + "Память", + "%", + None if not total or available is None else 100 * (1 - available / total), + ) + for index, load in enumerate((self.text("proc/loadavg") or "").split()[:3]): + metric( + "load." + str(index), + ["За 1 минуту", "За 5 минут", "За 15 минут"][index], + "CPU", + "Load average", + "", + float(load), + ) + for resource in ["cpu", "memory", "io"]: + for line in (self.text("proc/pressure/" + resource) or "").splitlines(): + parts = line.split() + fields = dict(v.split("=") for v in parts[1:]) + metric( + "pressure." + resource + "." + parts[0], + "Задержка за 10 секунд", + "Ожидание ресурсов", + resource + " · " + parts[0], + "%", + float(fields["avg10"]), + ) + for interface in self.paths("sys/class/net/*"): + name = interface.name + for field, label in [ + ("rx_bytes", "Приём"), + ("tx_bytes", "Передача"), + ("rx_errors", "Ошибки приёма"), + ("tx_errors", "Ошибки передачи"), + ("rx_dropped", "Потери приёма"), + ("tx_dropped", "Потери передачи"), + ]: + raw = self.number(interface.relative_to(self.root) / "statistics" / field) + metric( + "net." + name + "." + field, + label, + "Сеть", + name, + "bytes/s" if field.endswith("bytes") else "1/s", + delta("net." + name + "." + field, raw), + ) + speed = self.number(interface.relative_to(self.root) / "speed") + metric( + "net." + name + ".speed", + "Скорость соединения", + "Сеть", + name, + "Mbit/s", + speed if speed is not None and speed > 0 else None, + ) + diskstats = (self.text("proc/diskstats") or "").splitlines() + for line in diskstats: + p = line.split() + if ( + len(p) < 14 + or p[2].startswith(("loop", "ram")) + or not (self.root / "sys/block" / p[2]).exists() + ): + continue + name = p[2] + for index, label, unit, multiplier in [ + (5, "Чтение", "bytes/s", 512), + (9, "Запись", "bytes/s", 512), + (3, "Операции чтения", "1/s", 1), + (7, "Операции записи", "1/s", 1), + (12, "Занятость", "%", 0.1), + ]: + key = "disk." + name + "." + str(index) + rate = delta(key, int(p[index])) + metric(key, label, "Диски", name, unit, None if rate is None else rate * multiplier) + if self.root == Path("/"): + mounts = set() + for line in (self.text("proc/mounts") or "").splitlines(): + p = line.split() + if len(p) < 3 or not p[0].startswith("/dev/") or p[1] in mounts: + continue + mount = p[1].replace("\\040", " ") + mounts.add(mount) + try: + st = os.statvfs(mount) + key = "fs." + str(len(mounts)) + metric( + key + ".free", + "Свободно", + "Хранилище", + mount, + "bytes", + st.f_bavail * st.f_frsize, + ) + metric( + key + ".usage", + "Использовано", + "Хранилище", + mount, + "%", + 100 * (1 - st.f_bfree / st.f_blocks) if st.f_blocks else None, + ) + metric( + key + ".inodes", + "Использовано inode", + "Хранилище", + mount, + "%", + 100 * (1 - st.f_ffree / st.f_files) if st.f_files else None, + ) + except OSError: + metric( + "fs." + str(len(mounts)) + ".free", + "Свободно", + "Хранилище", + mount, + "bytes", + None, + ) + for path in self.paths("sys/class/hwmon/hwmon*/temp*_input"): + relative = path.relative_to(self.root) + name = self.text(path.parent.relative_to(self.root) / "name") or path.parent.name + raw = self.number(relative) + metric( + "temperature." + path.parent.name + "." + path.stem, + "Температура", + "Температура", + name + " · " + path.stem, + "°C", + None if raw is None else raw / 1000, + ) + for card in self.paths("sys/class/drm/card[0-9]"): + base = card.relative_to(self.root) + metric( + "gpu." + card.name + ".usage", + "Загрузка", + "GPU", + card.name, + "%", + self.number(base / "device/gpu_busy_percent"), + "Драйвер не публикует общий счётчик загрузки GPU", + ) + metric( + "gpu." + card.name + ".frequency", + "Частота", + "GPU", + card.name, + "MHz", + self.number(base / "gt_cur_freq_mhz"), + ) + for field, label in [ + ("mem_info_vram_used", "Видеопамять занята"), + ("mem_info_vram_total", "Видеопамять всего"), + ]: + metric( + "gpu." + card.name + "." + field, + label, + "GPU", + card.name, + "bytes", + self.number(base / "device" / field), + "Выделенная видеопамять или её счётчик недоступны", + ) + for usb in self.paths("sys/bus/usb/devices/*"): + if not (usb / "idVendor").exists(): + continue + base = usb.relative_to(self.root) + name = usb.name + metric( + "usb." + name + ".speed", + "Скорость соединения", + "USB", + name, + "Mbit/s", + self.number(base / "speed"), + ) + metric( + "usb." + name + ".traffic", + "Трафик", + "USB", + name, + "bytes/s", + None, + "USB-драйвер не публикует счётчик трафика порта", + ) + groups = self.paths("sys/fs/cgroup/system.slice/mission-core*.service") + groups += self.paths( + "sys/fs/cgroup/system.slice/system-postgresql.slice/postgresql@16-ndc-monitor.service" + ) + groups += self.paths( + "sys/fs/cgroup/user.slice/user-*.slice/user@*.service/app.slice/app-gnome-org.nodedc.MissionCoreNode-*.scope" + ) + for path in groups: + base = path.relative_to(self.root) + name = path.name + key = "service." + re.sub("[^A-Za-z0-9_.-]", "_", name) + metric( + key + ".memory", + "Память", + "Приложения", + name, + "bytes", + self.number(base / "memory.current"), + ) + metric( + key + ".peak", + "Пик памяти", + "Приложения", + name, + "bytes", + self.number(base / "memory.peak"), + ) + fields = dict( + line.split() for line in (self.text(base / "cpu.stat") or "").splitlines() + ) + rate = delta( + key + ".cpu", float(fields["usage_usec"]) if "usage_usec" in fields else None + ) + metric( + key + ".cpu", + "Загрузка CPU", + "Приложения", + name, + "%", + None if rate is None else rate / 10000 / max(cpus, 1), + ) + fields = dict( + line.split() for line in (self.text(base / "memory.events") or "").splitlines() + ) + metric( + key + ".oom", + "Убийств из-за нехватки памяти", + "Приложения", + name, + "", + float(fields["oom_kill"]) if "oom_kill" in fields else None, + ) + self.previous = counters + # A bounded hardware inventory protects the control-plane payload budget. + keys = list(values)[:384] + self.definitions = {key: definitions[key] for key in keys} + return {key: values[key] for key in keys} diff --git a/apps/node-agent/monitor/schema.sql b/apps/node-agent/monitor/schema.sql new file mode 100644 index 0000000..923ea9a --- /dev/null +++ b/apps/node-agent/monitor/schema.sql @@ -0,0 +1,13 @@ +CREATE EXTENSION IF NOT EXISTS timescaledb; +CREATE TABLE IF NOT EXISTS monitor_identity (singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton), source_id text NOT NULL); +INSERT INTO monitor_identity(singleton,source_id) VALUES(true,gen_random_uuid()::text) ON CONFLICT DO NOTHING; +CREATE TABLE IF NOT EXISTS monitor_samples ( + seq bigint GENERATED ALWAYS AS IDENTITY, + observed_at timestamptz NOT NULL, + payload jsonb NOT NULL, + PRIMARY KEY(observed_at,seq) +); +SELECT create_hypertable('monitor_samples',by_range('observed_at',INTERVAL '1 hour'),if_not_exists=>true); +CREATE INDEX IF NOT EXISTS monitor_samples_cursor ON monitor_samples(seq); +CREATE TABLE IF NOT EXISTS monitor_definitions(singleton boolean PRIMARY KEY DEFAULT true CHECK(singleton),payload jsonb NOT NULL); +REVOKE ALL ON DATABASE mission_core_monitor FROM PUBLIC; diff --git a/apps/node-agent/monitor/storage.py b/apps/node-agent/monitor/storage.py new file mode 100644 index 0000000..f6440d8 --- /dev/null +++ b/apps/node-agent/monitor/storage.py @@ -0,0 +1,118 @@ +"""Bounded local Timescale archive; transport acknowledgements never delete data.""" + +import json +import shutil +from contextlib import contextmanager +from datetime import UTC, datetime + +SCHEMA = "missioncore.node-system-monitor/v1" +MAX_BATCH_BYTES = 196608 +DATABASE_BUDGET = 2 * 1024**3 + + +class Archive: + def __init__(self): + from psycopg2.pool import ThreadedConnectionPool + + self.pool = ThreadedConnectionPool( + 1, + 4, + host="/run/mission-core-monitor-db", + port=5433, + dbname="mission_core_monitor", + user="mission-core-monitor", + connect_timeout=3, + options="-c statement_timeout=3000", + ) + try: + with self.connection() as db, db.cursor() as cursor: + cursor.execute("SELECT source_id FROM monitor_identity") + self.source = cursor.fetchone()[0] + except Exception: + self.pool.closeall() + raise + self.definitions = None + self.database_bytes = None + + @contextmanager + def connection(self): + db = self.pool.getconn() + try: + yield db + db.commit() + except BaseException: + db.rollback() + raise + finally: + self.pool.putconn(db) + + def append(self, sample, definitions): + # This reserve belongs to telemetry, never to acquisition storage. + if shutil.disk_usage("/var/lib/mission-core-monitor").free < 2 * 1024**3: + raise RuntimeError("disk-reserve") + if self.database_bytes is None or self.database_bytes >= DATABASE_BUDGET: + raise RuntimeError("archive-budget") + payload = json.dumps(sample, separators=(",", ":"), allow_nan=False) + if len(payload.encode()) > 32768: + raise ValueError("sample-budget") + with self.connection() as db, db.cursor() as cursor: + if definitions != self.definitions: + cursor.execute( + "INSERT INTO monitor_definitions VALUES(true,%s) ON CONFLICT(singleton) " + "DO UPDATE SET payload=excluded.payload", + (json.dumps(definitions),), + ) + cursor.execute( + "INSERT INTO monitor_samples(observed_at,payload) VALUES(%s,%s) RETURNING seq", + (datetime.fromtimestamp(sample["at"], UTC), payload), + ) + seq = cursor.fetchone()[0] + self.definitions = definitions + return {**sample, "seq": seq} + + def batch(self, after): + with self.connection() as db, db.cursor() as cursor: + cursor.execute( + "SELECT seq,payload FROM monitor_samples WHERE seq>%s ORDER BY seq LIMIT 32", + (after,), + ) + rows = [] + size = 0 + for seq, payload in cursor.fetchall(): + record = {**payload, "seq": seq} + size += len(json.dumps(record).encode()) + if size > MAX_BATCH_BYTES: + break + rows.append(record) + cursor.execute("SELECT min(seq),max(seq),min(observed_at) FROM monitor_samples") + first, last, oldest = cursor.fetchone() + return dict( + schema=SCHEMA, + source_id=self.source, + samples=rows, + first_seq=first, + last_seq=last, + retained_since=oldest.timestamp() if oldest else None, + ) + + def maintain(self): + with self.connection() as db, db.cursor() as cursor: + cursor.execute("SELECT drop_chunks('monitor_samples',INTERVAL '7 days')") + cursor.execute("SELECT pg_database_size(current_database())") + size = cursor.fetchone()[0] + if size >= DATABASE_BUDGET: + # Drop one oldest completed chunk per pass, retaining the live hour. + cursor.execute( + "SELECT min(range_end) FROM timescaledb_information.chunks " + "WHERE hypertable_name='monitor_samples' AND range_end%s::timestamptz)", + (boundary,), + ) + cursor.execute("SELECT pg_database_size(current_database())") + size = cursor.fetchone()[0] + self.database_bytes = size + return size diff --git a/apps/node-agent/packaging/build_deb.py b/apps/node-agent/packaging/build_deb.py index 7a98886..8b5c567 100644 --- a/apps/node-agent/packaging/build_deb.py +++ b/apps/node-agent/packaging/build_deb.py @@ -11,7 +11,7 @@ import sys ROOT = Path(__file__).resolve().parents[1] -VERSION = "0.8.14" +VERSION = "0.8.15" sys.path.insert(0, str(ROOT.parents[1] / "scripts/packaging")) from debian import package @@ -43,7 +43,7 @@ Architecture: amd64 Maintainer: NODE.DC local build Section: admin Priority: optional -Depends: adduser, systemd, python3, python3-gi, gir1.2-gtk-3.0, gir1.2-webkit2-4.1, pkexec, polkitd, ca-certificates, hicolor-icon-theme, network-manager, iproute2 +Depends: adduser, systemd, python3, python3-gi, gir1.2-gtk-3.0, gir1.2-webkit2-4.1, pkexec, polkitd, ca-certificates, hicolor-icon-theme, network-manager, iproute2, python3-psycopg2, postgresql-16 (>= 16.15), timescaledb-2-oss-postgresql-16 (= 2.29.2~ubuntu24.04-1615) Description: Mission Core onboard computer configuration Local graphical setup, host inventory, SSH access and persistent node identity. """.encode() @@ -67,8 +67,13 @@ Description: Mission Core onboard computer configuration ("environment_helper.py", "usr/lib/mission-core-node/environment_helper.py", 0o644), ("mission-core-node-environment.service", "usr/lib/systemd/system/mission-core-node-environment.service", 0o644), ("60-environment.conf", "usr/share/mission-core-node/60-environment.conf", 0o644), + ("setup-monitor", "usr/lib/mission-core-node/setup-monitor", 0o755), + ("mission-core-node-monitor.service", "usr/lib/systemd/system/mission-core-node-monitor.service", 0o644), ]: files.append((path, (p / source).read_bytes(), mode)) + for path in sorted((ROOT / 'monitor').iterdir()): + if path.suffix in {'.py', '.sql'}: + files.append(('usr/lib/mission-core-node/monitor/' + path.name, path.read_bytes(), 0o644)) files.append(("usr/share/mission-core-node/environment-profile.json", (ROOT / "internal/node/environment-profile.json").read_bytes(), 0o644)) for name in ("realsense_prepare.py", "realsense_iio_access.py"): files.append(("usr/lib/mission-core-node/" + name, (p / name).read_bytes(), 0o644)) diff --git a/apps/node-agent/packaging/fetch_monitor_dependencies.py b/apps/node-agent/packaging/fetch_monitor_dependencies.py new file mode 100644 index 0000000..710a4e4 --- /dev/null +++ b/apps/node-agent/packaging/fetch_monitor_dependencies.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Stage verified Timescale OSS Ubuntu packages, without APT/system mutation.""" + +import argparse +import gzip +import hashlib +import json +import subprocess +import urllib.request +from pathlib import Path + +BASE = "https://packagecloud.io/timescale/timescaledb/ubuntu/" +VERSION = "2.29.2" + + +def fetch(url, limit): + with urllib.request.urlopen(url, timeout=30) as response: + data = response.read(limit + 1) + if len(data) > limit: + raise ValueError("Download budget exceeded") + return data + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("destination", type=Path) + args = parser.parse_args() + root = args.destination + root.mkdir(parents=True, exist_ok=True) + key = root / "timescale.asc" + key.write_bytes(fetch("https://packagecloud.io/timescale/timescaledb/gpgkey", 65536)) + keyring = root / "timescale.gpg" + subprocess.run( + ["gpg", "--batch", "--yes", "--dearmor", "--output", str(keyring), str(key)], check=True + ) + release = root / "InRelease" + release.write_bytes(fetch(BASE + "dists/noble/InRelease", 1048576)) + subprocess.run(["gpgv", "--keyring", str(keyring.resolve()), str(release)], check=True) + lines = release.read_text().splitlines() + hashes = {} + inside = False + for line in lines: + if line == "SHA256:": + inside = True + continue + if inside and not line.startswith(" "): + break + if inside: + sha, size, path = line.split() + hashes[path] = (sha, int(size)) + sha, size = hashes["main/binary-amd64/Packages.gz"] + compressed = fetch(BASE + "dists/noble/main/binary-amd64/Packages.gz", 16000000) + if len(compressed) != size or hashlib.sha256(compressed).hexdigest() != sha: + raise ValueError("Signed package index mismatch") + (root / "Packages.gz").write_bytes(compressed) + wanted = { + name: VERSION + "~ubuntu24.04-1615" + for name in ("timescaledb-2-oss-postgresql-16", "timescaledb-2-loader-postgresql-16") + } + artifacts = [] + for block in gzip.decompress(compressed).decode().split("\n\n"): + fields = dict( + line.split(": ", 1) + for line in block.splitlines() + if ": " in line and not line.startswith(" ") + ) + if ( + fields.get("Package") not in wanted + or fields.get("Version") != wanted[fields["Package"]] + ): + continue + path = fields["Filename"] + if not path.startswith("pool/") or ".." in path.split("/"): + raise ValueError("Invalid package path") + data = fetch(BASE + path, 64000000) + if hashlib.sha256(data).hexdigest() != fields["SHA256"]: + raise ValueError("Package hash mismatch") + target = root / Path(path).name + target.write_bytes(data) + artifacts.append( + dict( + package=fields["Package"], + version=fields["Version"], + file=target.name, + sha256=fields["SHA256"], + bytes=len(data), + depends=fields["Depends"], + ) + ) + if len(artifacts) != len(wanted): + raise ValueError("Pinned package not found") + manifest = dict( + source=BASE, + version=VERSION, + key_sha256=hashlib.sha256(key.read_bytes()).hexdigest(), + inrelease_sha256=hashlib.sha256(release.read_bytes()).hexdigest(), + artifacts=artifacts, + ) + (root / "manifest.json").write_text(json.dumps(manifest, indent=2) + "\n") + print(json.dumps(manifest, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/apps/node-agent/packaging/install-monitor-release b/apps/node-agent/packaging/install-monitor-release new file mode 100644 index 0000000..96d5da8 --- /dev/null +++ b/apps/node-agent/packaging/install-monitor-release @@ -0,0 +1,56 @@ +#!/bin/sh +# Run by the owner-facing release installer after hash verification. Scope is +# this product's packages and its dedicated cluster; never remove another DB. +set -eu +test "$(id -u)" = 0 +mc_release_dir=$1 +cd "$mc_release_dir" +sha256sum --check SHA256SUMS +mc_pg_guard=/etc/postgresql-common/createcluster.d/60-mission-core-install.conf +mc_created_guard=0 +cleanup() { + if [ "$mc_created_guard" = 1 ]; then + if [ "$(cat "$mc_pg_guard")" = 'create_main_cluster = false' ]; then + rm "$mc_pg_guard" + fi + fi +} +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' HUP TERM +if ! dpkg-query -W -f='${Status}' postgresql-16 2>/dev/null | grep -qx 'install ok installed'; then + # postgresql-common explicitly supports this drop-in. Suppress only automatic + # creation of the unrelated default main cluster during the first install. + test ! -e "$mc_pg_guard" + install -d -m 0755 /etc/postgresql-common/createcluster.d + printf '%s\n' 'create_main_cluster = false' > "$mc_pg_guard" + chmod 0644 "$mc_pg_guard" + mc_created_guard=1 +fi +DEBIAN_FRONTEND=noninteractive apt-get install -y --no-remove \ + "$mc_release_dir/timescaledb-2-loader-postgresql-16_2.29.2~ubuntu24.04-1615_amd64.deb" \ + "$mc_release_dir/timescaledb-2-oss-postgresql-16_2.29.2~ubuntu24.04-1615_amd64.deb" \ + "$mc_release_dir/mission-core-node_0.8.15_amd64.deb" \ + "$mc_release_dir/mission-core-xgrids-k1_0.1.14_amd64.deb" +systemctl is-active --quiet mission-core-node.service +systemctl is-active --quiet mission-core-k1.service +systemctl is-active --quiet postgresql@16-ndc-monitor.service +systemctl is-active --quiet mission-core-node-monitor.service +runuser -u mission-core-monitor -- /usr/bin/python3 - <<'PY' +import http.client,json,socket,time +deadline=time.monotonic()+20 +while True: + c=http.client.HTTPConnection('monitor',timeout=3) + c.sock=socket.socket(socket.AF_UNIX);c.sock.settimeout(3) + try: + c.sock.connect('/run/mission-core-monitor/monitor.sock') + c.request('GET','/status');r=c.getresponse();v=json.load(r) + if r.status==200 and v['storage']=='ready' and v['latest']['seq']>0:break + except (OSError, ValueError, KeyError, TypeError, http.client.HTTPException): + pass + finally:c.close() + if time.monotonic()>deadline:raise SystemExit('Local telemetry did not become ready') + time.sleep(1) +print('Local telemetry: ready') +PY +printf '%s\n' 'Mission Core Node R18: package and local archive checks passed.' diff --git a/apps/node-agent/packaging/launcher.py b/apps/node-agent/packaging/launcher.py index 42f532a..dae51e8 100644 --- a/apps/node-agent/packaging/launcher.py +++ b/apps/node-agent/packaging/launcher.py @@ -9,6 +9,7 @@ import re import socket import subprocess import threading +import syslog from urllib.parse import urlsplit import gi @@ -261,11 +262,15 @@ class NodeApplication(Gtk.Application): return False def load_failed(self, _view, _event, uri, _error): + syslog.openlog('mission-core-node-ui') + syslog.syslog(syslog.LOG_ERR, 'node-ui-load-failed') if local_url(uri): self.problem("Служба ноды недоступна. Повторно откройте приложение после восстановления службы.") return True - def process_failed(self, *_): + def process_failed(self, _view, reason): + syslog.openlog('mission-core-node-ui') + syslog.syslog(syslog.LOG_ERR, 'node-ui-process-terminated reason=' + str(int(reason))) self.problem("Окно приложения остановилось. Закройте и повторно откройте Mission Core Node. Служба борта продолжает работать отдельно.") def deny_permission(self, _view, permission): diff --git a/apps/node-agent/packaging/mission-core-node-monitor.service b/apps/node-agent/packaging/mission-core-node-monitor.service new file mode 100644 index 0000000..eb6e80b --- /dev/null +++ b/apps/node-agent/packaging/mission-core-node-monitor.service @@ -0,0 +1,38 @@ +[Unit] +Description=Mission Core onboard system telemetry collector +After=postgresql@16-ndc-monitor.service +Wants=postgresql@16-ndc-monitor.service + +[Service] +Type=simple +User=mission-core-monitor +Group=mission-core-node +SupplementaryGroups=systemd-journal +ExecStart=/usr/bin/python3 /usr/lib/mission-core-node/monitor/collector.py +RuntimeDirectory=mission-core-monitor +RuntimeDirectoryMode=0750 +StateDirectory=mission-core-monitor +StateDirectoryMode=0750 +UMask=0007 +Restart=on-failure +RestartSec=5 +MemoryHigh=96M +MemoryMax=128M +CPUQuota=15% +TasksMax=24 +Nice=10 +IOSchedulingClass=idle +NoNewPrivileges=yes +ProtectSystem=strict +ProtectHome=yes +PrivateTmp=yes +PrivateDevices=yes +ProtectKernelTunables=yes +ProtectKernelModules=yes +ProtectControlGroups=yes +RestrictAddressFamilies=AF_UNIX +CapabilityBoundingSet= +RestrictSUIDSGID=yes + +[Install] +WantedBy=multi-user.target diff --git a/apps/node-agent/packaging/postinst b/apps/node-agent/packaging/postinst index 8592e7c..ef9ce64 100644 --- a/apps/node-agent/packaging/postinst +++ b/apps/node-agent/packaging/postinst @@ -23,6 +23,7 @@ case "$1" in fi rm -f /run/mission-core-node-k1-upgrade-active fi + /usr/lib/mission-core-node/setup-monitor fi ;; esac diff --git a/apps/node-agent/packaging/preinst b/apps/node-agent/packaging/preinst index 814de8d..5cd888d 100644 --- a/apps/node-agent/packaging/preinst +++ b/apps/node-agent/packaging/preinst @@ -34,6 +34,7 @@ if [ "$1" = install ] || [ "$1" = upgrade ]; then (umask 077; : > /run/mission-core-node-k1-upgrade-active) systemctl stop mission-core-k1.service fi + systemctl stop mission-core-node-monitor.service 2>/dev/null || true fi . /etc/os-release if [ "${ID:-}" != ubuntu ] || [ "${VERSION_ID:-}" != 24.04 ]; then diff --git a/apps/node-agent/packaging/prerm b/apps/node-agent/packaging/prerm index da15f13..ba758fd 100644 --- a/apps/node-agent/packaging/prerm +++ b/apps/node-agent/packaging/prerm @@ -52,6 +52,10 @@ case "$1" in systemctl disable mission-core-realsense.service || true systemctl stop mission-core-node.service systemctl disable mission-core-node.service + systemctl stop mission-core-node-monitor.service + systemctl disable mission-core-node-monitor.service + systemctl stop postgresql@16-ndc-monitor.service + systemctl disable postgresql@16-ndc-monitor.service fi ;; esac diff --git a/apps/node-agent/packaging/setup-monitor b/apps/node-agent/packaging/setup-monitor new file mode 100644 index 0000000..ed9c466 --- /dev/null +++ b/apps/node-agent/packaging/setup-monitor @@ -0,0 +1,54 @@ +#!/bin/sh +# Fixed local database bootstrap; no supplied SQL, credentials or network address. +set -eu +test "$(id -u)" = 0 +if ! getent passwd mission-core-monitor >/dev/null; then + adduser --system --group --home /var/lib/mission-core-monitor --no-create-home --disabled-login mission-core-monitor +fi +install -d -m 0750 -o mission-core-monitor -g mission-core-node /var/lib/mission-core-monitor +install -d -m 0755 -o postgres -g postgres /run/mission-core-monitor-db +install -d -m 0755 /etc/tmpfiles.d +printf '%s\n' 'd /run/mission-core-monitor-db 0755 postgres postgres -' > /etc/tmpfiles.d/mission-core-monitor.conf +if [ ! -d /etc/postgresql/16/ndc-monitor ]; then + pg_createcluster 16 ndc-monitor --port=5433 --socketdir=/run/mission-core-monitor-db --datadir=/var/lib/mission-core-monitor-db --start-conf=auto -- --auth-local=peer --auth-host=reject +fi +install -d -m 0755 /etc/postgresql/16/ndc-monitor/conf.d +cat > /etc/postgresql/16/ndc-monitor/conf.d/60-mission-core-monitor.conf <<'CONF' +listen_addresses = '' +unix_socket_directories = '/run/mission-core-monitor-db' +shared_preload_libraries = 'timescaledb' +shared_buffers = '32MB' +work_mem = '2MB' +maintenance_work_mem = '16MB' +max_connections = 12 +max_worker_processes = 4 +max_parallel_workers = 0 +timescaledb.max_background_workers = 2 +timescaledb.telemetry_level = 'off' +max_wal_size = '128MB' +min_wal_size = '32MB' +temp_file_limit = '32MB' +statement_timeout = '5s' +log_statement = 'none' +log_min_error_statement = 'panic' +CONF +install -d -m 0755 /etc/systemd/system/postgresql@16-ndc-monitor.service.d +cat > /etc/systemd/system/postgresql@16-ndc-monitor.service.d/60-mission-core-monitor.conf <<'CONF' +[Service] +ExecStartPre=+/usr/bin/install -d -m 0755 -o postgres -g postgres /run/mission-core-monitor-db +MemoryHigh=256M +MemoryMax=384M +CPUQuota=25% +TasksMax=32 +Nice=10 +CONF +systemctl daemon-reload +systemctl restart postgresql@16-ndc-monitor.service +runuser -u postgres -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d postgres <<'SQL' +SELECT 'CREATE ROLE "mission-core-monitor" LOGIN NOSUPERUSER NOCREATEDB NOCREATEROLE' WHERE NOT EXISTS(SELECT FROM pg_roles WHERE rolname='mission-core-monitor') \gexec +SELECT 'CREATE DATABASE mission_core_monitor OWNER "mission-core-monitor"' WHERE NOT EXISTS(SELECT FROM pg_database WHERE datname='mission_core_monitor') \gexec +SQL +runuser -u postgres -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d mission_core_monitor -c 'CREATE EXTENSION IF NOT EXISTS timescaledb' +runuser -u mission-core-monitor -- psql -X -v ON_ERROR_STOP=1 -h /run/mission-core-monitor-db -p 5433 -d mission_core_monitor -f /usr/lib/mission-core-node/monitor/schema.sql +systemctl enable mission-core-node-monitor.service +systemctl restart mission-core-node-monitor.service diff --git a/apps/node-agent/ui/src/NodeSensors.tsx b/apps/node-agent/ui/src/NodeSensors.tsx index 04e069e..fffec59 100644 --- a/apps/node-agent/ui/src/NodeSensors.tsx +++ b/apps/node-agent/ui/src/NodeSensors.tsx @@ -3,5 +3,5 @@ import {createIsolatedRerunHost} from '../../../control-station/src/components/r import {SensorWorkspace} from '../../../../packages/sensor-ui/src/SensorWorkspace'; import type {SensorTransport} from '../../../../packages/sensor-ui/src/contracts'; import {request} from './api'; -const transport:SensorTransport={enrollment:{state:()=>request("/api/devices/enrollment"),submit:value=>request("/api/devices/enrollment/operations","POST",value),operation:id=>request("/api/devices/enrollment/operations/"+encodeURIComponent(id))},subscribe:(receive,unavailable)=>{const events=new EventSource('/api/devices/events');events.onmessage=e=>{try{receive(JSON.parse(e.data));}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();},inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))}; +const transport:SensorTransport={localPreview:{open:(command,signal)=>fetch("/api/devices/preview",{method:"POST",credentials:"same-origin",headers:{"Content-Type":"application/json"},body:JSON.stringify(command),signal}),read:(peer,after,signal)=>fetch("/api/devices/preview/read",{method:"POST",credentials:"same-origin",headers:{"Content-Type":"application/json"},body:JSON.stringify({peer_id:peer,after}),signal})},enrollment:{state:()=>request("/api/devices/enrollment"),submit:value=>request("/api/devices/enrollment/operations","POST",value),operation:id=>request("/api/devices/enrollment/operations/"+encodeURIComponent(id))},subscribe:(receive,unavailable)=>{const events=new EventSource('/api/devices/events');events.onmessage=e=>{try{receive(JSON.parse(e.data));}catch{unavailable();}};events.onerror=unavailable;return()=>events.close();},inventory:()=>request('/api/devices'),submit:value=>request('/api/devices/operations','POST',value),operation:id=>request('/api/devices/operations/'+encodeURIComponent(id))}; export function NodeSensors(){return ;} diff --git a/apps/node-agent/ui/src/RuntimeBoundary.tsx b/apps/node-agent/ui/src/RuntimeBoundary.tsx new file mode 100644 index 0000000..88fb623 --- /dev/null +++ b/apps/node-agent/ui/src/RuntimeBoundary.tsx @@ -0,0 +1,20 @@ +import {Component,type ReactNode} from 'react'; +import {Button,SettingsCard} from '@nodedc/ui-react'; + +export function runtimeDiagnostic(error:unknown,code='ui-error') { + const kind=error instanceof Error&&/^[A-Za-z]{1,48}$/.test(error.name)?error.name:'Error'; + // Never send exception messages, URLs, form values or arbitrary stack text. + const locations=error instanceof Error?(error.stack??'').match(/[A-Za-z0-9_-]+\.js:\d+:\d+/g)?.slice(0,8)??[]:[]; + void fetch('/api/monitor/events',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({code,kind,locations}),keepalive:true}).catch(()=>{}); +} +export function observeRuntime(){ + window.addEventListener('error',event=>runtimeDiagnostic(event.error)); + window.addEventListener('unhandledrejection',event=>runtimeDiagnostic(event.reason,'ui-rejection')); +} +export class RuntimeBoundary extends Component<{children:ReactNode;resetKey?:string},{failed:boolean}> { + state={failed:false}; + static getDerivedStateFromError(){return {failed:true};} + componentDidCatch(error:Error){runtimeDiagnostic(error,'ui-render-error');} + componentDidUpdate(previous:Readonly<{children:ReactNode;resetKey?:string}>){if(this.state.failed&&previous.resetKey!==this.props.resetKey)this.setState({failed:false});} + render(){return this.state.failed?:this.props.children;} +} diff --git a/apps/node-agent/ui/src/main.tsx b/apps/node-agent/ui/src/main.tsx index feff68e..b8d9f51 100644 --- a/apps/node-agent/ui/src/main.tsx +++ b/apps/node-agent/ui/src/main.tsx @@ -18,6 +18,7 @@ import "./node.css"; import { NodeSensors } from "./NodeSensors"; import { Home, HomeSettings } from "./Home"; import { usePresentation } from "./usePresentation"; +import {RuntimeBoundary,observeRuntime} from './RuntimeBoundary'; function App() { const node = useNode(); @@ -53,7 +54,7 @@ function App() { contexts={value ? [{ id: "board", label: value.name, description: value.host.hostname, icon: , onSelect: () => openView("overview") }] : []} items={views.filter(item => item.root === root).map(item => ({ id: item.id, label: item.label, icon: }))} activeId={workspace.activeView ?? undefined} onItemChange={id => openView(id as ViewId)} footer={Mission Core Node · {value?.version}} />} content={currentView && setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}>{content}} + utilityActions={[...(workspace.activeView === "ssh" ? [{ label: "Добавить доверенное устройство", icon: "plus" as const, onClick: () => setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}>{content}} stage={value ? :
    {pending ? :

    {locked ? "Подтвердите доступ в системном окне." : "Не удалось связаться со службой. Повторите подключение."}

    } @@ -64,4 +65,5 @@ function App() { setSettingsOpen(false)} presentation={presentation} /> ; } -createRoot(document.getElementById("root")!).render(); +observeRuntime(); +createRoot(document.getElementById("root")!).render(); diff --git a/docs/audits/2026-09-08-node-runtime-and-system-monitor.md b/docs/audits/2026-09-08-node-runtime-and-system-monitor.md new file mode 100644 index 0000000..0cc63ba --- /dev/null +++ b/docs/audits/2026-09-08-node-runtime-and-system-monitor.md @@ -0,0 +1,100 @@ +# Node local viewer and onboard system monitoring + +## Approved operator job and placement + +Owner request: investigate the black onboard window after local K1 START, and +add «Мониторинг системы БК» to the selected apparatus in Fleet. The selected +board owns measurements and history; UI navigation, capture and network outages +must not control collection. This is a detail view in the existing Fleet +workspace, not the compute-worker contour or a new primary navigation root. +An independent System workspace would lose the selected apparatus context; a +modal would constrain history inspection. The owner explicitly selected the +apparatus-detail entry. Novelty class A: domain content in admitted composition. + +Reuse SettingsCard, ApplicationPanel, Button, Select, StatusBadge, ResourceRow, +Icon(activity/refresh/chevron-left), and the existing TelemetrySeries domain +renderer. Domain cards contain current metrics; selected series shows source +time, min/max/mean and gaps. No missing generic visual primitive is required. +States: collecting, historical/offline, waiting for first sample, unavailable +metric, storage failure, replication backlog, loading/query failure, empty range. +No acquisition, provisioning, shell or host-reconfiguration action belongs here. + +## Reproduced diagnosis + +R17 installed WebKitGTK 2.52.6 on Ubuntu. Public standalone Rerun 0.36.3 starts +and stays responsive using WebGL2. WebGPU is unavailable. RTCPeerConnection is +undefined, and constructing it raises ReferenceError. The shared preview effect +constructs that API synchronously after viewer readiness, outside its catch. +Node had no React error boundary. The historical window and WebKit processes +remained alive with an empty accessible document; Node/K1 services did not restart. +App cgroup memory peak was about 811 MiB and OOM counters were all zero. Ten-minute +sysstat samples cannot rule out brief peaks but do not support memory exhaustion. +The camera consumer lease subsequently expired from queue age; that is a delivery +symptom, not proof of GPU/RAM exhaustion. + +Enabling WebKit's WebRTC setting alone and supplying missing GStreamer modules +in an isolated diagnostic environment did not expose the API. No global graphics +workaround or new browser engine is admitted on this evidence. The local carrier +uses authenticated loopback HTTP and the same plugin-owned bounded RRD/fMP4 +delivery, acknowledgements, recording cursor and release logic. Remote paired +WebRTC remains unchanged. There is no second capture implementation. + +## Telemetry architecture and limits + +An independent native collector samples Linux counters every second and commits +to a dedicated local PostgreSQL/TimescaleDB database. It uses only an OS-protected +Unix socket, with no database TCP listener or embedded credential. Acquisition +does not depend on that service. CPU/RAM/swap/pressure, per-interface bytes and +errors, per-disk I/O, filesystem capacity, temperatures, observable GPU counters, +USB topology/link speed and service cgroup usage/OOM counters are separate series. +Unsupported counters remain null with a reason; USB negotiated speed is never +presented as measured traffic. No USB payload, process arguments or environment +variables are collected. UI exceptions carry only category and asset positions. + +Raw retention target: seven days, with a two-GiB database target and two-GiB free +disk reserve. Capacity is checked once per minute; writes pause at the target +and oldest completed hourly chunks are removed. This is not a strict filesystem +quota: a minute of writes, WAL and temporary files may exceed the database target. Retention and capacity loss are visible, never disguised as zeros. +Bounded batches replicate over the existing paired mTLS connection. Core commits +idempotently before acknowledging; disconnects leave local collection running. +Latest status and historical delivery cursors are distinct so a fresh snapshot +cannot skip unreplicated history. Historical timestamps and boot identity survive +reconnection. Core's local archive is queryable while the board is offline. + +Acceptance must separate fixture checks, real host collection, local viewer +compatibility and owner hardware scans. A healthy repeated laboratory scan does +not establish critical-scenario qualification of the K1 Bridge contour. + + +## Implementation acceptance before installation + +A bounded native WebKitGTK probe on the owner Ubuntu loads the unmodified +production hook through a minified Vite bundle. Its source is synthetic RRD +(100 points) and a synthetic H.264/fMP4 camera; it cannot submit device commands. +The 2026-09-07 22:28 UTC probe reports the error boundary present with the shell +alive, then `hasScene=true`, `presented=true`, `cameraPresented=true`. Video time +advances to 2.9 s. After the fixture ends, freshness expires while the scene +remains. This establishes compatibility for these fixtures, not live K1 load. +Local transport uses completed bounded reads, a stable viewer cursor and native +EventTarget dispatch. Remote transport keeps WebRTC with guarded construction. + +Native PostgreSQL 16.15 / TimescaleDB 2.29.2 was exercised using privately +extracted signed packages and a temporary socket-only cluster. Three real host +samples with 184 discovered metrics were committed and replayed in order. The +temporary cluster was stopped and removed. Installed-service acceptance is a +separate gate. + +Core keeps seven-day raw data plus minute min/max/sum/count rollups. Long ranges +use complete minute buckets, preserving extrema; the selected end is rounded +to the last complete minute. Missing intervals break the plot. Replica writes +run outside the control heartbeat; only a durable committed cursor is ACKed. +The archive is a best-effort diagnostic record at one-second sampling resolution; +unobserved subsecond peaks and unsupported hardware counters remain unproved. + +Packages: Node 0.8.15 and K1 0.1.14, with exact signed Timescale OSS/loader +2.29.2~ubuntu24.04-1615. The dedicated `16/ndc-monitor` PostgreSQL cluster has no +TCP listener. Its service memory cap is 384 MiB, collector cap 128 MiB; CPU caps +are 25% and 15% of one core respectively. An initial PostgreSQL install suppresses +automatic creation of an unrelated default cluster through a temporary supported +postgresql-common configuration drop-in, removed on installer exit. Acquisition +is protected by the existing package preinst guard and is not a test action. diff --git a/packages/sensor-ui/src/contracts.ts b/packages/sensor-ui/src/contracts.ts index 803de5c..f5fbc6f 100644 --- a/packages/sensor-ui/src/contracts.ts +++ b/packages/sensor-ui/src/contracts.ts @@ -26,6 +26,10 @@ export interface SensorCommand { } export interface SensorOperation {state:string;error?:string;result?:unknown} export interface SensorTransport { + localPreview?: { + open: (command:SensorCommand, signal:AbortSignal) => Promise; + read: (peer:string, after:number, signal:AbortSignal) => Promise; + }; enrollment?: import('./enrollment').EnrollmentTransport; inventory: () => Promise; subscribe?: (receive:(value:SensorInventory)=>void, unavailable:()=>void) => (()=>void); diff --git a/plugins/xgrids-k1/frontend/src/sensors/localPreview.ts b/plugins/xgrids-k1/frontend/src/sensors/localPreview.ts new file mode 100644 index 0000000..d89cbd4 --- /dev/null +++ b/plugins/xgrids-k1/frontend/src/sensors/localPreview.ts @@ -0,0 +1,71 @@ +import {command, type Sensor, type SensorTransport} from '@mission-core/sensor-sdk'; +import {MEDIA_PROTOCOL} from './previewFrames'; + +export interface PreviewChannel { + readyState:string; + onmessage:((event:{data:string|ArrayBuffer})=>void)|null; + onclose:(()=>void)|null; + send:(value:string)=>void; + close:()=>void; +} + +class LocalChannel extends EventTarget implements PreviewChannel { + readyState='open'; + onmessage:PreviewChannel['onmessage']=null; + onclose:PreviewChannel['onclose']=null; + constructor(private acknowledge:(value:string)=>void){ + super(); + this.addEventListener('message',event=>this.onmessage?.({data:(event as MessageEvent).data})); + this.addEventListener('close',()=>this.onclose?.()); + } + send(value:string){this.acknowledge(value);} + close(){this.readyState='closed';} + receive(data:string|ArrayBuffer){if(this.readyState==='open')this.dispatchEvent(new MessageEvent('message',{data}));} + ended(){this.close();this.dispatchEvent(new Event('close'));} +} + +/** Framing may split at any TCP boundary. Only complete bounded messages leave. */ +export function localPreviewFrames(receive:(kind:number,payload:Uint8Array)=>void) { + let pending=new Uint8Array(0); + return (input:Uint8Array)=>{ + const joined=new Uint8Array(pending.length+input.length);joined.set(pending);joined.set(input,pending.length); + let offset=0; + while(joined.length-offset>=5){ + const kind=joined[offset], length=new DataView(joined.buffer,joined.byteOffset+offset+1,4).getUint32(0); + if(kind>6||length>32768)throw new Error('Invalid local preview frame'); + if(joined.length-offset<5+length)break; + receive(kind,joined.slice(offset+5,offset+5+length));offset+=5+length; + } + pending=joined.slice(offset); + }; +} + +/** Bounded completed responses work in the onboard WebKitGTK carrier. */ +export function localPreview(device:Sensor,transport:SensorTransport,parameters:Record,failed:()=>void) { + const carrier=transport.localPreview!; + const abort=new AbortController();let peer='',after=Number(parameters.after)||0; + const makeChannel=()=>new LocalChannel(value=>{if(value==='keepalive')return;const message=JSON.parse(value);if(!Number.isSafeInteger(message.ack)||message.ack{if(abort.signal.aborted)return;abort.abort();rrd.close();camera.close();if(peer)void transport.submit(command(device,'close-peer',{peer_id:peer})).catch(()=>{});}; + const start=async()=>{ + let response=await carrier.open(command(device,'offer',parameters),abort.signal); + const text=new TextDecoder(); + const parse=localPreviewFrames((kind,payload)=>{ + if(kind===0){const hello=JSON.parse(text.decode(payload));if(peer||hello.media_protocol!==MEDIA_PROTOCOL||!/^peer_[0-9a-f]{32}$/.test(hello.peer_id))throw new Error('Invalid local preview');peer=hello.peer_id;return;} + if(!peer)throw new Error('Missing preview admission'); + const channel=kind<=2||kind===6?rrd:camera; + if(kind>=5){channel.ended();return;} + channel.receive(kind%2?text.decode(payload):payload.buffer as ArrayBuffer); + }); + while(!abort.signal.aborted){ + if(!response.ok)throw new Error('Local preview unavailable'); + const bytes=await response.arrayBuffer(); + if(bytes.byteLength>196608)throw new Error('Local preview budget'); + if(abort.signal.aborted)return; + parse(new Uint8Array(bytes)); + if(!peer)throw new Error('Missing preview admission'); + if(!abort.signal.aborted)response=await carrier.read(peer,after,abort.signal); + } + }; + return {rrd,camera,start:async()=>{try{await start();}catch(error){if(!abort.signal.aborted)failed();throw error;}},close}; +} diff --git a/plugins/xgrids-k1/frontend/src/sensors/useK1Preview.ts b/plugins/xgrids-k1/frontend/src/sensors/useK1Preview.ts index 7f1bb11..a192811 100644 --- a/plugins/xgrids-k1/frontend/src/sensors/useK1Preview.ts +++ b/plugins/xgrids-k1/frontend/src/sensors/useK1Preview.ts @@ -4,6 +4,7 @@ import { perform, type Sensor, type SensorTransport } from './runtime'; import { MEDIA_PROTOCOL, previewFrames, previewRecording } from './previewFrames'; import { previewCamera } from './previewCamera'; import { previewFreshness } from './previewFreshness'; +import {localPreview, type PreviewChannel} from './localPreview'; export type PreviewStatus = { lidar: string; camera: string; @@ -76,13 +77,28 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR } } }; const viewer = native, freshness = previewFreshness(); - const pc = new RTCPeerConnection({ iceServers: [] }); - const rrd = pc.createDataChannel('rrd', { ordered: true }), camera = pc.createDataChannel('camera', { ordered: true }); - rrd.binaryType = 'arraybuffer'; - camera.binaryType = 'arraybuffer'; + if (!transport.localPreview && typeof RTCPeerConnection !== 'function') { + update({lidar:'Этот браузер не поддерживает просмотр. Откройте сцену в Mission Core.',camera:'Просмотр недоступен',retry:false}); + return; + } + let pc:RTCPeerConnection|null=null; + try {if(!transport.localPreview)pc=new RTCPeerConnection({iceServers:[]});} + catch {update({lidar:'Не удалось открыть канал просмотра',retry:false});return;} + const local=transport.localPreview?localPreview(device,transport,{acquisition_id:device.control?.acquisition_id,view_id:view.id,after:view.recording.after},()=>fail()):null; + let rrd:PreviewChannel|RTCDataChannel, camera:PreviewChannel|RTCDataChannel; + try { + rrd = local?.rrd ?? pc!.createDataChannel('rrd', { ordered: true }); + camera = local?.camera ?? pc!.createDataChannel('camera', { ordered: true }); + } catch { + local?.close();pc?.close(); + update({lidar:'Не удалось открыть канал просмотра',camera:'Просмотр недоступен',retry:false});return; + } + if('binaryType' in rrd)rrd.binaryType = 'arraybuffer'; + if('binaryType' in camera)camera.binaryType = 'arraybuffer'; + const closeConnection=()=>{local?.close();pc?.close();}; const {channel, previousRecording} = view; const fail = () => { if (!active || failed) - return; failed = true; update({ lidar: 'Лидар: восстанавливаем связь', camera: 'Камера: ожидаем соединение', retry: true, presented: false, cameraPresented: false }); clearInterval(follow); clearInterval(keepalive); pc.close(); }; + return; failed = true; update({ lidar: 'Лидар: восстанавливаем связь', camera: 'Камера: ожидаем соединение', retry: true, presented: false, cameraPresented: false }); clearInterval(follow); clearInterval(keepalive); closeConnection(); }; let cameraDecoded = false, cameraFailed = false, lastVideoTime = -1, lastVideoProgress = 0, seenLidar = false; const cameraFail = () => { cameraFailed = true; update({ camera: 'Камера: изображение недоступно', cameraPresented: false }); camera.close(); }; let decoder = previewCamera(video.current!, () => { cameraDecoded = true; lastVideoProgress = performance.now(); }, cameraFail); @@ -101,7 +117,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR fail(); }; camera.onclose = () => { if (active && !failed) cameraFail(); }; - rrd.onmessage = event => { + rrd.onmessage = (event:{data:string|ArrayBuffer}) => { if (!active) return; try { @@ -110,7 +126,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR if (value.type === 'preview-unavailable' && value.code === 'resume-expired') { failed = true; clearInterval(follow); clearInterval(keepalive); update({retry:false, presented:false, cameraPresented:false, lidar:'Сессия просмотра истекла. Закройте и откройте пространственную сцену.'}); - pc.close(); return; + closeConnection(); return; } if (batch || value.type !== 'rrd-batch' || !Number.isSafeInteger(value.sequence) || value.sequence < 1 || !value.lidar) throw new Error('Invalid RRD batch'); @@ -123,7 +139,7 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR fail(); } }; - camera.onmessage = event => { + camera.onmessage = (event:{data:string|ArrayBuffer}) => { if (!active || failed) return; try { @@ -176,11 +192,13 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR lidar: presented ? 'Лидар: данные поступают' : seenLidar ? 'Лидар: нет свежих данных' : 'Лидар: ожидаем данные', camera: cameraFailed ? 'Камера: изображение недоступно' : cameraPresented ? 'Камера: изображение поступает' : cameraDecoded ? 'Камера: нет свежих кадров' : 'Камера: ожидаем изображение' }); }, 250); + if(local){await local.start();return;} + if(!pc)return; await pc.setLocalDescription(await pc.createOffer()); if (pc.iceGatheringState !== 'complete') await new Promise((resolve, reject) => { iceTimeout = setTimeout(() => reject(new Error('ICE timeout')), 8000); - pc.onicegatheringstatechange = () => { if (pc.iceGatheringState === 'complete') { + pc.onicegatheringstatechange = () => { if (pc?.iceGatheringState === 'complete') { clearTimeout(iceTimeout); resolve(); } }; @@ -201,11 +219,11 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR if (answer.media_protocol !== MEDIA_PROTOCOL) { failed = true; clearInterval(follow); - pc.close(); + closeConnection(); update({ lidar: 'Для просмотра требуется обновление приложения на БК', camera: 'Камера: ожидаем обновление', retry: false }); return; } - pc.onconnectionstatechange = () => { if (active && ['failed', 'closed'].includes(pc.connectionState)) + pc.onconnectionstatechange = () => { if (active && ['failed', 'closed'].includes(pc?.connectionState ?? 'closed')) fail(); }; await pc.setRemoteDescription({ type: answer.type, sdp: answer.sdp }); keepalive = setInterval(() => { for (const c of [rrd, camera]) @@ -222,9 +240,8 @@ export function useK1Preview(device: Sensor, transport: SensorTransport, createR camera.onmessage = null; camera.onclose = null; rrd.onclose = null; - pc.onconnectionstatechange = null; - pc.onicegatheringstatechange = null; - pc.close(); + if(pc){pc.onconnectionstatechange = null;pc.onicegatheringstatechange = null;} + closeConnection(); rrdFrames.close(); cameraFrames.close(); decoder.close(); diff --git a/plugins/xgrids-k1/packaging/build_deb.py b/plugins/xgrids-k1/packaging/build_deb.py index 77e6e80..5878f6d 100644 --- a/plugins/xgrids-k1/packaging/build_deb.py +++ b/plugins/xgrids-k1/packaging/build_deb.py @@ -22,7 +22,7 @@ from credential_install import PROFILE_ID, validate # noqa: E402 from debian import package # noqa: E402 from runtime_payload import files as runtime_files # noqa: E402 -VERSION = "0.1.13" +VERSION = "0.1.14" RESOURCES = ( "plugins/xgrids-k1/profile_loader.py", "plugins/xgrids-k1/plugin.manifest.json", diff --git a/plugins/xgrids-k1/packaging/runtime-files.json b/plugins/xgrids-k1/packaging/runtime-files.json index 98c0257..82106ae 100644 --- a/plugins/xgrids-k1/packaging/runtime-files.json +++ b/plugins/xgrids-k1/packaging/runtime-files.json @@ -79,6 +79,7 @@ "src/k1link/viewer/__init__.py", "src/k1link/viewer/metrics.py", "src/k1link/viewer/node_media.py", + "src/k1link/viewer/node_local_media.py", "src/k1link/viewer/node_rerun.py", "src/k1link/viewer/recorded.py", "src/k1link/viewer/recorded_blueprint_lifecycle.py", diff --git a/src/k1link/device_plugins/xgrids_k1/node_bridge.py b/src/k1link/device_plugins/xgrids_k1/node_bridge.py index aa3cf27..5cd4bd6 100644 --- a/src/k1link/device_plugins/xgrids_k1/node_bridge.py +++ b/src/k1link/device_plugins/xgrids_k1/node_bridge.py @@ -382,6 +382,15 @@ def create_app(repository_root: Path): peers = NodeMediaPeers(bridge.rerun, bridge.service.camera_preview) sensor = NodeK1Sensor(bridge, peers) + from fastapi.responses import Response + from k1link.viewer.node_local_media import start_local, read_local + + async def preview_request(request): + data = await request.body() + if len(data) > 4096: + raise ValueError("Preview request exceeds bound") + return await request.json() + @asynccontextmanager async def lifespan(_app): yield @@ -398,6 +407,30 @@ def create_app(repository_root: Path): async def inventory(request: Request): return await sensor.inventory(request.headers["X-Node-Id"]) + @app.post("/local-preview") + async def preview(request: Request): + try: + command = await preview_request(request) + if datetime.fromisoformat(command["deadline_at"]) <= datetime.now(UTC): + raise ValueError("Preview request expired") + await sensor.admit_preview(command, request.headers["X-Node-Id"]) + identifier = await start_local(peers, command['parameters']) + data = await read_local(peers, identifier, command['parameters'].get('after', 0)) + return Response(data, media_type="application/vnd.missioncore.preview", + headers={"Cache-Control": "no-store"}) + except (ValueError, KeyError, TypeError): + return JSONResponse({"error": "Preview unavailable"}, status_code=409) + + @app.post("/local-preview/read") + async def preview_read(request: Request): + try: + data = await preview_request(request) + payload = await read_local(peers, data['peer_id'], data['after']) + return Response(payload, media_type="application/vnd.missioncore.preview", + headers={"Cache-Control": "no-store"}) + except (ValueError, KeyError, TypeError): + return JSONResponse({"error": "Preview expired"}, status_code=409) + @app.get("/prepare-safe") async def prepare_safe(): state = await sensor.raw_state() diff --git a/src/k1link/device_plugins/xgrids_k1/node_sensor.py b/src/k1link/device_plugins/xgrids_k1/node_sensor.py index abe0b02..e57a188 100644 --- a/src/k1link/device_plugins/xgrids_k1/node_sensor.py +++ b/src/k1link/device_plugins/xgrids_k1/node_sensor.py @@ -132,6 +132,17 @@ class NodeK1Sensor: "expected_state_revision": control["state_revision"], } + async def admit_preview(self, command, node_id): + item = project_sensor(await self.raw_state(), node_id) + if (not item or command.get("action_id") != "offer" + or command.get("session", {}).get("device_id") != item["id"] + or command["session"].get("session_id") != item["snapshot"]["context"]["session_id"] + or item["snapshot"]["acquisition"] != "streaming" + or not item["control"]["acquisition_id"] + or command.get("parameters", {}).get("acquisition_id") != item["control"]["acquisition_id"]): + raise ValueError("Acquisition is not active or changed") + return item + async def execute(self, command, node_id): state = await self.raw_state() item = project_sensor(state, node_id) diff --git a/src/k1link/fleet/monitor.py b/src/k1link/fleet/monitor.py new file mode 100644 index 0000000..8ce1001 --- /dev/null +++ b/src/k1link/fleet/monitor.py @@ -0,0 +1,390 @@ +"""Idempotent local replica of a board-owned Timescale history, behind fleet mTLS.""" + +import json +import math +import re +import sqlite3 +import threading +import time +from uuid import UUID + +SCHEMA = "missioncore.node-system-monitor/v1" +KEY = re.compile(r"^[A-Za-z0-9_.:-]{1,180}$") + + +def checked_sample(value): + if not isinstance(value, dict) or set(value) != { + "seq", + "at", + "boot_id", + "uptime", + "values", + "events", + }: + raise ValueError("Invalid sample") + if type(value["seq"]) is not int or not 0 < value["seq"] < 2**53: + raise ValueError("Invalid sequence") + if str(UUID(value["boot_id"])) != value["boot_id"]: + raise ValueError("Invalid boot") + for name in ["at", "uptime"]: + if ( + type(value[name]) not in (int, float) + or not math.isfinite(value[name]) + or value[name] < 0 + ): + raise ValueError("Invalid time") + if not isinstance(value["values"], dict) or len(value["values"]) > 384: + raise ValueError("Metric budget") + for key, number in value["values"].items(): + if not KEY.fullmatch(key) or ( + number is not None and (type(number) not in (int, float) or not math.isfinite(number)) + ): + raise ValueError("Invalid metric") + if not isinstance(value["events"], list) or len(value["events"]) > 64: + raise ValueError("Event budget") + for event in value["events"]: + if ( + not isinstance(event, dict) + or set(event) - {"code", "kind", "locations", "at"} + or event.get("code") + not in { + "ui-error", + "ui-rejection", + "ui-render-error", + "system-oom", + "gpu-reset", + "disk-error", + "usb-disconnected", + "usb-error", + "service-failed", + "ui-process-exit", + "ui-load-failed", + } + or not re.fullmatch("[A-Za-z]{1,48}", event.get("kind", "")) + or not isinstance(event.get("locations"), list) + or len(event["locations"]) > 8 + or any( + not isinstance(v, str) + or not re.fullmatch(r"[A-Za-z0-9_-]{1,96}\.js:[0-9]{1,9}:[0-9]{1,9}", v) + for v in event["locations"] + ) + ): + raise ValueError("Invalid event") + if ( + type(event.get("at")) not in (int, float) + or not math.isfinite(event["at"]) + or event["at"] < 0 + ): + raise ValueError("Invalid event time") + return json.dumps(value, allow_nan=False, separators=(",", ":"), sort_keys=True) + + +class MonitorReplica: + def __init__(self, root): + path = root / "monitor.sqlite3" + if path.is_symlink(): + raise ValueError("Invalid monitor store") + self.db = sqlite3.connect(path, check_same_thread=False, timeout=2) + try: + path.chmod(0o600) + self.lock = threading.RLock() + self.maintenance = 0 + self.db.execute("PRAGMA journal_mode=WAL") + self.db.execute("PRAGMA synchronous=FULL") + self.db.execute("PRAGMA max_page_count=524288") + self.db.execute( + "CREATE TABLE IF NOT EXISTS samples(node TEXT,source TEXT,seq INTEGER,at REAL," + "body TEXT,PRIMARY KEY(node,source,seq))" + ) + self.db.execute("CREATE INDEX IF NOT EXISTS samples_time ON samples(node,at)") + self.db.execute( + "CREATE TABLE IF NOT EXISTS minutes(node TEXT,at INTEGER,body TEXT," + "PRIMARY KEY(node,at))" + ) + self.db.execute( + "CREATE TABLE IF NOT EXISTS events(node TEXT,source TEXT,seq INTEGER,at REAL," + "body TEXT,PRIMARY KEY(node,source,seq))" + ) + self.db.execute("CREATE INDEX IF NOT EXISTS events_time ON events(node,at)") + self.db.execute( + "CREATE TABLE IF NOT EXISTS states(node TEXT PRIMARY KEY,source TEXT," + "body TEXT,received REAL,ack INTEGER)" + ) + self.db.commit() + except Exception: + self.db.close() + raise + + def ingest(self, node, value): + if not isinstance(value, dict) or value.get("schema") != SCHEMA: + return None + source = value.get("source_id") + if not isinstance(source, str) or str(UUID(source)) != source: + raise ValueError("Invalid source") + definitions = value.get("definitions") + if not isinstance(definitions, dict) or len(definitions) > 384: + raise ValueError("Definition budget") + for key, definition in definitions.items(): + if ( + not KEY.fullmatch(key) + or not isinstance(definition, dict) + or set(definition) != {"label", "group", "resource", "unit", "reason"} + ): + raise ValueError("Invalid definition") + if any( + v is not None and (not isinstance(v, str) or len(v) > 256) + for v in definition.values() + ): + raise ValueError("Invalid definition") + if value.get("latest") is not None: + checked_sample(value["latest"]) + batch = value.get("batch") or {} + rows = batch.get("samples", []) + if ( + not isinstance(rows, list) + or len(rows) > 32 + or (rows and (batch.get("source_id") != source or batch.get("schema") != SCHEMA)) + ): + raise ValueError("Invalid batch") + serialized = [(row, checked_sample(row)) for row in rows] + if sum(len(text.encode()) for _, text in serialized) > 196608: + raise ValueError("Batch budget") + if any( + a[0]["seq"] >= b[0]["seq"] for a, b in zip(serialized, serialized[1:], strict=False) + ): + raise ValueError("Unordered batch") + snapshot = { + key: item + for key, item in value.items() + if key + in { + "schema", + "source_id", + "latest", + "definitions", + "storage", + "database_bytes", + "sample_interval_seconds", + "retention_days", + "database_budget_bytes", + "free_reserve_bytes", + } + } + snapshot.update( + retained_since=batch.get("retained_since"), first_seq=batch.get("first_seq") + ) + with self.lock, self.db: + previous = self.db.execute( + "SELECT source,ack FROM states WHERE node=?", (node,) + ).fetchone() + ack = previous[1] if previous and previous[0] == source else 0 + minutes = {} + for row, text in serialized: + old = self.db.execute( + "SELECT body FROM samples WHERE node=? AND source=? AND seq=?", + (node, source, row["seq"]), + ).fetchone() + if old and old[0] != text: + raise ValueError("Conflicting history sequence") + self.db.execute( + "INSERT OR IGNORE INTO samples VALUES(?,?,?,?,?)", + (node, source, row["seq"], row["at"], text), + ) + if not old: + minute = int(row["at"] // 60) * 60 + if minute not in minutes: + saved = self.db.execute( + "SELECT body FROM minutes WHERE node=? AND at=?", (node, minute) + ).fetchone() + minutes[minute] = json.loads(saved[0]) if saved else {} + for key, number in row["values"].items(): + if number is not None: + aggregate = minutes[minute].setdefault(key, [number, number, 0, 0]) + aggregate[0] = min(aggregate[0], number) + aggregate[1] = max(aggregate[1], number) + aggregate[2] += number + aggregate[3] += 1 + if row["events"]: + self.db.execute( + "INSERT INTO events VALUES(?,?,?,?,?)", + (node, source, row["seq"], row["at"], json.dumps(row["events"])), + ) + ack = max(ack, row["seq"]) + for minute, aggregate in minutes.items(): + self.db.execute( + "INSERT INTO minutes VALUES(?,?,?) ON CONFLICT(node,at) " + "DO UPDATE SET body=excluded.body", + (node, minute, json.dumps(aggregate)), + ) + self.db.execute( + "INSERT INTO states VALUES(?,?,?,?,?) ON CONFLICT(node) DO UPDATE SET " + "source=excluded.source,body=excluded.body,received=excluded.received,ack=excluded.ack", + (node, source, json.dumps(snapshot), time.time(), ack), + ) + if time.time() - self.maintenance > 60: + self.db.execute("DELETE FROM samples WHERE at= 21600 + width = math.ceil(window / 300 / 60) * 60 if coarse else max(1, window // 300) + if coarse: + end = math.floor(end / 60) * 60 + start = end - window + path = '$.values."' + metric + '"' + with self.lock: + row = self.db.execute( + "SELECT body,received,ack FROM states WHERE node=?", (node,) + ).fetchone() + if not row: + return dict(available=False, series=[], events=[]) + snapshot = json.loads(row[0]) + if coarse: + aggregate = '$."' + metric + '"' + points = self.db.execute( + "SELECT CAST(at/? AS INTEGER),min(json_extract(body,?)), " + "max(json_extract(body,?)),sum(json_extract(body,?))*1.0/" + "nullif(sum(json_extract(body,?)),0),sum(json_extract(body,?)) " + "FROM minutes WHERE node=? AND at>=? AND at=? AND at<=? GROUP BY CAST(at/? AS INTEGER) ORDER BY 1", + (width, path, path, path, path, node, start, end, width), + ).fetchall() + event_rows = self.db.execute( + "SELECT body FROM events WHERE node=? AND at>=? AND at<=? " + "ORDER BY at DESC LIMIT 64", + (node, start, end), + ).fetchall() + earliest = self.db.execute( + "SELECT min(at) FROM samples WHERE node=?", (node,) + ).fetchone()[0] + latest = snapshot.get("latest") + fresh = bool(latest and -5 <= time.time() - latest["at"] < 10 and time.time() - row[1] < 15) + return dict( + available=True, + **snapshot, + fresh=fresh, + received_at=row[1], + replicated_after=row[2], + archive_since=earliest, + backlog=max(0, (latest or {}).get("seq", 0) - row[2]), + start=start, + end=end, + bucket_seconds=width, + series=[ + dict(at=bucket * width, min=minimum, max=maximum, mean=mean, count=count) + for bucket, minimum, maximum, mean, count in points + ], + events=[event for record in event_rows for event in json.loads(record[0])][:64], + ) + + def close(self): + with self.lock: + self.db.close() + + +class MonitorReceiver: + """Replica I/O never runs under the fleet/control lock, including startup. + + Heartbeats may repeat a batch until this worker commits it. Replacing a + queued retry is safe: the sender advances only from the returned commit ACK. + """ + + def __init__(self, root): + self.root = root + self.lock = threading.Lock() + self.pending = {} + self.acks = {} + self.archive = None + self.worker = None + self.wake = threading.Event() + self.stop = threading.Event() + + def submit(self, node, value): + if not isinstance(value, dict) or value.get("schema") != SCHEMA: + return None + with self.lock: + if node not in self.pending and len(self.pending) >= 32: + return None + self.pending[node] = value + if self.worker is None: + self.worker = threading.Thread( + target=self.run, name="fleet-monitor-replica", daemon=True + ) + self.worker.start() + self.wake.set() + ack = self.acks.get(node) + return ack if ack and ack["source_id"] == value.get("source_id") else None + + def run(self): + try: + while not self.stop.is_set(): + self.wake.wait(1) + with self.lock: + pending, self.pending = self.pending, {} + self.wake.clear() + if self.archive is None: + try: + self.archive = MonitorReplica(self.root) + except (OSError, ValueError, sqlite3.Error): + continue + for node, value in pending.items(): + try: + if self.archive is None: + self.archive = MonitorReplica(self.root) + ack = self.archive.ingest(node, value) + with self.lock: + self.acks[node] = ack + except (OSError, ValueError, TypeError, KeyError, sqlite3.Error): + # No ACK: the board retains and retries its own archive. + continue + finally: + if self.archive is not None: + self.archive.close() + + def query(self, *args, **kwargs): + with self.lock: + if self.worker is None: + self.worker = threading.Thread( + target=self.run, name="fleet-monitor-replica", daemon=True + ) + self.worker.start() + self.wake.set() + try: + if self.archive is not None: + return self.archive.query(*args, **kwargs) + except sqlite3.Error: + pass + return dict(available=False, series=[], events=[]) + + def close(self): + self.stop.set() + self.wake.set() + if self.worker is not None: + self.worker.join(5) diff --git a/src/k1link/fleet/registry.py b/src/k1link/fleet/registry.py index 5f7ab01..04fdda9 100644 --- a/src/k1link/fleet/registry.py +++ b/src/k1link/fleet/registry.py @@ -33,6 +33,9 @@ class FleetRegistry: from .device_enrollment import DeviceEnrollment self.device_enrollment = DeviceEnrollment() + from .monitor import MonitorReceiver + + self.monitor = MonitorReceiver(root) self.trust = CoreTrust(root) path = root / "fleet.sqlite3" if path.is_symlink(): @@ -93,6 +96,7 @@ class FleetRegistry: server.server_close() with self.lock: self.db.close() + self.monitor.close() def listen(self, address): from .transport import NodeChannelServer @@ -407,4 +411,10 @@ class FleetRegistry: } row["binding"]["client_pem"] = self.trust.leaf(node_id, key) self.save(row) - return 200, {"ok": True, "client_pem": row["binding"]["client_pem"], **sensor_response, **enrollment_response} + monitor_ack = None + try: + monitor_ack = self.monitor.submit(node_id, value.get("monitor")) + except (ValueError, TypeError, KeyError, sqlite3.Error): + # Telemetry persistence failure must not block device control. + pass + return 200, {"monitor_ack": monitor_ack, "ok": True, "client_pem": row["binding"]["client_pem"], **sensor_response, **enrollment_response} diff --git a/src/k1link/viewer/node_local_media.py b/src/k1link/viewer/node_local_media.py new file mode 100644 index 0000000..9a4ccc8 --- /dev/null +++ b/src/k1link/viewer/node_local_media.py @@ -0,0 +1,137 @@ +"""Local HTTP carrier for the same bounded, acknowledged Node preview delivery. + +The Unix-only plugin endpoint is proxied by the authenticated Node UI. It never +starts acquisition, rewrites network settings, or creates a second recorder. +""" + +import asyncio +import json +import time +from uuid import UUID, uuid4 + +from .node_media import MEDIA_PROTOCOL + + +class LocalConnection: + def __init__(self): + self.queue = asyncio.Queue(maxsize=40) + self.buffered = 0 + self.closed = False + + def push(self, kind, payload): + if self.closed: + return + if isinstance(payload, str): + payload = payload.encode() + if len(payload) > 32768: + raise ValueError("Local preview message exceeds bound") + frame = bytes([kind]) + len(payload).to_bytes(4, "big") + payload + self.queue.put_nowait(frame) + self.buffered += len(frame) + + async def close(self): + self.closed = True + + +class LocalChannel: + def __init__(self, connection, label): + self.connection, self.label = connection, label + self.closed = False + + @property + def readyState(self): + return "closed" if self.closed or self.connection.closed else "open" + + @property + def bufferedAmount(self): + return self.connection.buffered + + def send(self, payload): + kind = (1 if self.label == "rrd" else 3) + (not isinstance(payload, str)) + self.connection.push(kind, payload) + + def close(self): + if not self.closed: + self.closed = True + self.connection.push(5 if self.label == "camera" else 6, b"") + + +def admit_local(parameters): + view_id, after = parameters.get("view_id"), parameters.get("after", 0) + if not isinstance(view_id, str) or str(UUID(view_id)) != view_id: + raise ValueError("Preview view identifier required") + if type(after) is not int or not 0 <= after <= 2**53 - 1: + raise ValueError("Invalid preview cursor") + return view_id, after + + +async def start_local(peers, parameters): + view_id, after = admit_local(parameters) + for identifier, entry in list(peers.items.items()): + if entry.get("local") and entry["view_id"] == view_id: + await peers.close(identifier) + if len(peers.items) >= 2: + raise ValueError("Close another live viewer") + connection = LocalConnection() + identifier = "peer_" + uuid4().hex + entry = { + "pc": connection, + "seen": time.monotonic(), + "tasks": [], + "labels": {"rrd", "camera"}, + "view_id": view_id, + "after": after, + "subscriber": None, + "local": True, + "reading": False, + } + peers.items[identifier] = entry + connection.push(0, json.dumps({"peer_id": identifier, "media_protocol": MEDIA_PROTOCOL})) + for label in ("rrd", "camera"): + entry["tasks"].append( + asyncio.create_task(peers.deliver(identifier, LocalChannel(connection, label))) + ) + + async def expiry(): + while identifier in peers.items: + await asyncio.sleep(5) + if time.monotonic() - entry["seen"] > 30: + await peers.close(identifier) + + entry["tasks"].append(asyncio.create_task(expiry())) + return identifier + + +async def read_local(peers, identifier, sequence): + acknowledge(peers, identifier, sequence) + entry = peers.items[identifier] + if entry["reading"]: + raise ValueError("Concurrent preview read") + entry["reading"] = True + connection = entry["pc"] + try: + try: + first = await asyncio.wait_for(connection.queue.get(), 1) + except TimeoutError: + return b"" + frames = [first] + size = len(first) + while size < 131072 and not connection.queue.empty(): + frame = connection.queue.get_nowait() + frames.append(frame) + size += len(frame) + connection.buffered -= size + return b"".join(frames) + finally: + entry["reading"] = False + + +def acknowledge(peers, identifier, sequence): + entry = peers.items.get(identifier) + if not entry or not entry.get("local"): + raise ValueError("Local preview expired") + if type(sequence) is not int or not 0 <= sequence <= 2**53 - 1: + raise ValueError("Invalid preview acknowledgement") + entry["seen"] = time.monotonic() + if entry["subscriber"] is not None: + entry["subscriber"].acknowledge(sequence) diff --git a/src/k1link/viewer/node_media.py b/src/k1link/viewer/node_media.py index b97dd7e..17fd7a0 100644 --- a/src/k1link/viewer/node_media.py +++ b/src/k1link/viewer/node_media.py @@ -252,9 +252,11 @@ class NodeMediaPeers: async def close(self, identifier): entry = self.items.pop(identifier, None) if entry: - for task in entry["tasks"]: - if task is not asyncio.current_task(): - task.cancel() + tasks = [task for task in entry["tasks"] if task is not asyncio.current_task()] + for task in tasks: + task.cancel() + # A reconnect must not race the previous delivery lease's finally. + await asyncio.gather(*tasks, return_exceptions=True) with suppress(Exception): await entry["pc"].close() diff --git a/src/k1link/web/fleet_api.py b/src/k1link/web/fleet_api.py index befe47d..59c498e 100644 --- a/src/k1link/web/fleet_api.py +++ b/src/k1link/web/fleet_api.py @@ -51,6 +51,21 @@ class AddRequest(BaseModel): router = APIRouter(prefix="/api/v1/fleet", tags=["fleet"]) +@router.get("/{vehicle_id}/monitor") +def board_monitor(vehicle_id: str, response: Response, + fleet: Annotated[FleetRegistry, Depends(local_operator)], + metric: str = "cpu.usage", window: int = 900, end: float | None = None): + response.headers["Cache-Control"] = "no-store" + try: + with fleet.lock: + node_id = fleet.find(vehicle_id)["node_id"] + return fleet.monitor.query(node_id, metric, window, end) + except PairingError as error: + raise HTTPException(404, str(error)) from None + except ValueError: + raise HTTPException(400, "Выберите доступный период и показатель.") from None + + @router.get("") def fleet_list(response: Response, fleet: Annotated[FleetRegistry, Depends(local_operator)]): response.headers["Cache-Control"] = "no-store" diff --git a/tests/test_board_monitor.py b/tests/test_board_monitor.py new file mode 100644 index 0000000..48c87e7 --- /dev/null +++ b/tests/test_board_monitor.py @@ -0,0 +1,168 @@ +import importlib.util +import sqlite3 +import time +from pathlib import Path +from uuid import uuid4 + +import pytest + +from k1link.fleet.monitor import SCHEMA, MonitorReceiver, MonitorReplica + +ROOT = Path(__file__).resolve().parents[1] + + +def module(name): + spec = importlib.util.spec_from_file_location( + name, ROOT / "apps/node-agent/monitor" / (name + ".py") + ) + result = importlib.util.module_from_spec(spec) + spec.loader.exec_module(result) + return result + + +def record(seq, at, boot, value=25): + return dict(seq=seq, at=at, boot_id=boot, uptime=seq, values={"cpu.usage": value}, events=[]) + + +def envelope(source, rows, latest=None): + return dict( + schema=SCHEMA, + source_id=source, + storage="ready", + latest=latest or rows[-1], + definitions={ + "cpu.usage": dict(label="CPU", group="CPU", resource="cpu", unit="%", reason=None) + }, + batch=dict(schema=SCHEMA, source_id=source, samples=rows), + ) + + +def test_replica_retries_restart_backfill_and_latest_are_separate(tmp_path): + source, boot = str(uuid4()), str(uuid4()) + now = time.time() + rows = [record(i, now - 40 + i, boot) for i in range(1, 41)] + archive = MonitorReplica(tmp_path) + assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32 + result = archive.query("board") + assert result["latest"]["seq"] == 40 and result["backlog"] == 8 + assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 32 + archive.close() + archive = MonitorReplica(tmp_path) + # A lost reply cannot duplicate history; ACK comes from the durable commit. + assert archive.ingest("board", envelope(source, rows[:32], rows[-1]))["after"] == 32 + assert archive.ingest("board", envelope(source, rows[32:]))["after"] == 40 + assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 40 + assert archive.query("board")["backlog"] == 0 + # Retention on the board can leave a real gap; no zero samples are invented. + assert archive.ingest("board", envelope(source, [record(51, now + 11, boot)]))["after"] == 51 + archive.close() + + +def test_conflicting_or_nonfinite_batch_rolls_back_and_missing_values_are_null(tmp_path): + source, boot = str(uuid4()), str(uuid4()) + now = time.time() + archive = MonitorReplica(tmp_path) + original = record(2, now - 20, boot) + archive.ingest("board", envelope(source, [original])) + bad = envelope(source, [record(1, now - 21, boot), {**original, "values": {"cpu.usage": 99}}]) + with pytest.raises(ValueError): + archive.ingest("board", bad) + assert archive.db.execute("SELECT count(*) FROM samples").fetchone()[0] == 1 + with pytest.raises(ValueError): + archive.ingest("board", envelope(source, [record(3, now, boot, float("nan"))])) + archive.ingest("board", envelope(source, [record(3, now, boot, None)])) + point = archive.query("board")["series"][-1] + assert point["max"] is None and point["count"] == 0 + archive.close() + + +def test_receiver_failure_isolated_and_offline_history_reopens(tmp_path, monkeypatch): + source, boot = str(uuid4()), str(uuid4()) + archive = MonitorReplica(tmp_path) + archive.ingest("board", envelope(source, [record(1, time.time(), boot)])) + archive.close() + receiver = MonitorReceiver(tmp_path) + try: + deadline = time.monotonic() + 2 + while not receiver.query("board")["available"] and time.monotonic() < deadline: + time.sleep(0.01) + assert receiver.query("board")["available"] + monkeypatch.setattr( + receiver.archive, + "ingest", + lambda *_: (_ for _ in ()).throw(sqlite3.OperationalError("synthetic full disk")), + ) + start = time.monotonic() + assert receiver.submit("board", envelope(source, [record(2, time.time(), boot)])) is None + assert time.monotonic() - start < 0.1 + time.sleep(0.05) + assert receiver.acks.get("board") is None + finally: + receiver.close() + + +def test_linux_rates_distinguish_absent_initial_reset_and_gap(tmp_path): + now = [100] + + def write(path, text): + target = tmp_path / path + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + + def counters(cpu, idle, rx): + write("proc/stat", f"cpu {cpu} 0 0 {idle} 0 0 0 0\ncpu0 {cpu} 0 0 {idle} 0 0 0 0\n") + write("sys/class/net/eth0/statistics/rx_bytes", str(rx)) + + write("sys/class/net/eth0/speed", "-1") + write("sys/bus/usb/devices/1-2/idVendor", "0000") + write("sys/bus/usb/devices/1-2/speed", "480") + write( + "sys/fs/cgroup/system.slice/mission-core-node.service/memory.events", "oom 3\noom_kill 2\n" + ) + metrics = module("linux_metrics").LinuxMetrics(tmp_path, lambda: now[0]) + counters(100, 100, 1000) + first = metrics.sample() + assert first["cpu.usage"] is None and first["net.eth0.rx_bytes"] is None + now[0] += 1 + counters(150, 150, 1512) + second = metrics.sample() + assert second["cpu.usage"] == 50 and second["net.eth0.rx_bytes"] == 512 + assert second["net.eth0.speed"] is None + assert second["usb.1-2.speed"] == 480 and second["usb.1-2.traffic"] is None + assert second["service.mission-core-node.service.oom"] == 2 + now[0] += 1 + counters(5, 5, 100) + assert metrics.sample()["net.eth0.rx_bytes"] is None + now[0] += 11 + counters(25, 25, 900) + assert metrics.sample()["cpu.usage"] is None + + +def test_journal_retains_classification_without_freeform_data(): + classify = module("journal_events").classify + value = classify( + { + "__REALTIME_TIMESTAMP": "123000000", + "_TRANSPORT": "kernel", + "MESSAGE": "Out of memory: Killed process 23 secret argument", + } + ) + assert value == dict(code="system-oom", kind="Kernel", at=123, locations=[]) + assert classify({"MESSAGE": "private Wi-Fi password"}) is None + + +def test_long_range_aggregates_keep_peaks_counts_and_exclude_incomplete_minute(tmp_path): + source, boot = str(uuid4()), str(uuid4()) + now = int(time.time()//60)*60 + archive = MonitorReplica(tmp_path) + rows = [record(1, now-120, boot, 10), record(2, now-110, boot, 90), + record(3, now-60, boot, 30), record(4, now+1, boot, 100)] + value = envelope(source, rows) + archive.ingest('board', value) + archive.ingest('board', value) # Duplicate delivery cannot inflate aggregates. + result = archive.query('board', window=21600, end=now+10) + assert result['end'] == now + assert sum(point['count'] for point in result['series']) == 3 + assert max(point['max'] for point in result['series']) == 90 + assert min(point['min'] for point in result['series']) == 10 + archive.close() diff --git a/tests/test_node_k1_package_lifecycle.py b/tests/test_node_k1_package_lifecycle.py index 8bbb145..2c7134d 100644 --- a/tests/test_node_k1_package_lifecycle.py +++ b/tests/test_node_k1_package_lifecycle.py @@ -34,6 +34,7 @@ case "$1" in esac ''') (binary / "getent").write_text("#!/bin/sh\nexit 0\n") + (binary / "setup-monitor").write_text("#!/bin/sh\nprintf '%s\\n' monitor-setup >> \"$TEST_EVENTS\"\n") (binary / "dpkg-query").write_text( "#!/bin/sh\necho 'install ok " + ("installed" if configured else "unpacked") + "'\n" ) @@ -45,6 +46,7 @@ esac script = tmp_path / name script.write_text((PACKAGING / name).read_text() .replace("/run/", str(run) + "/") + .replace("/usr/lib/mission-core-node/setup-monitor", str(binary / "setup-monitor")) .replace("/etc/os-release", str(release))) subprocess.run(["/bin/sh", str(script), *args], env=env, check=True, capture_output=True, text=True) diff --git a/tests/test_node_local_media.py b/tests/test_node_local_media.py new file mode 100644 index 0000000..c7dd81b --- /dev/null +++ b/tests/test_node_local_media.py @@ -0,0 +1,98 @@ +import asyncio +import json +from uuid import uuid4 + +import pytest + +from k1link.viewer.node_local_media import acknowledge, admit_local, read_local, start_local +from k1link.viewer.node_media import NodeMediaPeers + + +def test_local_stream_fragments_ack_resume_and_releases_before_reconnect(): + payload = b"RRF2" + bytes(range(256)) * 256 + released = [] + subscriptions = [] + + class Subscription: + pending = None + + def __init__(self, after): + self.after = after + self.sent = False + + def next_batch(self, **_): + if self.sent: + return b"" + self.sent = True + self.pending = (self.after + 1, payload, None) + return self.pending + + def snapshot(self, _): + return dict(sequence=self.after + 1, age_ms=0, points=5000) + + def acknowledge(self, seq): + if seq == self.after + 1: + self.pending = None + + def release(self): + released.append(self.after) + + class Hub: + def subscribe(self, view_id, after): + sub = Subscription(after) + subscriptions.append((view_id, sub)) + return sub + + class Camera: + def snapshot(self): + return {"recording": {"source_end_expected": True}} + + async def run(): + peers = NodeMediaPeers(Hub(), Camera()) + view = str(uuid4()) + for after in (0, 1): + peer = await start_local(peers, dict(view_id=view, after=after)) + if after: + assert released == [0] # Lost response/reconnect retires the old lease. + collected = [] + while sum(map(len, collected)) < len(payload): + response = await asyncio.wait_for(read_local(peers, peer, after), 1.2) + assert len(response) <= 196608 + offset = 0 + while offset < len(response): + kind = response[offset] + size = int.from_bytes(response[offset + 1 : offset + 5], "big") + data = response[offset + 5 : offset + 5 + size] + if kind == 0: + assert json.loads(data)["peer_id"] == peer + if kind == 2: + if data.startswith(b"MCF1"): + assert int.from_bytes(data[4:], "big") == len(payload) + else: + collected.append(data) + offset += size + 5 + assert b"".join(collected) == payload + acknowledge(peers, peer, after + 1) + assert subscriptions[-1][1].pending is None + await peers.close_all() + assert not peers.items and released == [0, 1] + assert subscriptions[0][0] == subscriptions[1][0] == view + + asyncio.run(run()) + + +@pytest.mark.parametrize("after", [True, -1, 2**53, "1"]) +def test_local_cursor_rejects_noncanonical_values(after): + with pytest.raises(ValueError): + admit_local(dict(view_id=str(uuid4()), after=after)) + + +def test_local_carrier_does_not_exceed_peer_capacity(): + async def run(): + class Peers: + items = {"a": {}, "b": {}} + + with pytest.raises(ValueError): + await start_local(Peers(), dict(view_id=str(uuid4()), after=0)) + + asyncio.run(run())