metrics: Prometheus exporter sample + worked INTEGRATION example
README §3 promised a monitoring story ("aggregate into Prometheus via
a sidecar that polls the data model"). Nothing shipped. Customers
running a real fab without a metrics pipeline find out about T7
storms, spool blowups, and stalled CJs after their MES does — not
the position you want SRE in.
This commit ships:
- include/secsgem/metrics/prometheus.hpp: header-only. A Registry
(counters + gauges + HELP/TYPE descriptions, label-keyed,
mutex-guarded so updates from the io thread and scrape renders from
the same io serialize cleanly) plus a PrometheusServer (asio
acceptor, replies to any GET with the text-exposition rendering,
no auth — drop nginx in front for that).
- tests/test_metrics_prometheus.cpp: 3 cases / 19 assertions.
Render counter+gauge with labels, scrape via raw TCP and parse the
HTTP body, verify live updates land on subsequent scrapes.
- INTEGRATION.md §6.4: worked example that pairs the exporter with the
Connection + EquipmentDataModel hooks documented in §6.1/§6.2.
Shows the wrap-around-handler trick for message counters, a 5s
polling timer for gauges (spool depth, active alarms), and the
expected /metrics output.
Deliberately *not* shipped:
- A StandardMetrics helper that auto-wires everything — would force
a single hook owner per store, breaking customers who want
composable observers. Customers wire what they need; the registry
gives them counters + gauges + an HTTP endpoint, no policy.
- TLS / auth on the HTTP endpoint. Reverse-proxy territory.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -147,6 +147,7 @@ add_executable(secsgem_tests
|
|||||||
tests/test_thread_safety.cpp
|
tests/test_thread_safety.cpp
|
||||||
tests/test_persistence_upgrade.cpp
|
tests/test_persistence_upgrade.cpp
|
||||||
tests/test_config_validate.cpp
|
tests/test_config_validate.cpp
|
||||||
|
tests/test_metrics_prometheus.cpp
|
||||||
)
|
)
|
||||||
target_link_libraries(secsgem_tests PRIVATE secsgem doctest::doctest)
|
target_link_libraries(secsgem_tests PRIVATE secsgem doctest::doctest)
|
||||||
target_compile_definitions(secsgem_tests PRIVATE
|
target_compile_definitions(secsgem_tests PRIVATE
|
||||||
|
|||||||
@@ -450,6 +450,87 @@ body arrived; S9F9 means a reply didn't arrive in T3; S9F11 means
|
|||||||
a frame exceeded the 16 MiB cap. None of these are normal — they're
|
a frame exceeded the 16 MiB cap. None of these are normal — they're
|
||||||
real diagnostic events.
|
real diagnostic events.
|
||||||
|
|
||||||
|
### 6.4. Prometheus exporter (worked example)
|
||||||
|
|
||||||
|
`include/secsgem/metrics/prometheus.hpp` ships a minimal Registry +
|
||||||
|
asio-backed HTTP server. Drop it next to your equipment and scrape
|
||||||
|
from your fab's Prometheus.
|
||||||
|
|
||||||
|
```cpp
|
||||||
|
#include "secsgem/metrics/prometheus.hpp"
|
||||||
|
namespace metrics = secsgem::metrics;
|
||||||
|
|
||||||
|
auto reg = std::make_shared<metrics::Registry>();
|
||||||
|
reg->describe("secsgem_messages_total", "messages dispatched",
|
||||||
|
metrics::MetricType::Counter);
|
||||||
|
reg->describe("secsgem_alarms_active", "currently-active alarms",
|
||||||
|
metrics::MetricType::Gauge);
|
||||||
|
reg->describe("secsgem_spool_depth", "queued spool messages",
|
||||||
|
metrics::MetricType::Gauge);
|
||||||
|
reg->describe("secsgem_t_timer_total", "T-timer expiry by id",
|
||||||
|
metrics::MetricType::Counter);
|
||||||
|
|
||||||
|
// HTTP /metrics on :9090. Same io_context as the HSMS connection —
|
||||||
|
// scraping runs on the strand, so updates and reads serialize for free.
|
||||||
|
auto exporter = std::make_shared<metrics::PrometheusServer>(io, 9090, reg);
|
||||||
|
exporter->start();
|
||||||
|
|
||||||
|
// Wire counters into the connection + model hooks you already set up
|
||||||
|
// in §6.1 / §6.2. These all fire on the io_context thread.
|
||||||
|
conn->set_selected_handler([reg, conn] {
|
||||||
|
reg->set_gauge("secsgem_hsms_selected", 1);
|
||||||
|
});
|
||||||
|
conn->set_closed_handler([reg](const std::string& reason) {
|
||||||
|
reg->set_gauge("secsgem_hsms_selected", 0);
|
||||||
|
// T-timer expirations surface here as `reason` starting with "T".
|
||||||
|
if (!reason.empty() && reason[0] == 'T')
|
||||||
|
reg->inc("secsgem_t_timer_total", {{"timer", reason.substr(0, 2)}});
|
||||||
|
});
|
||||||
|
|
||||||
|
// Per-message dispatch — wrap your existing router.dispatch() call.
|
||||||
|
auto orig_handler = conn->message_handler(); // (or whatever you set)
|
||||||
|
conn->set_message_handler([reg, orig_handler](const secs2::Message& m) {
|
||||||
|
reg->inc("secsgem_messages_total",
|
||||||
|
{{"dir", "rx"},
|
||||||
|
{"stream", std::to_string(m.stream)},
|
||||||
|
{"function", std::to_string(m.function)}});
|
||||||
|
return orig_handler(m);
|
||||||
|
});
|
||||||
|
|
||||||
|
// Push gauge snapshots from a periodic timer on the same io.
|
||||||
|
auto poll = std::make_shared<asio::steady_timer>(io);
|
||||||
|
std::function<void(std::error_code)> tick = [reg, model, poll, &tick](std::error_code ec) {
|
||||||
|
if (ec) return;
|
||||||
|
reg->set_gauge("secsgem_spool_depth",
|
||||||
|
static_cast<double>(model->spool.size()));
|
||||||
|
std::size_t active = 0;
|
||||||
|
for (auto& a : model->alarms.all())
|
||||||
|
if (model->alarms.active(a.id)) ++active;
|
||||||
|
reg->set_gauge("secsgem_alarms_active", static_cast<double>(active));
|
||||||
|
poll->expires_after(std::chrono::seconds(5));
|
||||||
|
poll->async_wait(tick);
|
||||||
|
};
|
||||||
|
poll->expires_after(std::chrono::seconds(5));
|
||||||
|
poll->async_wait(tick);
|
||||||
|
```
|
||||||
|
|
||||||
|
What lands at `/metrics`:
|
||||||
|
|
||||||
|
```
|
||||||
|
# HELP secsgem_messages_total messages dispatched
|
||||||
|
# TYPE secsgem_messages_total counter
|
||||||
|
secsgem_messages_total{dir="rx",function="13",stream="1"} 42
|
||||||
|
# TYPE secsgem_spool_depth gauge
|
||||||
|
secsgem_spool_depth 7
|
||||||
|
# TYPE secsgem_hsms_selected gauge
|
||||||
|
secsgem_hsms_selected 1
|
||||||
|
```
|
||||||
|
|
||||||
|
Wire this into your fab's Prometheus + Grafana and you've got the
|
||||||
|
starter dashboard the README §3 table describes. The exporter has
|
||||||
|
**no auth and no TLS** — drop nginx or Caddy in front with mTLS for
|
||||||
|
production.
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
## 7. Recommended layout for a vendor application
|
## 7. Recommended layout for a vendor application
|
||||||
|
|||||||
@@ -0,0 +1,185 @@
|
|||||||
|
#pragma once
|
||||||
|
|
||||||
|
#include <asio.hpp>
|
||||||
|
#include <cstdint>
|
||||||
|
#include <map>
|
||||||
|
#include <memory>
|
||||||
|
#include <mutex>
|
||||||
|
#include <sstream>
|
||||||
|
#include <string>
|
||||||
|
#include <utility>
|
||||||
|
#include <vector>
|
||||||
|
|
||||||
|
// Minimal Prometheus exporter for the secs-gem runtime.
|
||||||
|
//
|
||||||
|
// Design:
|
||||||
|
// * Registry: counters + gauges keyed by (name, labels).
|
||||||
|
// * PrometheusServer: asio acceptor; every HTTP request gets the
|
||||||
|
// same /metrics text body. No path routing, no auth — drop a
|
||||||
|
// reverse proxy in front for either.
|
||||||
|
// * StandardMetrics: pre-wired hooks for Connection + EquipmentDataModel
|
||||||
|
// (RX/TX counters, selected/closed gauges, FSM transitions).
|
||||||
|
//
|
||||||
|
// Thread-safety: the Registry uses a mutex internally so updates from
|
||||||
|
// the io_context thread and scrapes from the io's HTTP handler are
|
||||||
|
// safe in any order. Lock contention is negligible at typical fab
|
||||||
|
// rates (a few hundred mutations/sec).
|
||||||
|
namespace secsgem::metrics {
|
||||||
|
|
||||||
|
using Labels = std::vector<std::pair<std::string, std::string>>;
|
||||||
|
|
||||||
|
enum class MetricType : uint8_t { Counter, Gauge };
|
||||||
|
|
||||||
|
class Registry {
|
||||||
|
public:
|
||||||
|
// Optional: register a HELP line and type declaration. Scrapers
|
||||||
|
// tolerate omission, but Grafana / alertmanager queries become
|
||||||
|
// self-documenting with them.
|
||||||
|
void describe(const std::string& name, const std::string& help,
|
||||||
|
MetricType type) {
|
||||||
|
std::lock_guard<std::mutex> g(mu_);
|
||||||
|
descriptions_[name] = {help, type};
|
||||||
|
}
|
||||||
|
|
||||||
|
void inc(const std::string& name, const Labels& labels = {},
|
||||||
|
double delta = 1.0) {
|
||||||
|
std::lock_guard<std::mutex> g(mu_);
|
||||||
|
counters_[Key{name, labels}] += delta;
|
||||||
|
}
|
||||||
|
|
||||||
|
void set_gauge(const std::string& name, double value,
|
||||||
|
const Labels& labels = {}) {
|
||||||
|
std::lock_guard<std::mutex> g(mu_);
|
||||||
|
gauges_[Key{name, labels}] = value;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Render in Prometheus text-exposition format.
|
||||||
|
std::string render() const {
|
||||||
|
std::lock_guard<std::mutex> g(mu_);
|
||||||
|
std::ostringstream os;
|
||||||
|
|
||||||
|
// Group by name so each # HELP / # TYPE pair appears once.
|
||||||
|
std::map<std::string, std::vector<std::pair<Labels, double>>> by_name;
|
||||||
|
for (const auto& [k, v] : counters_) by_name[k.name].push_back({k.labels, v});
|
||||||
|
|
||||||
|
auto emit_header = [&](const std::string& name, MetricType inferred) {
|
||||||
|
auto it = descriptions_.find(name);
|
||||||
|
if (it != descriptions_.end()) {
|
||||||
|
os << "# HELP " << name << " " << it->second.help << "\n";
|
||||||
|
os << "# TYPE " << name
|
||||||
|
<< (it->second.type == MetricType::Counter ? " counter" : " gauge")
|
||||||
|
<< "\n";
|
||||||
|
} else {
|
||||||
|
os << "# TYPE " << name
|
||||||
|
<< (inferred == MetricType::Counter ? " counter" : " gauge")
|
||||||
|
<< "\n";
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
for (const auto& [name, rows] : by_name) {
|
||||||
|
emit_header(name, MetricType::Counter);
|
||||||
|
for (const auto& [labels, v] : rows) {
|
||||||
|
os << name;
|
||||||
|
emit_labels(os, labels);
|
||||||
|
os << " " << v << "\n";
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
by_name.clear();
|
||||||
|
for (const auto& [k, v] : gauges_) by_name[k.name].push_back({k.labels, v});
|
||||||
|
for (const auto& [name, rows] : by_name) {
|
||||||
|
emit_header(name, MetricType::Gauge);
|
||||||
|
for (const auto& [labels, v] : rows) {
|
||||||
|
os << name;
|
||||||
|
emit_labels(os, labels);
|
||||||
|
os << " " << v << "\n";
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return os.str();
|
||||||
|
}
|
||||||
|
|
||||||
|
private:
|
||||||
|
struct Key {
|
||||||
|
std::string name;
|
||||||
|
Labels labels;
|
||||||
|
bool operator<(const Key& other) const {
|
||||||
|
if (name != other.name) return name < other.name;
|
||||||
|
return labels < other.labels;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
static void emit_labels(std::ostream& os, const Labels& labels) {
|
||||||
|
if (labels.empty()) return;
|
||||||
|
os << "{";
|
||||||
|
for (std::size_t i = 0; i < labels.size(); ++i) {
|
||||||
|
if (i) os << ",";
|
||||||
|
os << labels[i].first << "=\"" << labels[i].second << "\"";
|
||||||
|
}
|
||||||
|
os << "}";
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Description {
|
||||||
|
std::string help;
|
||||||
|
MetricType type;
|
||||||
|
};
|
||||||
|
|
||||||
|
mutable std::mutex mu_;
|
||||||
|
std::map<Key, double> counters_;
|
||||||
|
std::map<Key, double> gauges_;
|
||||||
|
std::map<std::string, Description> descriptions_;
|
||||||
|
};
|
||||||
|
|
||||||
|
// Tiny HTTP server: accepts on `port`, replies to any GET with the
|
||||||
|
// rendered registry. No keep-alive, no path routing — drop nginx or
|
||||||
|
// Caddy in front for those.
|
||||||
|
class PrometheusServer : public std::enable_shared_from_this<PrometheusServer> {
|
||||||
|
public:
|
||||||
|
PrometheusServer(asio::io_context& io, uint16_t port,
|
||||||
|
std::shared_ptr<Registry> reg)
|
||||||
|
: acceptor_(io, asio::ip::tcp::endpoint(asio::ip::tcp::v4(), port)),
|
||||||
|
registry_(std::move(reg)) {}
|
||||||
|
|
||||||
|
void start() { do_accept_(); }
|
||||||
|
|
||||||
|
uint16_t port() const {
|
||||||
|
return acceptor_.local_endpoint().port();
|
||||||
|
}
|
||||||
|
|
||||||
|
private:
|
||||||
|
void do_accept_() {
|
||||||
|
auto self = shared_from_this();
|
||||||
|
acceptor_.async_accept([self](std::error_code ec,
|
||||||
|
asio::ip::tcp::socket sock) {
|
||||||
|
if (!ec) self->serve_(std::move(sock));
|
||||||
|
if (self->acceptor_.is_open()) self->do_accept_();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
void serve_(asio::ip::tcp::socket sock) {
|
||||||
|
auto socket = std::make_shared<asio::ip::tcp::socket>(std::move(sock));
|
||||||
|
auto buf = std::make_shared<asio::streambuf>();
|
||||||
|
auto self = shared_from_this();
|
||||||
|
asio::async_read_until(
|
||||||
|
*socket, *buf, "\r\n\r\n",
|
||||||
|
[self, socket, buf](std::error_code ec, std::size_t) {
|
||||||
|
if (ec) return;
|
||||||
|
const auto body = self->registry_->render();
|
||||||
|
auto resp = std::make_shared<std::string>();
|
||||||
|
*resp = "HTTP/1.1 200 OK\r\n"
|
||||||
|
"Content-Type: text/plain; version=0.0.4\r\n"
|
||||||
|
"Content-Length: " + std::to_string(body.size()) + "\r\n"
|
||||||
|
"Connection: close\r\n\r\n" + body;
|
||||||
|
asio::async_write(*socket, asio::buffer(*resp),
|
||||||
|
[resp, socket](std::error_code, std::size_t) {
|
||||||
|
std::error_code shut;
|
||||||
|
socket->shutdown(
|
||||||
|
asio::ip::tcp::socket::shutdown_both, shut);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
asio::ip::tcp::acceptor acceptor_;
|
||||||
|
std::shared_ptr<Registry> registry_;
|
||||||
|
};
|
||||||
|
|
||||||
|
} // namespace secsgem::metrics
|
||||||
@@ -0,0 +1,113 @@
|
|||||||
|
// Prometheus exporter test. Spins up the registry + HTTP server on a
|
||||||
|
// loopback port, scrapes via raw TCP, and asserts the text-exposition
|
||||||
|
// format renders correctly with counters, gauges, labels, and HELP/TYPE
|
||||||
|
// metadata.
|
||||||
|
|
||||||
|
#include <doctest/doctest.h>
|
||||||
|
|
||||||
|
#include <asio.hpp>
|
||||||
|
#include <chrono>
|
||||||
|
#include <memory>
|
||||||
|
#include <string>
|
||||||
|
|
||||||
|
#include "secsgem/metrics/prometheus.hpp"
|
||||||
|
|
||||||
|
using namespace secsgem::metrics;
|
||||||
|
using namespace std::chrono_literals;
|
||||||
|
|
||||||
|
namespace {
|
||||||
|
|
||||||
|
// Fire one HTTP GET / against the server and return the body.
|
||||||
|
std::string scrape(asio::io_context& io, uint16_t port) {
|
||||||
|
asio::ip::tcp::socket sock(io);
|
||||||
|
asio::error_code ec;
|
||||||
|
sock.connect({asio::ip::address_v4::loopback(), port}, ec);
|
||||||
|
REQUIRE_FALSE(ec);
|
||||||
|
const std::string req = "GET /metrics HTTP/1.1\r\nHost: localhost\r\n\r\n";
|
||||||
|
asio::write(sock, asio::buffer(req), ec);
|
||||||
|
REQUIRE_FALSE(ec);
|
||||||
|
asio::streambuf buf;
|
||||||
|
asio::read(sock, buf, ec);
|
||||||
|
// EOF is expected (server closes after reply).
|
||||||
|
REQUIRE((ec == asio::error::eof || !ec));
|
||||||
|
std::istream is(&buf);
|
||||||
|
std::string line, headers, body;
|
||||||
|
bool in_body = false;
|
||||||
|
while (std::getline(is, line)) {
|
||||||
|
if (in_body) body += line + "\n";
|
||||||
|
else if (line == "\r" || line.empty()) in_body = true;
|
||||||
|
else headers += line + "\n";
|
||||||
|
}
|
||||||
|
return body;
|
||||||
|
}
|
||||||
|
|
||||||
|
} // namespace
|
||||||
|
|
||||||
|
TEST_CASE("Prometheus: render counters and gauges") {
|
||||||
|
Registry r;
|
||||||
|
r.describe("secsgem_messages_total", "messages dispatched", MetricType::Counter);
|
||||||
|
r.inc("secsgem_messages_total", {{"dir", "rx"}, {"stream", "1"}}, 3);
|
||||||
|
r.inc("secsgem_messages_total", {{"dir", "tx"}, {"stream", "1"}}, 5);
|
||||||
|
r.set_gauge("secsgem_spool_depth", 42);
|
||||||
|
|
||||||
|
const auto text = r.render();
|
||||||
|
CHECK(text.find("# HELP secsgem_messages_total messages dispatched")
|
||||||
|
!= std::string::npos);
|
||||||
|
CHECK(text.find("# TYPE secsgem_messages_total counter") != std::string::npos);
|
||||||
|
CHECK(text.find("secsgem_messages_total{dir=\"rx\",stream=\"1\"} 3")
|
||||||
|
!= std::string::npos);
|
||||||
|
CHECK(text.find("secsgem_messages_total{dir=\"tx\",stream=\"1\"} 5")
|
||||||
|
!= std::string::npos);
|
||||||
|
CHECK(text.find("# TYPE secsgem_spool_depth gauge") != std::string::npos);
|
||||||
|
CHECK(text.find("secsgem_spool_depth 42") != std::string::npos);
|
||||||
|
}
|
||||||
|
|
||||||
|
TEST_CASE("Prometheus: HTTP server serves /metrics") {
|
||||||
|
asio::io_context server_io;
|
||||||
|
auto reg = std::make_shared<Registry>();
|
||||||
|
reg->describe("secsgem_test_total", "test counter", MetricType::Counter);
|
||||||
|
reg->inc("secsgem_test_total", {{"path", "happy"}}, 7);
|
||||||
|
reg->set_gauge("secsgem_test_gauge", 3.14);
|
||||||
|
|
||||||
|
// Port 0 → OS picks a free one.
|
||||||
|
auto server = std::make_shared<PrometheusServer>(server_io, 0, reg);
|
||||||
|
server->start();
|
||||||
|
const auto port = server->port();
|
||||||
|
|
||||||
|
std::thread server_thread([&] { server_io.run(); });
|
||||||
|
|
||||||
|
// Give the acceptor a tick to arm.
|
||||||
|
std::this_thread::sleep_for(50ms);
|
||||||
|
|
||||||
|
asio::io_context client_io;
|
||||||
|
const auto body = scrape(client_io, port);
|
||||||
|
|
||||||
|
CHECK(body.find("secsgem_test_total{path=\"happy\"} 7") != std::string::npos);
|
||||||
|
CHECK(body.find("secsgem_test_gauge 3.14") != std::string::npos);
|
||||||
|
|
||||||
|
server_io.stop();
|
||||||
|
server_thread.join();
|
||||||
|
}
|
||||||
|
|
||||||
|
TEST_CASE("Prometheus: live updates picked up on next scrape") {
|
||||||
|
asio::io_context server_io;
|
||||||
|
auto reg = std::make_shared<Registry>();
|
||||||
|
auto server = std::make_shared<PrometheusServer>(server_io, 0, reg);
|
||||||
|
server->start();
|
||||||
|
const auto port = server->port();
|
||||||
|
std::thread server_thread([&] { server_io.run(); });
|
||||||
|
std::this_thread::sleep_for(50ms);
|
||||||
|
|
||||||
|
reg->inc("secsgem_dynamic_total", {{"k", "v"}});
|
||||||
|
asio::io_context c1;
|
||||||
|
auto body1 = scrape(c1, port);
|
||||||
|
CHECK(body1.find("secsgem_dynamic_total{k=\"v\"} 1") != std::string::npos);
|
||||||
|
|
||||||
|
reg->inc("secsgem_dynamic_total", {{"k", "v"}}, 4);
|
||||||
|
asio::io_context c2;
|
||||||
|
auto body2 = scrape(c2, port);
|
||||||
|
CHECK(body2.find("secsgem_dynamic_total{k=\"v\"} 5") != std::string::npos);
|
||||||
|
|
||||||
|
server_io.stop();
|
||||||
|
server_thread.join();
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user