# jamesolaitan/realtime-observability-monitor architecture

> Architecture diagram of the GitHub repository jamesolaitan/realtime-observability-monitor, generated by GitDiagram from its file tree, README and a sample of source files (last updated 2026-09-23). It is an AI-made overview: check details against the source.

- Interactive diagram: https://gitdiagram.com/jamesolaitan/realtime-observability-monitor
- Repository: https://github.com/jamesolaitan/realtime-observability-monitor

## Overview

rtobs is a low-overhead real-time observability monitor for QEC-shaped and robotics control workloads. Its principal workflow generates scheduled input, runs a decision workload, records bounded telemetry, exports observations, and receives them for delay and symptom checks; diagnostics can capture target-side process state. The README also describes in-process SPSC and cross-process shared-memory handoffs, raw/delta encoding, and UDP batching. Experiment and analysis tools compare configurations and report results. Some implementation details are documented but unsampled, so those subsystem links are shown only where the README or excerpts support them.

Main entry points: [`src/pipeline.cpp`](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/pipeline.cpp) (Decision pipeline), [`apps/rtobs_runner.cpp`](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp) (Runtime runner), [`experiments/run_experiment.py`](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/experiments/run_experiment.py) (Experiment orchestration), [`include/rtobs/collector.hpp`](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/include/rtobs/collector.hpp) (Telemetry collector)

## Diagram (Mermaid)

```mermaid
flowchart TD

subgraph group_execution["Control loop"]
  node_runner["Runtime runner<br/>[rtobs_runner.cpp]"]
  node_input["Input source<br/>[rtobs_runner.cpp]"]
  node_pipeline["Decision pipeline<br/>[pipeline.cpp]"]
  node_qec["QEC workload<br/>[qec_like.cpp]"]
  node_pid["PID controller<br/>[pid_controller.cpp]"]
  node_output["Output sink<br/>[rtobs_runner.cpp]"]
end

subgraph group_telemetry["Telemetry path"]
  node_spsc["SPSC handoff<br/>[ring_buffer.hpp]"]
  node_sharedmem[("Shared memory<br/>[shared_memory.cpp]")]
  node_collector["Telemetry collector<br/>[collector.hpp]"]
  node_shmcollector["SHM collector"]
  node_codec["Batch encoding<br/>[batch_codec.cpp]"]
  node_udp["UDP transport<br/>[udp_transport.cpp]"]
end

subgraph group_monitoring["Detection and diagnosis"]
  node_receiver["Telemetry receiver<br/>[rtobs_receiver.cpp]"]
  node_diagnostics["Diagnostic capture<br/>[diagnostics.cpp]"]
  node_diagapp["Diagnostic service<br/>[rtobs_diag.cpp]"]
end

subgraph group_evaluation["Evaluation"]
  node_experiment["Experiment orchestration<br/>[run_experiment.py]"]
  node_faults["Fault qualification"]
  node_analysis["Result analysis<br/>[analyze.py]"]
end

node_operator(("Experiment operator"))

node_operator -->|"starts run"| node_runner
node_runner -->|"configures source"| node_input
node_runner -->|"runs pipeline"| node_pipeline
node_runner -->|"starts collector"| node_collector
node_input -->|"supplies frames"| node_pipeline
node_pipeline -->|"runs workload"| node_qec
node_pipeline -->|"runs workload"| node_pid
node_pipeline -->|"sends output"| node_output
node_pipeline -->|"records telemetry"| node_spsc
node_pipeline -.->|"writes records"| node_sharedmem
node_collector -->|"reads records"| node_spsc
node_collector -->|"encodes batches"| node_codec
node_collector -->|"sends datagrams"| node_udp
node_shmcollector -.->|"reads records"| node_sharedmem
node_shmcollector -->|"encodes batches"| node_codec
node_shmcollector -->|"sends datagrams"| node_udp
node_udp -->|"delivers datagrams"| node_receiver
node_receiver -.->|"requests capture"| node_diagnostics
node_diagapp -->|"captures state"| node_diagnostics
node_experiment -->|"runs configurations"| node_runner
node_experiment -->|"starts receiver"| node_receiver
node_experiment -.->|"starts collector"| node_shmcollector
node_experiment -->|"analyzes results"| node_analysis
node_faults -->|"injects symptoms"| node_runner
node_faults -->|"checks detection"| node_receiver
node_faults -->|"checks capture"| node_diagapp

click node_runner "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp"
click node_input "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp"
click node_pipeline "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/pipeline.cpp"
click node_qec "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/workloads/qec_like.cpp"
click node_pid "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/workloads/pid_controller.cpp"
click node_output "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp"
click node_spsc "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/include/rtobs/ring_buffer.hpp"
click node_sharedmem "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/shared_memory.cpp"
click node_collector "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/include/rtobs/collector.hpp"
click node_shmcollector "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_shm_collector.cpp"
click node_codec "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/batch_codec.cpp"
click node_udp "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/udp_transport.cpp"
click node_receiver "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_receiver.cpp"
click node_diagnostics "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/diagnostics.cpp"
click node_diagapp "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_diag.cpp"
click node_experiment "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/experiments/run_experiment.py"
click node_faults "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/experiments/run_fault_qualification.py"
click node_analysis "https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/analysis/analyze.py"

classDef toneNeutral fill:#f8fafc,stroke:#334155,stroke-width:1.5px,color:#0f172a
classDef toneBlue fill:#dbeafe,stroke:#2563eb,stroke-width:1.5px,color:#172554
classDef toneAmber fill:#fef3c7,stroke:#d97706,stroke-width:1.5px,color:#78350f
classDef toneMint fill:#dcfce7,stroke:#16a34a,stroke-width:1.5px,color:#14532d
classDef toneRose fill:#ffe4e6,stroke:#e11d48,stroke-width:1.5px,color:#881337
classDef toneIndigo fill:#e0e7ff,stroke:#4f46e5,stroke-width:1.5px,color:#312e81
classDef toneTeal fill:#ccfbf1,stroke:#0f766e,stroke-width:1.5px,color:#134e4a
class node_runner,node_input,node_pipeline,node_qec,node_pid,node_output toneBlue
class node_spsc,node_sharedmem,node_collector,node_shmcollector,node_codec,node_udp toneAmber
class node_receiver,node_diagnostics,node_diagapp toneMint
class node_experiment,node_faults,node_analysis toneRose
class node_operator toneIndigo
```

## Components

### Control loop

- [Runtime runner](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp): `apps/rtobs_runner.cpp`
- [Input source](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp): `apps/rtobs_runner.cpp`
- [Decision pipeline](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/pipeline.cpp): `src/pipeline.cpp`
- [QEC workload](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/workloads/qec_like.cpp): `workloads/qec_like.cpp`
- [PID controller](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/workloads/pid_controller.cpp): `workloads/pid_controller.cpp`
- [Output sink](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_runner.cpp): `apps/rtobs_runner.cpp`

### Telemetry path

- [SPSC handoff](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/include/rtobs/ring_buffer.hpp): `include/rtobs/ring_buffer.hpp`
- [Shared memory](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/shared_memory.cpp): `src/shared_memory.cpp`
- [Telemetry collector](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/include/rtobs/collector.hpp): `include/rtobs/collector.hpp`
- [SHM collector](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_shm_collector.cpp): `apps/rtobs_shm_collector.cpp`
- [Batch encoding](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/batch_codec.cpp): `src/batch_codec.cpp`
- [UDP transport](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/udp_transport.cpp): `src/udp_transport.cpp`

### Detection and diagnosis

- [Telemetry receiver](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_receiver.cpp): `apps/rtobs_receiver.cpp`
- [Diagnostic capture](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/src/diagnostics.cpp): `src/diagnostics.cpp`
- [Diagnostic service](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/apps/rtobs_diag.cpp): `apps/rtobs_diag.cpp`

### Evaluation

- [Experiment orchestration](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/experiments/run_experiment.py): `experiments/run_experiment.py`
- [Fault qualification](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/experiments/run_fault_qualification.py): `experiments/run_fault_qualification.py`
- [Result analysis](https://github.com/jamesolaitan/realtime-observability-monitor/blob/main/analysis/analyze.py): `analysis/analyze.py`

### Other components

- **Experiment operator**

## Connections

- Experiment operator → Runtime runner: starts run
- Runtime runner → Input source: configures source
- Runtime runner → Decision pipeline: runs pipeline
- Runtime runner → Telemetry collector: starts collector
- Input source → Decision pipeline: supplies frames
- Decision pipeline → QEC workload: runs workload
- Decision pipeline → PID controller: runs workload
- Decision pipeline → Output sink: sends output
- Decision pipeline → SPSC handoff: records telemetry
- Decision pipeline → Shared memory: writes records
- Telemetry collector → SPSC handoff: reads records
- Telemetry collector → Batch encoding: encodes batches
- Telemetry collector → UDP transport: sends datagrams
- SHM collector → Shared memory: reads records
- SHM collector → Batch encoding: encodes batches
- SHM collector → UDP transport: sends datagrams
- UDP transport → Telemetry receiver: delivers datagrams
- Telemetry receiver → Diagnostic capture: requests capture
- Diagnostic service → Diagnostic capture: captures state
- Experiment orchestration → Runtime runner: runs configurations
- Experiment orchestration → Telemetry receiver: starts receiver
- Experiment orchestration → SHM collector: starts collector
- Experiment orchestration → Result analysis: analyzes results
- Fault qualification → Runtime runner: injects symptoms
- Fault qualification → Telemetry receiver: checks detection
- Fault qualification → Diagnostic service: checks capture