From a4edacc4713095ca441a7ed733a29bafa5a6e4ed Mon Sep 17 00:00:00 2001 From: Frank Kronewitter Date: Fri, 24 Jul 2026 20:36:23 -0700 Subject: [PATCH] Add TelemetryOps tests and CI --- .github/workflows/ci.yml | 36 ++++ .gitignore | 3 + CMakeLists.txt | 7 +- README.md | 55 ++++-- common/include/common/aggregation.hpp | 69 ++++++++ common/include/common/alerts.hpp | 62 +++++++ common/include/common/telemetry.hpp | 151 +++++++++++++++++ services/aggregator/main.cpp | 54 ++---- services/controlplane/main.cpp | 45 +---- services/ingest/main.cpp | 103 +----------- tests/CMakeLists.txt | 26 +++ tests/domain_tests.cpp | 190 +++++++++++++++++++++ tests/service_pipeline_test.py | 232 ++++++++++++++++++++++++++ 13 files changed, 839 insertions(+), 194 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 common/include/common/aggregation.hpp create mode 100644 common/include/common/alerts.hpp create mode 100644 tests/CMakeLists.txt create mode 100644 tests/domain_tests.cpp create mode 100644 tests/service_pipeline_test.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..3ad0607 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,36 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +permissions: + contents: read + +jobs: + build-and-test: + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - name: Check out repository + uses: actions/checkout@v6 + + - name: Install build dependencies + run: | + sudo apt-get update + sudo apt-get install --yes libsqlite3-dev ninja-build + + - name: Configure + run: > + cmake -S . -B build -G Ninja + -DCMAKE_BUILD_TYPE=Release + -DBUILD_TESTING=ON + + - name: Build + run: cmake --build build --parallel + + - name: Test + run: ctest --test-dir build --output-on-failure diff --git a/.gitignore b/.gitignore index c1e1296..5a9121e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,7 @@ build/ +build-*/ +__pycache__/ +*.py[cod] data/*.db *.db *.sqlite diff --git a/CMakeLists.txt b/CMakeLists.txt index d23335c..3d0483a 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -5,6 +5,7 @@ project(leo_telemetry LANGUAGES CXX) set(CMAKE_CXX_STANDARD 20) set(CMAKE_CXX_STANDARD_REQUIRED ON) +include(CTest) include(FetchContent) FetchContent_Declare( @@ -41,4 +42,8 @@ target_link_libraries(common INTERFACE add_subdirectory(services/ingest) add_subdirectory(services/aggregator) -add_subdirectory(services/controlplane) \ No newline at end of file +add_subdirectory(services/controlplane) + +if(BUILD_TESTING) + add_subdirectory(tests) +endif() diff --git a/README.md b/README.md index 8c58038..3c5ee32 100644 --- a/README.md +++ b/README.md @@ -1,5 +1,7 @@ # TelemetryOps +[![CI](https://github.com/derekk024/TelemetryOps/actions/workflows/ci.yml/badge.svg)](https://github.com/derekk024/TelemetryOps/actions/workflows/ci.yml) + 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 @@ -31,7 +33,7 @@ Each service also exposes process-local request counters in the Prometheus text - A C++20 compiler - SQLite 3 development headers and library - Git and network access during the first CMake configuration -- Python 3 to use the optional load generator +- Python 3 to run the integration test or optional load generator - `curl` to run the examples below CMake fetches these pinned dependencies during configuration: @@ -72,6 +74,25 @@ 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: + +```bash +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. @@ -293,31 +314,37 @@ The final line reports responses received by the generator. It is a paced traffi - 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 automated test suite or continuous-integration workflow - 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 -- `common/include/common/telemetry.hpp` is currently a placeholder; the services do not share a domain-model implementation ## Project structure ```text . +├── .github/workflows/ci.yml ├── CMakeLists.txt ├── common/ -│ └── include/common/telemetry.hpp +│ └── 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 +├── 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 diff --git a/common/include/common/aggregation.hpp b/common/include/common/aggregation.hpp new file mode 100644 index 0000000..e74738b --- /dev/null +++ b/common/include/common/aggregation.hpp @@ -0,0 +1,69 @@ +#pragma once + +#include +#include +#include + +namespace telemetry_ops { + +struct TelemetryRow { + double latency_ms; + int dropped; + int sent; + double link_quality; +}; + +struct AggregateMetrics { + std::size_t count = 0; + double drop_rate = 0.0; + double latency_p50_ms = 0.0; + double latency_p95_ms = 0.0; + double avg_link_quality = 0.0; +}; + +inline double percentile(std::vector values, double percentile_value) { + if (values.empty()) return 0.0; + + std::sort(values.begin(), values.end()); + const double index = + (percentile_value / 100.0) * static_cast(values.size() - 1); + const auto lower_index = static_cast(index); + const double fraction = index - static_cast(lower_index); + + if (lower_index + 1 < values.size()) { + return values[lower_index] * (1.0 - fraction) + + values[lower_index + 1] * fraction; + } + return values[lower_index]; +} + +inline AggregateMetrics aggregate(const std::vector& rows) { + AggregateMetrics metrics; + metrics.count = rows.size(); + if (rows.empty()) return metrics; + + long long sum_dropped = 0; + long long sum_sent = 0; + double sum_link_quality = 0.0; + std::vector latencies; + latencies.reserve(rows.size()); + + for (const auto& row : rows) { + sum_dropped += row.dropped; + sum_sent += row.sent; + sum_link_quality += row.link_quality; + latencies.push_back(row.latency_ms); + } + + if (sum_sent > 0) { + metrics.drop_rate = + static_cast(sum_dropped) / static_cast(sum_sent); + } + metrics.latency_p50_ms = percentile(latencies, 50.0); + metrics.latency_p95_ms = percentile(latencies, 95.0); + metrics.avg_link_quality = + sum_link_quality / static_cast(rows.size()); + return metrics; +} + +} // namespace telemetry_ops diff --git a/common/include/common/alerts.hpp b/common/include/common/alerts.hpp new file mode 100644 index 0000000..920e104 --- /dev/null +++ b/common/include/common/alerts.hpp @@ -0,0 +1,62 @@ +#pragma once + +#include + +namespace telemetry_ops { + +using json = nlohmann::json; + +struct Thresholds { + double latency_p95_ms = 200.0; + double drop_rate = 0.05; + double min_link_quality = 0.7; + int window_s = 600; +}; + +inline json eval_alerts(const json& metrics, const Thresholds& thresholds) { + json alerts = json::array(); + + if (!metrics.contains("ok") || !metrics["ok"].is_boolean() || + !metrics["ok"].get()) { + alerts.push_back({ + {"severity", "HIGH"}, + {"type", "AGGREGATOR_ERROR"}, + {"message", "metrics not ok"}, + }); + return alerts; + } + + if (metrics.value("count", 0) == 0) return alerts; + + const double latency_p95_ms = metrics.value("latency_p95_ms", 0.0); + const double drop_rate = metrics.value("drop_rate", 0.0); + const double link_quality = metrics.value("avg_link_quality", 0.0); + + if (latency_p95_ms > thresholds.latency_p95_ms) { + alerts.push_back({ + {"severity", "MED"}, + {"type", "LATENCY_P95"}, + {"value", latency_p95_ms}, + {"threshold", thresholds.latency_p95_ms}, + }); + } + if (drop_rate > thresholds.drop_rate) { + alerts.push_back({ + {"severity", "HIGH"}, + {"type", "DROP_RATE"}, + {"value", drop_rate}, + {"threshold", thresholds.drop_rate}, + }); + } + if (link_quality < thresholds.min_link_quality) { + alerts.push_back({ + {"severity", "MED"}, + {"type", "LINK_QUALITY"}, + {"value", link_quality}, + {"threshold", thresholds.min_link_quality}, + }); + } + return alerts; +} + +} // namespace telemetry_ops diff --git a/common/include/common/telemetry.hpp b/common/include/common/telemetry.hpp index e69de29..feee0c7 100644 --- a/common/include/common/telemetry.hpp +++ b/common/include/common/telemetry.hpp @@ -0,0 +1,151 @@ +#pragma once + +#include +#include + +#include +#include +#include + +namespace telemetry_ops { + +using json = nlohmann::json; + +struct EventStore { + sqlite3* db = nullptr; + + explicit EventStore(const std::string& path) { + if (sqlite3_open(path.c_str(), &db) != SQLITE_OK) { + std::string msg = db ? sqlite3_errmsg(db) : "unknown"; + if (db) sqlite3_close(db); + db = nullptr; + throw std::runtime_error("sqlite open failed: " + msg); + } + + sqlite3_busy_timeout(db, 5000); + exec("PRAGMA busy_timeout=5000;"); + exec("PRAGMA journal_mode=WAL;"); + exec(R"sql( + CREATE TABLE IF NOT EXISTS telemetry ( + event_id TEXT PRIMARY KEY, + sat_id TEXT NOT NULL, + ts_ms INTEGER NOT NULL, + latency_ms REAL NOT NULL, + dropped_packets INTEGER NOT NULL, + sent_packets INTEGER NOT NULL, + link_quality REAL NOT NULL + ); + )sql"); + exec("CREATE INDEX IF NOT EXISTS idx_telemetry_ts ON telemetry(ts_ms);"); + exec("CREATE INDEX IF NOT EXISTS idx_telemetry_sat ON telemetry(sat_id);"); + } + + ~EventStore() { + if (db) sqlite3_close(db); + } + + EventStore(const EventStore&) = delete; + EventStore& operator=(const EventStore&) = delete; + + void exec(const std::string& sql) { + char* err = nullptr; + if (sqlite3_exec(db, sql.c_str(), nullptr, nullptr, &err) != SQLITE_OK) { + std::string msg = err ? err : "unknown"; + sqlite3_free(err); + throw std::runtime_error("sqlite exec failed: " + msg); + } + } + + bool insert_event(const json& event) { + const char* sql = + "INSERT OR IGNORE INTO telemetry(event_id,sat_id,ts_ms,latency_ms,dropped_packets,sent_packets,link_quality) " + "VALUES(?,?,?,?,?,?,?);"; + sqlite3_stmt* statement = nullptr; + if (sqlite3_prepare_v2(db, sql, -1, &statement, nullptr) != SQLITE_OK) { + throw std::runtime_error( + std::string("sqlite prepare failed: ") + sqlite3_errmsg(db)); + } + + sqlite3_bind_text( + statement, 1, event["event_id"].get().c_str(), -1, + SQLITE_TRANSIENT); + sqlite3_bind_text( + statement, 2, event["sat_id"].get().c_str(), -1, + SQLITE_TRANSIENT); + sqlite3_bind_int64(statement, 3, event["ts_ms"].get()); + sqlite3_bind_double(statement, 4, event["latency_ms"].get()); + sqlite3_bind_int(statement, 5, event["dropped_packets"].get()); + sqlite3_bind_int(statement, 6, event["sent_packets"].get()); + sqlite3_bind_double(statement, 7, event["link_quality"].get()); + + const int result = sqlite3_step(statement); + sqlite3_finalize(statement); + if (result != SQLITE_DONE) { + throw std::runtime_error( + std::string("sqlite step failed: ") + sqlite3_errmsg(db)); + } + + return sqlite3_changes(db) > 0; + } +}; + +inline bool validate_event(const json& event, std::string& error) { + const char* required_fields[] = { + "event_id", "sat_id", "ts_ms", "latency_ms", + "dropped_packets", "sent_packets", "link_quality", + }; + for (const auto* field : required_fields) { + if (!event.contains(field)) { + error = std::string("missing field: ") + field; + return false; + } + } + + if (!event["event_id"].is_string() || + event["event_id"].get().empty()) { + error = "event_id invalid"; + return false; + } + if (!event["sat_id"].is_string() || + event["sat_id"].get().empty()) { + error = "sat_id invalid"; + return false; + } + if (!event["ts_ms"].is_number_integer()) { + error = "ts_ms must be int64"; + return false; + } + if (!event["latency_ms"].is_number()) { + error = "latency_ms must be number"; + return false; + } + if (!event["dropped_packets"].is_number_integer()) { + error = "dropped_packets must be int"; + return false; + } + if (!event["sent_packets"].is_number_integer()) { + error = "sent_packets must be int"; + return false; + } + + const int sent = event["sent_packets"].get(); + const int dropped = event["dropped_packets"].get(); + if (sent <= 0) { + error = "sent_packets must be > 0"; + return false; + } + if (dropped < 0 || dropped > sent) { + error = "dropped_packets must be in [0,sent_packets]"; + return false; + } + + const double link_quality = event["link_quality"].get(); + if (link_quality < 0.0 || link_quality > 1.0) { + error = "link_quality out of range [0,1]"; + return false; + } + + return true; +} + +} // namespace telemetry_ops diff --git a/services/aggregator/main.cpp b/services/aggregator/main.cpp index bff0d55..19cbc30 100644 --- a/services/aggregator/main.cpp +++ b/services/aggregator/main.cpp @@ -1,9 +1,9 @@ +#include #include #include #include #include -#include #include #include #include @@ -41,14 +41,8 @@ struct SqliteRO { } } - struct Row { - double latency_ms; - int dropped; - int sent; - double link_quality; - }; - - std::vector select_rows(const std::string& sat_id, std::int64_t min_ts_ms) { + std::vector select_rows( + const std::string& sat_id, std::int64_t min_ts_ms) { const char* sql = "SELECT latency_ms, dropped_packets, sent_packets, link_quality " "FROM telemetry WHERE sat_id = ? AND ts_ms >= ?;"; @@ -61,11 +55,11 @@ struct SqliteRO { sqlite3_bind_text(stmt, 1, sat_id.c_str(), -1, SQLITE_TRANSIENT); sqlite3_bind_int64(stmt, 2, min_ts_ms); - std::vector out; + std::vector out; while (true) { int rc = sqlite3_step(stmt); if (rc == SQLITE_ROW) { - Row r; + telemetry_ops::TelemetryRow r; r.latency_ms = sqlite3_column_double(stmt, 0); r.dropped = sqlite3_column_int(stmt, 1); r.sent = sqlite3_column_int(stmt, 2); @@ -83,16 +77,6 @@ struct SqliteRO { } }; -static double percentile(std::vector v, double p) { - if (v.empty()) return 0.0; - std::sort(v.begin(), v.end()); - double idx = (p / 100.0) * (v.size() - 1); - std::size_t i = static_cast(idx); - double frac = idx - static_cast(i); - if (i + 1 < v.size()) return v[i] * (1.0 - frac) + v[i + 1] * frac; - return v[i]; -} - static std::atomic g_health{0}, g_ready{0}, g_prom{0}, g_query{0}; static std::string prom_metrics() { @@ -149,31 +133,17 @@ int main(int argc, char** argv) { try { auto rows = db.select_rows(sat_id, min_ts); - long long sum_dropped = 0; - long long sum_sent = 0; - double sum_lq = 0.0; - std::vector lat; - lat.reserve(rows.size()); - - for (auto& r : rows) { - sum_dropped += r.dropped; - sum_sent += r.sent; - sum_lq += r.link_quality; - lat.push_back(r.latency_ms); - } - - double drop_rate = (sum_sent > 0) ? (double)sum_dropped / (double)sum_sent : 0.0; - double avg_lq = (!rows.empty()) ? sum_lq / (double)rows.size() : 0.0; + const auto metrics = telemetry_ops::aggregate(rows); json out = { {"ok", true}, {"sat_id", sat_id}, {"window_s", window_s}, - {"count", (int)rows.size()}, - {"drop_rate", drop_rate}, - {"latency_p50_ms", percentile(lat, 50.0)}, - {"latency_p95_ms", percentile(lat, 95.0)}, - {"avg_link_quality", avg_lq} + {"count", static_cast(metrics.count)}, + {"drop_rate", metrics.drop_rate}, + {"latency_p50_ms", metrics.latency_p50_ms}, + {"latency_p95_ms", metrics.latency_p95_ms}, + {"avg_link_quality", metrics.avg_link_quality} }; res.set_content(out.dump(), "application/json"); @@ -190,4 +160,4 @@ int main(int argc, char** argv) { spdlog::error("aggregator fatal: {}", e.what()); return 1; } -} \ No newline at end of file +} diff --git a/services/controlplane/main.cpp b/services/controlplane/main.cpp index e65bb45..9f4b3cb 100644 --- a/services/controlplane/main.cpp +++ b/services/controlplane/main.cpp @@ -1,3 +1,4 @@ +#include #include #include #include @@ -14,42 +15,6 @@ using json = nlohmann::json; -struct Thresholds { - double latency_p95_ms = 200.0; - double drop_rate = 0.05; - double min_link_quality = 0.7; - int window_s = 600; -}; - -static json eval_alerts(const json& metrics, const Thresholds& t) { - json alerts = json::array(); - - if (!metrics.contains("ok") || !metrics["ok"].is_boolean() || !metrics["ok"].get()) { - alerts.push_back({{"severity","HIGH"},{"type","AGGREGATOR_ERROR"},{"message","metrics not ok"}}); - return alerts; - } - - int count = metrics.value("count", 0); - if (count == 0) { - return alerts; - } - - double p95 = metrics.value("latency_p95_ms", 0.0); - double dr = metrics.value("drop_rate", 0.0); - double lq = metrics.value("avg_link_quality", 0.0); - - if (p95 > t.latency_p95_ms) { - alerts.push_back({{"severity","MED"},{"type","LATENCY_P95"},{"value",p95},{"threshold",t.latency_p95_ms}}); - } - if (dr > t.drop_rate) { - alerts.push_back({{"severity","HIGH"},{"type","DROP_RATE"},{"value",dr},{"threshold",t.drop_rate}}); - } - if (lq < t.min_link_quality) { - alerts.push_back({{"severity","MED"},{"type","LINK_QUALITY"},{"value",lq},{"threshold",t.min_link_quality}}); - } - return alerts; -} - static std::atomic g_health{0}, g_ready{0}, g_config{0}, g_alerts{0}, g_prom{0}, g_watched{0}; static std::mutex g_alert_mu; @@ -97,7 +62,7 @@ int main(int argc, char** argv) { std::string aggregator_host = (argc > 2) ? argv[2] : std::string("localhost"); int aggregator_port = (argc > 3) ? std::atoi(argv[3]) : 8082; - Thresholds thresholds; + telemetry_ops::Thresholds thresholds; std::mutex thresholds_mu; std::vector watched = {"SAT-001","SAT-002","SAT-003","SAT-004","SAT-005"}; @@ -114,7 +79,7 @@ int main(int argc, char** argv) { while (!stop.load()) { g_poll_cycles++; - Thresholds t; + telemetry_ops::Thresholds t; { std::lock_guard lock(thresholds_mu); t = thresholds; @@ -143,7 +108,7 @@ int main(int argc, char** argv) { continue; } - json alerts = eval_alerts(metrics, t); + json alerts = telemetry_ops::eval_alerts(metrics, t); { std::lock_guard lock(g_state_mu); @@ -259,7 +224,7 @@ int main(int argc, char** argv) { } std::string sat_id = req.get_param_value("sat_id"); - Thresholds t; + telemetry_ops::Thresholds t; { std::lock_guard lock(thresholds_mu); t = thresholds; diff --git a/services/ingest/main.cpp b/services/ingest/main.cpp index 0e1c141..83c14ac 100644 --- a/services/ingest/main.cpp +++ b/services/ingest/main.cpp @@ -1,106 +1,15 @@ +#include #include #include #include -#include - -#include -#include -#include -#include #include +#include #include +#include using json = nlohmann::json; -struct Sqlite { - sqlite3* db = nullptr; - - explicit Sqlite(const std::string& path) { - if (sqlite3_open(path.c_str(), &db) != SQLITE_OK) { - std::string msg = db ? sqlite3_errmsg(db) : "unknown"; - if (db) sqlite3_close(db); - db = nullptr; - throw std::runtime_error("sqlite open failed: " + msg); - } - - sqlite3_busy_timeout(db, 5000); - - exec("PRAGMA busy_timeout=5000;"); - exec("PRAGMA journal_mode=WAL;"); - exec(R"sql( - CREATE TABLE IF NOT EXISTS telemetry ( - event_id TEXT PRIMARY KEY, - sat_id TEXT NOT NULL, - ts_ms INTEGER NOT NULL, - latency_ms REAL NOT NULL, - dropped_packets INTEGER NOT NULL, - sent_packets INTEGER NOT NULL, - link_quality REAL NOT NULL - ); - )sql"); - exec("CREATE INDEX IF NOT EXISTS idx_telemetry_ts ON telemetry(ts_ms);"); - exec("CREATE INDEX IF NOT EXISTS idx_telemetry_sat ON telemetry(sat_id);"); - } - - ~Sqlite() { if (db) sqlite3_close(db); } - - void exec(const std::string& sql) { - char* err = nullptr; - if (sqlite3_exec(db, sql.c_str(), nullptr, nullptr, &err) != SQLITE_OK) { - std::string msg = err ? err : "unknown"; - sqlite3_free(err); - throw std::runtime_error("sqlite exec failed: " + msg); - } - } - - bool insert_event(const json& j) { - const char* sql = - "INSERT OR IGNORE INTO telemetry(event_id,sat_id,ts_ms,latency_ms,dropped_packets,sent_packets,link_quality) " - "VALUES(?,?,?,?,?,?,?);"; - sqlite3_stmt* stmt = nullptr; - if (sqlite3_prepare_v2(db, sql, -1, &stmt, nullptr) != SQLITE_OK) { - throw std::runtime_error(std::string("sqlite prepare failed: ") + sqlite3_errmsg(db)); - } - - sqlite3_bind_text(stmt, 1, j["event_id"].get().c_str(), -1, SQLITE_TRANSIENT); - sqlite3_bind_text(stmt, 2, j["sat_id"].get().c_str(), -1, SQLITE_TRANSIENT); - sqlite3_bind_int64(stmt, 3, j["ts_ms"].get()); - sqlite3_bind_double(stmt, 4, j["latency_ms"].get()); - sqlite3_bind_int(stmt, 5, j["dropped_packets"].get()); - sqlite3_bind_int(stmt, 6, j["sent_packets"].get()); - sqlite3_bind_double(stmt, 7, j["link_quality"].get()); - - int rc = sqlite3_step(stmt); - sqlite3_finalize(stmt); - if (rc != SQLITE_DONE) throw std::runtime_error(std::string("sqlite step failed: ") + sqlite3_errmsg(db)); - - return sqlite3_changes(db) > 0; - } -}; - -static bool validate_event(const json& j, std::string& err) { - const char* req[] = {"event_id","sat_id","ts_ms","latency_ms","dropped_packets","sent_packets","link_quality"}; - for (auto k : req) if (!j.contains(k)) { err = std::string("missing field: ") + k; return false; } - - if (!j["event_id"].is_string() || j["event_id"].get().empty()) { err = "event_id invalid"; return false; } - if (!j["sat_id"].is_string() || j["sat_id"].get().empty()) { err = "sat_id invalid"; return false; } - if (!j["ts_ms"].is_number_integer()) { err = "ts_ms must be int64"; return false; } - if (!j["latency_ms"].is_number()) { err = "latency_ms must be number"; return false; } - if (!j["dropped_packets"].is_number_integer()) { err = "dropped_packets must be int"; return false; } - if (!j["sent_packets"].is_number_integer()) { err = "sent_packets must be int"; return false; } - - int sent = j["sent_packets"].get(); - int dropped = j["dropped_packets"].get(); - if (sent <= 0) { err = "sent_packets must be > 0"; return false; } - if (dropped < 0 || dropped > sent) { err = "dropped_packets must be in [0,sent_packets]"; return false; } - - double lq = j["link_quality"].get(); - if (lq < 0.0 || lq > 1.0) { err = "link_quality out of range [0,1]"; return false; } - - return true; -} - static std::atomic g_inserted{0}; static std::atomic g_duplicates{0}; static std::atomic g_health{0}, g_ready{0}, g_telemetry{0}, g_metrics{0}; @@ -123,7 +32,7 @@ int main(int argc, char** argv) { int port = (argc > 1) ? std::atoi(argv[1]) : 8081; std::string db_path = (argc > 2) ? argv[2] : std::string("data/telemetry.db"); - Sqlite db(db_path); + telemetry_ops::EventStore db(db_path); httplib::Server svr; svr.Get("/health", [](const httplib::Request&, httplib::Response& res) { @@ -147,7 +56,7 @@ int main(int argc, char** argv) { try { auto j = json::parse(req.body); std::string err; - if (!validate_event(j, err)) { + if (!telemetry_ops::validate_event(j, err)) { res.status = 400; res.set_content(json{{"ok",false},{"error",err}}.dump(), "application/json"); return; @@ -172,4 +81,4 @@ int main(int argc, char** argv) { spdlog::info("ingest listening on {} db={}", port, db_path); svr.listen("0.0.0.0", port); return 0; -} \ No newline at end of file +} diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt new file mode 100644 index 0000000..326ac07 --- /dev/null +++ b/tests/CMakeLists.txt @@ -0,0 +1,26 @@ +add_executable(domain_tests domain_tests.cpp) +target_link_libraries(domain_tests PRIVATE common) + +add_test( + NAME domain_event_validation_and_idempotency + COMMAND domain_tests event +) +add_test( + NAME domain_aggregation + COMMAND domain_tests aggregation +) +add_test( + NAME domain_alert_evaluation + COMMAND domain_tests alerts +) + +find_package(Python3 REQUIRED COMPONENTS Interpreter) +add_test( + NAME service_pipeline_integration + COMMAND + ${Python3_EXECUTABLE} + ${CMAKE_CURRENT_SOURCE_DIR}/service_pipeline_test.py + --ingest $ + --aggregator $ +) +set_tests_properties(service_pipeline_integration PROPERTIES TIMEOUT 30) diff --git a/tests/domain_tests.cpp b/tests/domain_tests.cpp new file mode 100644 index 0000000..bbbfd29 --- /dev/null +++ b/tests/domain_tests.cpp @@ -0,0 +1,190 @@ +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using json = nlohmann::json; + +namespace { + +void expect(bool condition, const std::string& message) { + if (!condition) throw std::runtime_error(message); +} + +void expect_near(double actual, double expected, double tolerance, + const std::string& message) { + if (std::abs(actual - expected) > tolerance) { + throw std::runtime_error( + message + ": expected " + std::to_string(expected) + ", got " + + std::to_string(actual)); + } +} + +json valid_event() { + return json::parse(R"json({ + "event_id": "event-001", + "sat_id": "SAT-TEST", + "ts_ms": 1721865600000, + "latency_ms": 42.5, + "dropped_packets": 2, + "sent_packets": 100, + "link_quality": 0.91 + })json"); +} + +void test_event_validation_and_idempotency() { + std::string error; + const json event = valid_event(); + expect(telemetry_ops::validate_event(event, error), + "valid parsed event should pass validation"); + + json missing_field = event; + missing_field.erase("sat_id"); + expect(!telemetry_ops::validate_event(missing_field, error), + "missing sat_id should fail validation"); + expect(error == "missing field: sat_id", + "missing field should report the field name"); + + json fractional_timestamp = event; + fractional_timestamp["ts_ms"] = 1.5; + expect(!telemetry_ops::validate_event(fractional_timestamp, error), + "fractional timestamp should fail validation"); + + json zero_sent = event; + zero_sent["sent_packets"] = 0; + zero_sent["dropped_packets"] = 0; + expect(!telemetry_ops::validate_event(zero_sent, error), + "zero sent packets should fail validation"); + + json excessive_drops = event; + excessive_drops["dropped_packets"] = 101; + expect(!telemetry_ops::validate_event(excessive_drops, error), + "drops greater than sent packets should fail validation"); + + json invalid_link_quality = event; + invalid_link_quality["link_quality"] = 1.01; + expect(!telemetry_ops::validate_event(invalid_link_quality, error), + "link quality above one should fail validation"); + + telemetry_ops::EventStore store(":memory:"); + expect(store.insert_event(event), "first event insert should succeed"); + expect(!store.insert_event(event), + "duplicate event ID should be ignored idempotently"); + + sqlite3_stmt* statement = nullptr; + expect(sqlite3_prepare_v2(store.db, "SELECT COUNT(*) FROM telemetry;", -1, + &statement, nullptr) == SQLITE_OK, + "row-count query should prepare"); + expect(sqlite3_step(statement) == SQLITE_ROW, + "row-count query should return a row"); + expect(sqlite3_column_int(statement, 0) == 1, + "duplicate insert should leave one stored row"); + sqlite3_finalize(statement); +} + +void test_aggregation() { + const std::vector rows = { + {10.0, 0, 50, 1.0}, + {20.0, 5, 100, 0.8}, + {100.0, 5, 150, 0.6}, + }; + + const auto metrics = telemetry_ops::aggregate(rows); + expect(metrics.count == 3, "aggregate should preserve event count"); + expect_near(metrics.drop_rate, 10.0 / 300.0, 1e-12, + "drop rate should be weighted by packets sent"); + expect_near(metrics.latency_p50_ms, 20.0, 1e-12, + "p50 should use sorted latency values"); + expect_near(metrics.latency_p95_ms, 92.0, 1e-12, + "p95 should interpolate adjacent latency values"); + expect_near(metrics.avg_link_quality, 0.8, 1e-12, + "link quality should be averaged across events"); + + expect_near(telemetry_ops::percentile({20.0, 10.0}, 25.0), 12.5, + 1e-12, "percentile should sort and interpolate"); + + const auto empty = telemetry_ops::aggregate({}); + expect(empty.count == 0, "empty aggregate should have zero count"); + expect_near(empty.drop_rate, 0.0, 0.0, + "empty aggregate should have zero drop rate"); + expect_near(empty.latency_p95_ms, 0.0, 0.0, + "empty aggregate should have zero percentiles"); +} + +void test_alert_evaluation() { + const telemetry_ops::Thresholds thresholds; + + const json exact_thresholds = { + {"ok", true}, + {"count", 1}, + {"latency_p95_ms", thresholds.latency_p95_ms}, + {"drop_rate", thresholds.drop_rate}, + {"avg_link_quality", thresholds.min_link_quality}, + }; + expect(telemetry_ops::eval_alerts(exact_thresholds, thresholds).empty(), + "values equal to thresholds should not alert"); + + const json degraded = { + {"ok", true}, + {"count", 10}, + {"latency_p95_ms", 250.0}, + {"drop_rate", 0.08}, + {"avg_link_quality", 0.5}, + }; + const json alerts = telemetry_ops::eval_alerts(degraded, thresholds); + expect(alerts.size() == 3, + "three degraded metrics should produce three alerts"); + expect(alerts[0]["type"] == "LATENCY_P95", + "latency degradation should be identified"); + expect(alerts[1]["type"] == "DROP_RATE", + "packet-loss degradation should be identified"); + expect(alerts[1]["severity"] == "HIGH", + "packet-loss alert should be high severity"); + expect(alerts[2]["type"] == "LINK_QUALITY", + "link degradation should be identified"); + + const json aggregator_error = {{"ok", false}, {"count", 5}}; + const json error_alerts = + telemetry_ops::eval_alerts(aggregator_error, thresholds); + expect(error_alerts.size() == 1, + "failed aggregator response should produce one alert"); + expect(error_alerts[0]["type"] == "AGGREGATOR_ERROR", + "failed aggregator response should identify its source"); + + const json no_samples = {{"ok", true}, {"count", 0}}; + expect(telemetry_ops::eval_alerts(no_samples, thresholds).empty(), + "empty windows should not produce threshold alerts"); +} + +} // namespace + +int main(int argc, char** argv) { + const std::unordered_map> tests = { + {"event", test_event_validation_and_idempotency}, + {"aggregation", test_aggregation}, + {"alerts", test_alert_evaluation}, + }; + + if (argc != 2 || !tests.contains(argv[1])) { + std::cerr << "usage: domain_tests \n"; + return 2; + } + + try { + tests.at(argv[1])(); + std::cout << "PASS: " << argv[1] << '\n'; + return 0; + } catch (const std::exception& error) { + std::cerr << "FAIL: " << argv[1] << ": " << error.what() << '\n'; + return 1; + } +} diff --git a/tests/service_pipeline_test.py b/tests/service_pipeline_test.py new file mode 100644 index 0000000..e831574 --- /dev/null +++ b/tests/service_pipeline_test.py @@ -0,0 +1,232 @@ +#!/usr/bin/env python3 + +import argparse +import json +import math +import socket +import subprocess +import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path + + +def available_port(): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.bind(("127.0.0.1", 0)) + return server.getsockname()[1] + + +def request(base_url, method, path, payload=None): + data = None + headers = {} + if payload is not None: + data = json.dumps(payload).encode() + headers["Content-Type"] = "application/json" + + outgoing = urllib.request.Request( + base_url + path, data=data, headers=headers, method=method + ) + try: + with urllib.request.urlopen(outgoing, timeout=2) as response: + return response.status, response.read().decode() + except urllib.error.HTTPError as error: + return error.code, error.read().decode() + + +def wait_until_ready(process, base_url): + deadline = time.monotonic() + 10 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError(f"service exited early with code {process.returncode}") + try: + status, _ = request(base_url, "GET", "/ready") + if status == 200: + return + except (OSError, urllib.error.URLError): + pass + time.sleep(0.05) + raise RuntimeError(f"service at {base_url} did not become ready") + + +def stop(process): + if process is None: + return "" + if process.poll() is None: + process.terminate() + try: + process.wait(timeout=3) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=3) + return process.stdout.read() if process.stdout else "" + + +def assert_close(actual, expected, label): + if not math.isclose(actual, expected, rel_tol=1e-9, abs_tol=1e-9): + raise AssertionError(f"{label}: expected {expected}, got {actual}") + + +def run_pipeline(ingest_binary, aggregator_binary): + ingest_process = None + aggregator_process = None + ingest_log = "" + aggregator_log = "" + + try: + with tempfile.TemporaryDirectory(prefix="telemetryops-test-") as temp_dir: + database = str(Path(temp_dir) / "telemetry.db") + ingest_port = available_port() + aggregator_port = available_port() + while aggregator_port == ingest_port: + aggregator_port = available_port() + ingest_url = f"http://127.0.0.1:{ingest_port}" + aggregator_url = f"http://127.0.0.1:{aggregator_port}" + + ingest_process = subprocess.Popen( + [ingest_binary, str(ingest_port), database], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + wait_until_ready(ingest_process, ingest_url) + + now_ms = int(time.time() * 1000) + events = [ + { + "event_id": "current-1", + "sat_id": "SAT-TEST", + "ts_ms": now_ms, + "latency_ms": 10.0, + "dropped_packets": 0, + "sent_packets": 50, + "link_quality": 1.0, + }, + { + "event_id": "current-2", + "sat_id": "SAT-TEST", + "ts_ms": now_ms, + "latency_ms": 20.0, + "dropped_packets": 5, + "sent_packets": 100, + "link_quality": 0.8, + }, + { + "event_id": "current-3", + "sat_id": "SAT-TEST", + "ts_ms": now_ms, + "latency_ms": 100.0, + "dropped_packets": 5, + "sent_packets": 150, + "link_quality": 0.6, + }, + { + "event_id": "stale", + "sat_id": "SAT-TEST", + "ts_ms": now_ms - 700_000, + "latency_ms": 999.0, + "dropped_packets": 100, + "sent_packets": 100, + "link_quality": 0.0, + }, + { + "event_id": "other-satellite", + "sat_id": "SAT-OTHER", + "ts_ms": now_ms, + "latency_ms": 40.0, + "dropped_packets": 1, + "sent_packets": 10, + "link_quality": 0.7, + }, + ] + + invalid = dict(events[0]) + invalid.pop("link_quality") + status, body = request(ingest_url, "POST", "/telemetry", invalid) + assert status == 400, (status, body) + assert json.loads(body)["error"] == "missing field: link_quality" + + for event in events: + status, body = request(ingest_url, "POST", "/telemetry", event) + assert status == 202, (status, body) + assert json.loads(body) == {"inserted": True, "ok": True} + + status, body = request(ingest_url, "POST", "/telemetry", events[0]) + assert status == 202, (status, body) + assert json.loads(body) == {"inserted": False, "ok": True} + + aggregator_process = subprocess.Popen( + [aggregator_binary, str(aggregator_port), database], + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + wait_until_ready(aggregator_process, aggregator_url) + + query = urllib.parse.urlencode( + {"sat_id": "SAT-TEST", "window_s": "600"} + ) + status, body = request( + aggregator_url, "GET", f"/metrics?{query}" + ) + assert status == 200, (status, body) + metrics = json.loads(body) + assert metrics["count"] == 3 + assert metrics["sat_id"] == "SAT-TEST" + assert metrics["window_s"] == 600 + assert_close(metrics["drop_rate"], 10.0 / 300.0, "drop rate") + assert_close(metrics["latency_p50_ms"], 20.0, "p50 latency") + assert_close(metrics["latency_p95_ms"], 92.0, "p95 latency") + assert_close(metrics["avg_link_quality"], 0.8, "link quality") + + other_query = urllib.parse.urlencode({"sat_id": "SAT-OTHER"}) + status, body = request( + aggregator_url, "GET", f"/metrics?{other_query}" + ) + assert status == 200, (status, body) + assert json.loads(body)["count"] == 1 + + status, body = request(aggregator_url, "GET", "/metrics") + assert status == 400, (status, body) + assert json.loads(body)["error"] == "missing sat_id" + + status, body = request(ingest_url, "GET", "/metrics") + assert status == 200 + assert "telemetry_inserted_total 5" in body + assert "telemetry_duplicates_total 1" in body + + status, body = request(aggregator_url, "GET", "/prom") + assert status == 200 + assert ( + 'http_requests_total{service="aggregator",route="/metrics"} 3' + in body + ) + except Exception: + ingest_log = stop(ingest_process) + ingest_process = None + aggregator_log = stop(aggregator_process) + aggregator_process = None + print("ingest output:\n" + ingest_log) + print("aggregator output:\n" + aggregator_log) + raise + finally: + if aggregator_process is not None: + stop(aggregator_process) + if ingest_process is not None: + stop(ingest_process) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--ingest", required=True) + parser.add_argument("--aggregator", required=True) + arguments = parser.parse_args() + + run_pipeline(arguments.ingest, arguments.aggregator) + print("PASS: service pipeline integration") + + +if __name__ == "__main__": + main()