TelemetryOps

TelemetryOps is a small C++20 telemetry pipeline for accepting simulated satellite events, storing them in SQLite, calculating rolling health metrics, and evaluating configurable alert thresholds. It is split into three HTTP services so ingestion, aggregation, and alert evaluation can be run and inspected independently.

Architecture

flowchart LR
    C["Telemetry clients<br/>or load_test.py"] -->|"POST /telemetry"| I["Ingest<br/>:8081"]
    I -->|"writes events"| DB[("SQLite<br/>data/telemetry.db")]
    DB -->|"read-only queries"| A["Aggregator<br/>:8082"]
    CP["Control plane<br/>:8083"] -->|"polls /metrics every 5 s"| A
    U["Operators / monitoring"] -->|"config, watched IDs, alerts"| CP
    U -->|"Prometheus text endpoints"| I
    U -->|"Prometheus text endpoints"| A
    U -->|"Prometheus text endpoints"| CP

Loading

Component	Default port	Responsibility
ingest	8081	Validates telemetry, stores it in SQLite, and ignores duplicate event IDs.
aggregator	8082	Reads a time window from SQLite and calculates drop rate, latency percentiles, and average link quality.
controlplane	8083	Polls the aggregator every five seconds for watched satellites and evaluates configurable thresholds.
scripts/load_test.py	—	Generates synthetic telemetry and periodic degraded conditions using only the Python standard library.

Each service also exposes process-local request counters in the Prometheus text exposition format.

Prerequisites

• CMake 3.20 or newer
• A C++20 compiler
• SQLite 3 development headers and library
• Git and network access during the first CMake configuration
• Python 3 to run the integration test or optional load generator
• curl to run the examples below

CMake fetches these pinned dependencies during configuration:

• nlohmann/json v3.11.3
• spdlog v1.14.1
• cpp-httplib v0.15.3

On Debian or Ubuntu, the system prerequisites can be installed with:

sudo apt-get update
sudo apt-get install -y build-essential cmake git libsqlite3-dev python3 curl

On macOS, install CMake and SQLite with Homebrew after installing the Xcode Command Line Tools:

brew install cmake sqlite

Build

git clone https://github.com/derekk024/TelemetryOps.git
cd TelemetryOps

cmake -S . -B build -DCMAKE_BUILD_TYPE=Release
cmake --build build --parallel
mkdir -p data

The build creates:

build/services/ingest/ingest
build/services/aggregator/aggregator
build/services/controlplane/controlplane

Test

The automated suite covers both domain rules and the running service pipeline:

• JSON event validation and SQLite event-ID idempotency
• weighted packet-drop aggregation, percentile interpolation, and link-quality averaging
• alert threshold boundaries and aggregator failures
• real ingest and aggregator processes using a temporary SQLite database, including validation errors, duplicate suppression, satellite/window filtering, aggregate responses, and Prometheus counters

Build and run every test through CTest:

cmake -S . -B build -DCMAKE_BUILD_TYPE=Debug -DBUILD_TESTING=ON
cmake --build build --parallel
ctest --test-dir build --output-on-failure

GitHub Actions runs the same build and test sequence for pull requests and pushes to main.

Run locally

Start the services in order. The ingest service must create the database and schema before the aggregator opens that database in read-only mode.

Terminal 1:

./build/services/ingest/ingest

Terminal 2:

./build/services/aggregator/aggregator

Terminal 3:

./build/services/controlplane/controlplane

The optional positional arguments are:

ingest       [port=8081] [database=data/telemetry.db]
aggregator   [port=8082] [database=data/telemetry.db]
controlplane [port=8083] [aggregator_host=localhost] [aggregator_port=8082]

When using a custom database path, pass the same path to ingest and aggregator and create its parent directory first.

Send and inspect one event

now_ms="$(($(date +%s) * 1000))"

curl -i -X POST http://localhost:8081/telemetry \
  -H 'Content-Type: application/json' \
  --data "{
    \"event_id\":\"demo-001\",
    \"sat_id\":\"SAT-001\",
    \"ts_ms\":${now_ms},
    \"latency_ms\":350.5,
    \"dropped_packets\":8,
    \"sent_packets\":100,
    \"link_quality\":0.65
  }"

A valid event returns 202 Accepted:

{"inserted":true,"ok":true}

Sending another valid event with the same event_id also returns 202, but reports "inserted": false. After ingestion, query a rolling window:

curl 'http://localhost:8082/metrics?sat_id=SAT-001&window_s=600'

The control plane polls on a five-second interval. After at most one polling interval, inspect its last metrics and alerts:

curl 'http://localhost:8083/alerts?sat_id=SAT-001'

HTTP API

Ingest service — :8081

POST /telemetry

Accepts one JSON event:

{
  "event_id": "3cb2a080-4b90-40dd-8a22-929761cf37e0",
  "sat_id": "SAT-001",
  "ts_ms": 1721865600000,
  "latency_ms": 42.5,
  "dropped_packets": 1,
  "sent_packets": 100,
  "link_quality": 0.96
}

Validation requires:

• non-empty string event_id and sat_id
• integer ts_ms
• numeric latency_ms
• integer dropped_packets and sent_packets
• sent_packets > 0
• 0 <= dropped_packets <= sent_packets
• 0.0 <= link_quality <= 1.0

Malformed or invalid input returns 400. The event_id is the SQLite primary key, so retries are idempotent at storage time: a duplicate is ignored and returned as "inserted": false.

Other ingest endpoints

Method	Path	Behavior
GET	/health	Returns {"ok":true} while the process is serving requests.
GET	/ready	Returns {"ok":true} after startup opened and initialized the database.
GET	/metrics	Returns ingest and HTTP request counters in Prometheus text format.

Aggregator service — :8082

GET /metrics?sat_id=<id>&window_s=<seconds>

sat_id is required. window_s is optional, defaults to 600, and is clamped to at least one second.

The service selects events whose timestamps fall within the requested window and returns:

• event count
• drop_rate, calculated as total dropped packets divided by total sent packets
• interpolated latency_p50_ms and latency_p95_ms
• avg_link_quality

An empty window returns a count of zero and zero-valued aggregate fields. A missing sat_id returns 400.

Other aggregator endpoints

Method	Path	Behavior
GET	/health	Returns {"ok":true} while the process is serving requests.
GET	/ready	Returns {"ok":true} after startup opened the database read-only.
GET	/prom	Returns HTTP request counters in Prometheus text format.

Control-plane service — :8083

The control plane watches SAT-001 through SAT-005 by default. Every five seconds it requests the current aggregation window for each watched ID and stores the most recent metrics and evaluated alerts in memory.

Default thresholds:

Setting	Default	Alert when
latency_p95_ms	200.0	p95 latency is greater than the threshold
drop_rate	0.05	drop rate is greater than the threshold
min_link_quality	0.7	average link quality is less than the threshold
window_s	600	defines the aggregator query window

POST /config

Updates any supplied threshold fields:

curl -X POST http://localhost:8083/config \
  -H 'Content-Type: application/json' \
  --data '{
    "latency_p95_ms":250,
    "drop_rate":0.03,
    "min_link_quality":0.75,
    "window_s":300
  }'

The current implementation parses field types but does not enforce ranges on configuration values.

GET /watched and POST /watched

Read or replace the watched satellite IDs:

curl -X POST http://localhost:8083/watched \
  -H 'Content-Type: application/json' \
  --data '{"sats":["SAT-001","SAT-003"]}'

The replacement must contain at least one string ID.

GET /alerts?sat_id=<id>

Returns the last polled metrics, evaluated alerts, active thresholds, and poll counters for one ID. Before an ID has been polled, the response remains HTTP 200 and contains "metrics":{"ok":false,"error":"no data yet"}.

Other control-plane endpoints

Method	Path	Behavior
GET	/health	Reports the control-plane process as healthy.
GET	/ready	Calls the aggregator's /health; returns 503 if it is unreachable.
GET	/prom	Returns request, alert-evaluation, poll-cycle, and poll-failure counters in Prometheus text format.

Generate sample load

With the ingest service running:

python3 scripts/load_test.py \
  --host http://localhost:8081 \
  --qps 20 \
  --seconds 60 \
  --sats 5

All arguments are optional; the values above are their defaults. The generator chooses among SAT-001 through the requested satellite count and periodically simulates:

• elevated latency for SAT-001
• packet loss for SAT-002
• degraded link quality for SAT-003

The final line reports responses received by the generator. It is a paced traffic generator, not a rigorous throughput benchmark: request failures are suppressed, requests are sent synchronously, and achieved events per second can be lower than --qps.

Persistence and failure behavior

• Ingest creates the telemetry table and timestamp/satellite indexes and enables SQLite write-ahead logging.
• The ingest and aggregator SQLite connections use a five-second busy timeout.
• The aggregator opens the database read-only. Start it only after ingest has created the database and schema.
• Accepted events persist across service restarts. Duplicate event_id values are ignored with INSERT OR IGNORE.
• Thresholds, watched IDs, cached metrics and alerts, and all exported counters are process memory only and reset on restart.
• If the aggregator is unavailable, the control-plane poller increments poll_failures_total, keeps running, and reports 503 from /ready.
• alerts_total counts every polling evaluation that produces an alert. It does not count unique alert incidents.

Current limitations

• No authentication, authorization, or TLS
• No retention policy or database cleanup
• No durable control-plane configuration or alert history
• No alert delivery mechanism beyond the HTTP response and counters
• No container or multi-host deployment configuration
• Aggregation loads all matching window rows into memory before calculating percentiles
• One SQLite connection per data service, without an application-level connection pool

Project structure

.
├── .github/workflows/ci.yml
├── CMakeLists.txt
├── common/
│   └── include/common/
│       ├── aggregation.hpp
│       ├── alerts.hpp
│       └── telemetry.hpp
├── scripts/
│   └── load_test.py
├── services/
│   ├── ingest/
│   │   ├── CMakeLists.txt
│   │   └── main.cpp
│   ├── aggregator/
│   │   ├── CMakeLists.txt
│   │   └── main.cpp
│   └── controlplane/
│       ├── CMakeLists.txt
│       └── main.cpp
└── tests/
    ├── CMakeLists.txt
    ├── domain_tests.cpp
    └── service_pipeline_test.py

License

This repository does not currently include a license.