Fix onboard WebKit preview and archive board telemetry locally

This commit is contained in:
DCCONSTRUCTIONS
2026-09-08 01:37:52 +03:00
parent d8afb61229
commit f0802d2713
46 changed files with 2623 additions and 40 deletions
@@ -6,19 +6,17 @@ interface TelemetrySeriesProps {
ceiling?: number;
}
function linePoints(values: Array<number | null>, 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<number | null>, 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 (
<div className="system-telemetry-series">
<div>
@@ -45,7 +43,7 @@ export function TelemetrySeries({
aria-label={`${label}: ${value}`}
>
<path d="M0 37 H100" />
{points ? <polyline points={points} /> : null}
{segments.map((points,index)=><polyline key={index} points={points} />)}
</svg>
</div>
);
@@ -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<string,number|null>}|null;
definitions:Record<string,MetricDefinition>;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<BoardMonitor|null>(null),[error,setError]=useState('');
useEffect(()=>{
let active=true;let pending:AbortController|undefined;let timer:ReturnType<typeof setTimeout>|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<number|null>=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<values.length)values[index]=point[mode];}
return values;
}
export function monitorValue(value:number|null|undefined,unit=''){
if(value==null||!Number.isFinite(value))return 'Нет измерения';
if(unit==='bytes'||unit==='bytes/s'){
const labels=unit==='bytes'?['Б','КиБ','МиБ','ГиБ']:['Б/с','КиБ/с','МиБ/с','ГиБ/с'];let index=0;
while(value>=1024&&index<3){value/=1024;index++;}
return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})} ${labels[index]}`;
}
return `${value.toLocaleString('ru-RU',{maximumFractionDigits:1})}${unit?' '+unit:''}`;
}
@@ -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<string,string>={'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<number|null>(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 <div className="board-monitor">
<div><Button onClick={back}><Icon name="chevron-left"/>К аппарату</Button></div>
<SettingsCard title="Мониторинг системы БК" description={name} actions={<StatusBadge tone={value?.fresh&&!error?'success':'neutral'}>{value?.fresh&&!error?'Данные поступают':'Нет свежих данных'}</StatusBadge>}>
{latest&&<p>Измерено на БК: {time(latest.at)}. {value?.storage==='ready'?'Локальная история записывается.':'Запись локальной истории недоступна.'} {value?.backlog?'История догружается в Mission Core.':''}</p>}
{value?.database_bytes!=null&&<p>Архив на БК: {monitorValue(value.database_bytes,'bytes')} из {monitorValue(value.database_budget_bytes,'bytes')}. {value.archive_since?`Копия в Mission Core с ${time(value.archive_since)}.`:''}</p>}
</SettingsCard>
{!value?<ActivityIndicator label="Получаем мониторинг БК"/>:!value.available?<SettingsCard title="История БК пока не получена" description="Данные появятся после подключения бортового сборщика телеметрии."/>:<>
<SettingsCard title="История нагрузки">
<div className="board-monitor-controls">
<Select label="Ресурс" value={group} options={groups.map(label=>({value:label,label}))} onChange={next=>{setGroup(next);const first=Object.entries(definitions).find(([,item])=>item.group===next);if(first)setMetric(first[0]);}}/>
<Select label="Показатель" value={metric} options={metrics.map(([key,item])=>({value:key,label:`${item.resource} · ${item.label}`}))} onChange={setMetric}/>
<Select label="Период" value={String(window)} options={[{value:'900',label:'15 минут'},{value:'3600',label:'1 час'},{value:'21600',label:'6 часов'},{value:'86400',label:'24 часа'},{value:'604800',label:'7 дней'}]} onChange={next=>setWindow(Number(next))}/>
<Select label="Значение в интервале" value={mode} options={[{value:'max',label:'Максимум'},{value:'mean',label:'Среднее'},{value:'min',label:'Минимум'}]} onChange={next=>setMode(next as typeof mode)}/>
</div>
<div className="board-monitor-controls"><Button onClick={()=>setEnd(end===null?Date.now()/1000:null)}>{end===null?'Зафиксировать время':'В реальном времени'}</Button>{end!==null&&<TextField label="Конец периода" type="datetime-local" step="1" value={new Date((end-new Date(end*1000).getTimezoneOffset()*60)*1000).toISOString().slice(0,19)} onChange={event=>{const next=new Date(event.target.value).getTime()/1000;if(Number.isFinite(next))setEnd(next);}}/>}</div>
<TelemetrySeries label={definition?`${definition.resource} · ${definition.label}`:'Загрузка CPU'} resource="Последний интервал периода" value={monitorValue(value.series.at(-1)?.[mode],definition?.unit)} values={monitorSeries(value,mode)} ceiling={definition?.unit==='%'?100:undefined}/>
<p>{time(value.start)} {time(value.end)} · интервал {value.bucket_seconds} с. Разрывы означают отсутствие измерений.</p>
{!value.series.length&&<p>За выбранный период измерений нет.</p>}
</SettingsCard>
<SettingsCard title={`Последние измерения · ${group}`} description={latest?time(latest.at):'Ожидаем измерения'}>
<ResourceList aria-label="Показатели системы">{metrics.map(([key,item])=><li key={key}><ResourceRow title={`${item.resource} · ${item.label}`} description={latest?.values[key]==null?item.reason??'Счётчик недоступен':undefined} metadata={monitorValue(latest?.values[key],item.unit)}/></li>)}</ResourceList>
</SettingsCard>
<SettingsCard title="События системы и приложения" description="За выбранный период">
{value.events.length?<ResourceList aria-label="События БК">{value.events.map((event,index)=><li key={`${event.at}-${index}`}><ResourceRow title={eventLabels[event.code]??'Событие системы'} description={time(event.at)}/></li>)}</ResourceList>:<p>Событий не получено.</p>}
</SettingsCard>
</>}
<ToastStack items={error?[{id:'monitor-error',title:error,tone:'error',durationMs:null}]:[]} onDismiss={dismissError}/>
</div>;
}
@@ -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<string | null>(null);
const [sensorOpen, setSensorOpen] = useState(false);
const [monitorOpen,setMonitorOpen]=useState(false);
const [revoking, setRevoking] = useState<Vehicle | null>(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 <BoardMonitorView vehicle={detail.id} name={detail.name} back={()=>setMonitorOpen(false)}/>;
return <div className="fleet-workspace">
{fleet.error && <p role="alert">{fleet.error}</p>}
{!adding && error && <p role="alert">{error}</p>}
@@ -51,6 +54,7 @@ export function VehiclesWorkspace({ createRequest = 0 }: { createRequest?: numbe
<div><dt>Последняя связь</dt><dd>{detail.last_seen ? new Date(detail.last_seen * 1000).toLocaleString("ru-RU") : "Соединение ещё не получено"}</dd></div>
{detail.host && <><div><dt>Имя БК в системе</dt><dd>{detail.host.hostname}</dd></div><div><dt>Операционная система</dt><dd>{detail.host.os}</dd></div><div><dt>Архитектура</dt><dd>{detail.host.architecture}</dd></div><div><dt>Процессоры</dt><dd>{detail.host.cpus}</dd></div><div><dt>Память</dt><dd>{detail.host.memory_kib ? `${(detail.host.memory_kib / 1024 / 1024).toFixed(1)} ГиБ` : "Нет сведений"}</dd></div></>}
</dl>
<Button onClick={()=>setMonitorOpen(true)}><Icon name="activity"/>Мониторинг системы БК</Button>
{detail.enrollment !== "revoked" && <Button onClick={() => setRevoking(detail)}>Отозвать привязку БК</Button>}
</SettingsCard>}
<SettingsCard title="Устройства аппарата"><VehicleSensors onDetailChange={setSensorOpen} vehicleID={detail.id} enabled={!fleet.error && detail.enrollment === "paired" && detail.connectivity === "online"} /></SettingsCard>
@@ -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); }
@@ -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);
});
+3
View File
@@ -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")
@@ -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)
})
}
}
+159
View File
@@ -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})
})
}
@@ -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)
}
}
+1
View File
@@ -39,6 +39,7 @@ type PairState struct {
Revocations []CoreBinding `json:"revocations,omitempty"`
}
type Pairing struct {
Monitor *Monitor
Sensors *Sensors
DeviceEnrollment *DeviceEnrollment
mu sync.Mutex
@@ -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 {
+3
View File
@@ -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)
}
+201
View File
@@ -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()
+103
View File
@@ -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)
+353
View File
@@ -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}
+13
View File
@@ -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;
+118
View File
@@ -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<NOW()"
)
boundary = cursor.fetchone()[0]
if boundary:
cursor.execute(
"SELECT drop_chunks('monitor_samples',older_than=>%s::timestamptz)",
(boundary,),
)
cursor.execute("SELECT pg_database_size(current_database())")
size = cursor.fetchone()[0]
self.database_bytes = size
return size
+7 -2
View File
@@ -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 <noreply@example.invalid>
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))
@@ -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()
@@ -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.'
+6 -1
View File
@@ -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):
@@ -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
+1
View File
@@ -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
+1
View File
@@ -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
+4
View File
@@ -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
+54
View File
@@ -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
+1 -1
View File
@@ -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 <SensorWorkspace contributions={[xgridsK1SensorUi]} createRerunHost={createIsolatedRerunHost} transport={transport}/>;}
@@ -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?<SettingsCard title="Не удалось открыть представление" description="Вернитесь к устройствам или повторите открытие."><Button onClick={()=>this.setState({failed:false})}>Открыть повторно</Button></SettingsCard>:this.props.children;}
}
+4 -2
View File
@@ -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: <Icon name="activity" />, onSelect: () => openView("overview") }] : []}
items={views.filter(item => item.root === root).map(item => ({ id: item.id, label: item.label, icon: <Icon name={item.icon} /> }))} activeId={workspace.activeView ?? undefined} onItemChange={id => openView(id as ViewId)} footer={<span>Mission Core Node · {value?.version}</span>} />}
content={currentView && <ApplicationPanel title={currentView.label} eyebrow={currentRoot.label} expanded={workspace.contentExpanded} onExpandedChange={workspace.setContentExpanded} onClose={workspace.closeView}
utilityActions={[...(workspace.activeView === "ssh" ? [{ label: "Добавить доверенное устройство", icon: "plus" as const, onClick: () => setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}>{content}</ApplicationPanel>}
utilityActions={[...(workspace.activeView === "ssh" ? [{ label: "Добавить доверенное устройство", icon: "plus" as const, onClick: () => setAdding(true) }] : []), { label: "Обновить сведения", icon: "refresh", disabled: pending || environment.running, onClick: refreshAll }]}><RuntimeBoundary resetKey={workspace.activeView??undefined}>{content}</RuntimeBoundary></ApplicationPanel>}
stage={value ? <Home page={presentation.settings.pages.home} openView={openView} /> : <div className="node-stage" aria-busy={pending}>
<SettingsCard className="node-entry" title={pending ? "Подключаемся к ноде" : locked ? "Вход в Mission Core Node" : "Нода недоступна"}>
{pending ? <ActivityIndicator label="Получение сведений о ноде" /> : <p className="node-note">{locked ? "Подтвердите доступ в системном окне." : "Не удалось связаться со службой. Повторите подключение."}</p>}
@@ -64,4 +65,5 @@ function App() {
<HomeSettings open={!!value && settingsOpen} onClose={() => setSettingsOpen(false)} presentation={presentation} />
</>;
}
createRoot(document.getElementById("root")!).render(<App />);
observeRuntime();
createRoot(document.getElementById("root")!).render(<RuntimeBoundary><App /></RuntimeBoundary>);
@@ -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.
+4
View File
@@ -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<Response>;
read: (peer:string, after:number, signal:AbortSignal) => Promise<Response>;
};
enrollment?: import('./enrollment').EnrollmentTransport;
inventory: () => Promise<SensorInventory>;
subscribe?: (receive:(value:SensorInventory)=>void, unavailable:()=>void) => (()=>void);
@@ -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<string,unknown>,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<after)throw new Error('Invalid ACK');after=message.ack;});
const rrd=makeChannel(),camera=makeChannel();
const close=()=>{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};
}
@@ -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<void>((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();
+1 -1
View File
@@ -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",
@@ -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",
@@ -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()
@@ -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)
+390
View File
@@ -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<?", (time.time() - 7 * 86400,))
self.db.execute("DELETE FROM minutes WHERE at<?", (time.time() - 7 * 86400,))
self.db.execute("DELETE FROM events WHERE at<?", (time.time() - 7 * 86400,))
self.maintenance = time.time()
return dict(source_id=source, after=ack)
def query(self, node, metric="cpu.usage", window=900, end=None):
if not KEY.fullmatch(metric) or window not in (900, 3600, 21600, 86400, 604800):
raise ValueError("Invalid range")
end = time.time() if end is None else end
if not math.isfinite(end) or end < 0:
raise ValueError("Invalid time")
coarse = window >= 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<? "
"GROUP BY CAST(at/? AS INTEGER) ORDER BY 1",
(
width,
aggregate + "[0]",
aggregate + "[1]",
aggregate + "[2]",
aggregate + "[3]",
aggregate + "[3]",
node,
start,
end,
width,
),
).fetchall()
else:
points = self.db.execute(
"SELECT CAST(at/? AS INTEGER),min(json_extract(body,?)), "
"max(json_extract(body,?)),avg(json_extract(body,?)),"
"count(json_extract(body,?)) FROM samples "
"WHERE node=? 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)
+11 -1
View File
@@ -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}
+137
View File
@@ -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)
+5 -3
View File
@@ -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()
+15
View File
@@ -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"
+168
View File
@@ -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()
+2
View File
@@ -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)
+98
View File
@@ -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())