initial commit
This commit is contained in:
Martino Ferrari
@@ -0,0 +1,191 @@
|
||||
# RMon Wire Protocol
|
||||
|
||||
## Transport
|
||||
|
||||
- **TCP** over an SSH local port-forward tunnel
|
||||
- **Port**: configurable, default `7891`
|
||||
- One persistent TCP connection per UI client
|
||||
- Multiple clients can connect to the same agent simultaneously
|
||||
|
||||
## Framing
|
||||
|
||||
Every message (in both directions) uses the same length-prefix frame:
|
||||
|
||||
```
|
||||
┌─────────────────────┬────────────────────────────┐
|
||||
│ length : u32 LE │ payload : [u8; length] │
|
||||
└─────────────────────┴────────────────────────────┘
|
||||
```
|
||||
|
||||
- `length` is the byte count of the serialized payload (excludes the 4-byte length field itself)
|
||||
- `payload` is a `bincode`-serialized `ClientMessage` (client→agent) or `AgentMessage` (agent→client)
|
||||
- `bincode` configuration: little-endian, fixed-size integers, no trailing bytes
|
||||
|
||||
Maximum frame size: **16 MiB** (`length` values above this are a protocol error; close connection).
|
||||
|
||||
## Versioning
|
||||
|
||||
The `Handshake` / `HandshakeAck` exchange negotiates compatibility. If the agent does not support the client's version it sends `HandshakeAck { accepted: false }` and closes the connection. Both sides must be built from the same `rmon-common` version to guarantee `bincode` schema compatibility.
|
||||
|
||||
## Client → Agent Messages
|
||||
|
||||
```rust
|
||||
enum ClientMessage {
|
||||
/// Must be the first message sent. No other messages are valid before HandshakeAck.
|
||||
Handshake {
|
||||
protocol_version: u16, // current: 1
|
||||
client_version: String,
|
||||
},
|
||||
|
||||
/// Request the full list of signals the agent knows about.
|
||||
/// Agent replies with one SignalList message.
|
||||
ListSignals,
|
||||
|
||||
/// Subscribe to live data for the given signals.
|
||||
/// Agent begins sending DataBatch messages for these signals.
|
||||
/// `from_ns`: if Some, also request buffered/stored data from that timestamp forward
|
||||
/// before live data begins.
|
||||
Subscribe {
|
||||
ids: Vec<SignalId>,
|
||||
from_ns: Option<i64>,
|
||||
},
|
||||
|
||||
/// Stop live delivery for these signals. Does not affect storage.
|
||||
Unsubscribe {
|
||||
ids: Vec<SignalId>,
|
||||
},
|
||||
|
||||
/// Request stored historical data for a signal.
|
||||
/// Agent replies with one or more HistoryChunk messages (done=true on last chunk).
|
||||
/// `max_points`: agent may downsample to fit; 0 = no limit.
|
||||
QueryHistory {
|
||||
id: SignalId,
|
||||
from_ns: i64,
|
||||
to_ns: i64,
|
||||
max_points: u32,
|
||||
},
|
||||
|
||||
/// Start recording the given signals. Mode overrides any existing mode for these signals.
|
||||
StartRecording {
|
||||
ids: Vec<SignalId>,
|
||||
mode: StorageMode,
|
||||
},
|
||||
|
||||
/// Stop recording (data already written is kept).
|
||||
StopRecording {
|
||||
ids: Vec<SignalId>,
|
||||
},
|
||||
|
||||
/// Request current agent status (version, uptime, source states).
|
||||
GetStatus,
|
||||
}
|
||||
```
|
||||
|
||||
## Agent → Client Messages
|
||||
|
||||
```rust
|
||||
enum AgentMessage {
|
||||
/// Response to Handshake.
|
||||
HandshakeAck {
|
||||
protocol_version: u16,
|
||||
agent_version: String,
|
||||
accepted: bool,
|
||||
reject_reason: Option<String>,
|
||||
},
|
||||
|
||||
/// Full signal list, sent in response to ListSignals.
|
||||
/// Sent unsolicited again if the signal list changes (source added/removed).
|
||||
SignalList {
|
||||
signals: Vec<SignalInfo>,
|
||||
},
|
||||
|
||||
/// Live data batch for subscribed signals.
|
||||
/// The agent batches samples collected since the last send (target: ~50ms intervals).
|
||||
/// Multiple signals may be grouped into one message to reduce framing overhead.
|
||||
DataBatch {
|
||||
batches: Vec<SignalBatch>,
|
||||
},
|
||||
|
||||
/// One chunk of a historical query response.
|
||||
HistoryChunk {
|
||||
id: SignalId,
|
||||
samples: Vec<Sample>,
|
||||
done: bool, // true on the final chunk for this query
|
||||
},
|
||||
|
||||
/// Acknowledgement of StartRecording / StopRecording.
|
||||
RecordingStatus {
|
||||
id: SignalId,
|
||||
recording: bool,
|
||||
mode: Option<StorageMode>,
|
||||
},
|
||||
|
||||
/// Response to GetStatus.
|
||||
StatusReport {
|
||||
agent_version: String,
|
||||
hostname: String,
|
||||
uptime_secs: u64,
|
||||
sources: Vec<SourceStatus>,
|
||||
},
|
||||
|
||||
/// Unrecoverable error for a specific operation. Connection remains open.
|
||||
Error {
|
||||
context: String, // e.g. "QueryHistory" or "Subscribe"
|
||||
message: String,
|
||||
},
|
||||
}
|
||||
|
||||
struct SignalBatch {
|
||||
id: SignalId,
|
||||
samples: Vec<Sample>,
|
||||
}
|
||||
|
||||
struct SourceStatus {
|
||||
name: String,
|
||||
source_type: SourceType,
|
||||
connected: bool,
|
||||
last_sample_ns: Option<i64>,
|
||||
error: Option<String>,
|
||||
}
|
||||
```
|
||||
|
||||
## Session Lifecycle
|
||||
|
||||
```
|
||||
Client Agent
|
||||
│ │
|
||||
│──── Handshake ────────────────────► │
|
||||
│ │
|
||||
│◄─── HandshakeAck (accepted=true) ─── │
|
||||
│ │
|
||||
│──── ListSignals ───────────────────► │
|
||||
│◄─── SignalList ───────────────────── │
|
||||
│ │
|
||||
│──── Subscribe {ids, from_ns} ──────► │ ←── live data starts flowing
|
||||
│◄═══ DataBatch (repeated) ═══════════ │
|
||||
│ │
|
||||
│──── QueryHistory ──────────────────► │
|
||||
│◄─── HistoryChunk (done=false) ────── │
|
||||
│◄─── HistoryChunk (done=false) ────── │
|
||||
│◄─── HistoryChunk (done=true) ─────── │
|
||||
│ │
|
||||
│──── StartRecording ────────────────► │
|
||||
│◄─── RecordingStatus ──────────────── │
|
||||
│ │
|
||||
│ [TCP close / SSH tunnel drops] │
|
||||
│ │ ←── agent continues running,
|
||||
data continues to be stored
|
||||
```
|
||||
|
||||
## Batching Policy
|
||||
|
||||
The agent collects samples from all sources in a shared `tokio::broadcast` channel. A per-session task drains the channel and sends `DataBatch` messages at a target interval of **50 ms** (20 Hz UI update rate). If a batch would exceed 1 MiB, it is split and sent immediately.
|
||||
|
||||
High-frequency signals (e.g., 10 kHz UDP) are automatically decimated to 10× the batch rate (200 Hz) for live display; full-resolution data is still stored.
|
||||
|
||||
## Error Handling
|
||||
|
||||
- `Error` messages are non-fatal; the session continues
|
||||
- If the agent cannot fulfill a `QueryHistory` (signal not found, time range not stored), it sends `Error` then continues
|
||||
- If the TCP connection drops, the agent discards the session's subscriptions; no other sessions are affected
|
||||
- The UI reconnects automatically (with exponential backoff) if the tunnel drops
|
||||
Reference in New Issue
Block a user