From 037fea28802a9c3d2e26a2ae862c0685c49174db Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 09:19:48 -0300 Subject: [PATCH 1/9] feat(metrics): add PrometheusSerializer + MetricLabels (grouped, FMT, escaped, numeric-parity) --- src/CMakeLists.txt | 1 + src/MetricLabels.h | 17 ++++ src/Metrics.cpp | 11 ++- src/Metrics.h | 16 ++-- src/PrometheusSerializer.h | 117 +++++++++++++++++++++++ src/tests/test_prometheus_serializer.cpp | 93 ++++++++++++++++++ 6 files changed, 243 insertions(+), 12 deletions(-) create mode 100644 src/MetricLabels.h create mode 100644 src/PrometheusSerializer.h create mode 100644 src/tests/test_prometheus_serializer.cpp diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index c5d7dc270..b3cec03a4 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -87,6 +87,7 @@ add_executable(unit-tests-visor-core tests/test_policies.cpp tests/test_handlers.cpp tests/test_module_plugins.cpp + tests/test_prometheus_serializer.cpp ) target_include_directories(unit-tests-visor-core PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) diff --git a/src/MetricLabels.h b/src/MetricLabels.h new file mode 100644 index 000000000..78dea217b --- /dev/null +++ b/src/MetricLabels.h @@ -0,0 +1,17 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ +#pragma once + +#include +#include + +namespace visor { + +using LabelMap = std::map; + +// Process-global static labels (e.g. instance). Storage defined in Metrics.cpp. +LabelMap &prometheus_static_labels_mutable(); +inline const LabelMap &prometheus_static_labels() { return prometheus_static_labels_mutable(); } + +} diff --git a/src/Metrics.cpp b/src/Metrics.cpp index 149423d62..c08ed9234 100644 --- a/src/Metrics.cpp +++ b/src/Metrics.cpp @@ -95,8 +95,11 @@ void Cardinality::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &s } } -// static storage for base labels -Metric::LabelMap Metric::_static_labels; +LabelMap &prometheus_static_labels_mutable() +{ + static LabelMap labels; + return labels; +} void Metric::name_json_assign(json &j, const json &val) const { @@ -129,8 +132,8 @@ std::string Metric::base_name_snake() const std::string Metric::name_snake(std::initializer_list add_names, Metric::LabelMap add_labels) const { std::string label_text{"{"}; - if (!_static_labels.empty()) { - for (const auto &[key, value] : _static_labels) { + if (!prometheus_static_labels().empty()) { + for (const auto &[key, value] : prometheus_static_labels()) { label_text.append(key + "=\"" + value + "\","); } } diff --git a/src/Metrics.h b/src/Metrics.h index ae979d2af..2f5436caf 100644 --- a/src/Metrics.h +++ b/src/Metrics.h @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #pragma once +#include "MetricLabels.h" #include #include #include @@ -67,19 +68,13 @@ static inline std::vector get_quantiles(const datasketches::kll_sketch &qu class Metric { public: - typedef std::map LabelMap; + typedef visor::LabelMap LabelMap; enum class Aggregate { DEFAULT, SUM }; -private: - /** - * static labels which will be applied to all metrics - */ - static LabelMap _static_labels; - protected: std::vector _name; std::string _desc; @@ -121,7 +116,12 @@ class Metric static void add_static_label(const std::string &label, const std::string &value) { - _static_labels.emplace(label, value); + prometheus_static_labels_mutable().emplace(label, value); + } + + static void reset_static_labels() + { + prometheus_static_labels_mutable().clear(); } void name_json_assign(json &j, const json &val) const; diff --git a/src/PrometheusSerializer.h b/src/PrometheusSerializer.h new file mode 100644 index 000000000..fbb728316 --- /dev/null +++ b/src/PrometheusSerializer.h @@ -0,0 +1,117 @@ +/* This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ +#pragma once + +#include "MetricLabels.h" +#include +#include +#include +#include +#include +#include +#include + +namespace visor { + +class PrometheusSerializer +{ +public: + enum class Type { Gauge, Counter, Histogram, Summary }; + + template + void write(const std::string &base, Type type, const std::string &help, + std::initializer_list suffix, const LabelMap &labels, V value) + { + auto it = _families.find(base); + if (it == _families.end()) { + _order.push_back(base); + it = _families.emplace(base, Family{type, help, {}}).first; + } + std::string line = base; + for (const auto &s : suffix) { + line.push_back('_'); + line.append(s); + } + append_labels(line, labels); + line.push_back(' '); + if constexpr (std::is_floating_point_v) { + std::ostringstream oss; + oss << value; + line.append(oss.str()); + } else { + fmt::format_to(std::back_inserter(line), "{}", value); + } + it->second.series.push_back(std::move(line)); + } + + std::string finalize() const + { + std::string out; + for (const auto &base : _order) { + const auto &fam = _families.at(base); + fmt::format_to(std::back_inserter(out), "# HELP {} {}\n", base, fam.help); + fmt::format_to(std::back_inserter(out), "# TYPE {} {}\n", base, type_str(fam.type)); + for (const auto &series : fam.series) { + out.append(series); + out.push_back('\n'); + } + } + return out; + } + +private: + struct Family { + Type type; + std::string help; + std::vector series; + }; + std::vector _order; + std::map _families; + + static const char *type_str(Type t) + { + switch (t) { + case Type::Counter: return "counter"; + case Type::Histogram: return "histogram"; + case Type::Summary: return "summary"; + case Type::Gauge: + default: return "gauge"; + } + } + + static void append_escaped(std::string &out, const std::string &v) + { + for (char c : v) { + switch (c) { + case '\\': out.append("\\\\"); break; + case '"': out.append("\\\""); break; + case '\n': out.append("\\n"); break; + default: out.push_back(c); + } + } + } + + static void append_labels(std::string &line, const LabelMap &labels) + { + line.push_back('{'); + bool any = false; + auto emit = [&](const LabelMap &m) { + for (const auto &[k, v] : m) { + line.append(k); + line.append("=\""); + append_escaped(line, v); + line.append("\","); + any = true; + } + }; + emit(prometheus_static_labels()); + emit(labels); + if (any) { + line.pop_back(); + } + line.push_back('}'); + } +}; + +} diff --git a/src/tests/test_prometheus_serializer.cpp b/src/tests/test_prometheus_serializer.cpp new file mode 100644 index 000000000..f13bffa26 --- /dev/null +++ b/src/tests/test_prometheus_serializer.cpp @@ -0,0 +1,93 @@ +#include "PrometheusSerializer.h" +#include "Metrics.h" +#include +#include + +using namespace visor; +using Type = PrometheusSerializer::Type; + +TEST_CASE("single gauge family", "[prometheus][serializer]") +{ + PrometheusSerializer s; + s.write("net_packets_total", Type::Gauge, "Total packets", {}, {}, 42); + CHECK(s.finalize() == + "# HELP net_packets_total Total packets\n" + "# TYPE net_packets_total gauge\n" + "net_packets_total{} 42\n"); +} + +TEST_CASE("repeated family writes one header, contiguous series", "[prometheus][serializer]") +{ + PrometheusSerializer s; + s.write("flow_bytes", Type::Gauge, "bytes", {}, {{"device", "A"}}, 10); + s.write("other_metric", Type::Gauge, "x", {}, {}, 1); + s.write("flow_bytes", Type::Gauge, "bytes", {}, {{"device", "B"}}, 20); + CHECK(s.finalize() == + "# HELP flow_bytes bytes\n" + "# TYPE flow_bytes gauge\n" + "flow_bytes{device=\"A\"} 10\n" + "flow_bytes{device=\"B\"} 20\n" + "# HELP other_metric x\n" + "# TYPE other_metric gauge\n" + "other_metric{} 1\n"); +} + +TEST_CASE("histogram suffix + le label", "[prometheus][serializer]") +{ + PrometheusSerializer s; + s.write("dns_xact", Type::Histogram, "latency", {"bucket"}, {{"le", "10"}}, 3.0); + s.write("dns_xact", Type::Histogram, "latency", {"count"}, {}, 7); + CHECK(s.finalize() == + "# HELP dns_xact latency\n" + "# TYPE dns_xact histogram\n" + "dns_xact_bucket{le=\"10\"} 3\n" + "dns_xact_count{} 7\n"); +} + +TEST_CASE("static labels merged before passed labels", "[prometheus][serializer]") +{ + Metric::reset_static_labels(); + Metric::add_static_label("instance", "host1"); + PrometheusSerializer s; + s.write("net_x", Type::Gauge, "d", {}, {{"dir", "in"}}, 1); + CHECK(s.finalize() == + "# HELP net_x d\n" + "# TYPE net_x gauge\n" + "net_x{instance=\"host1\",dir=\"in\"} 1\n"); + Metric::reset_static_labels(); +} + +TEST_CASE("label values are escaped", "[prometheus][serializer]") +{ + PrometheusSerializer s; + s.write("m", Type::Gauge, "d", {}, {{"desc", "a\"b\\c\nd"}}, 1); + CHECK(s.finalize() == + "# HELP m d\n" + "# TYPE m gauge\n" + "m{desc=\"a\\\"b\\\\c\\nd\"} 1\n"); +} + +TEST_CASE("numeric formatting matches ostream operator<<", "[prometheus][serializer]") +{ + auto oss = [](auto v) { std::stringstream o; o << v; return o.str(); }; + PrometheusSerializer s; + s.write("i", Type::Gauge, "d", {}, {}, int64_t{1234567}); + s.write("neg", Type::Gauge, "d", {}, {}, int64_t{-5}); + s.write("d_half", Type::Gauge, "d", {}, {}, 0.5); + s.write("d_third", Type::Gauge, "d", {}, {}, 1.0 / 3.0); + s.write("d_big", Type::Gauge, "d", {}, {}, 1000000.0); + s.write("d_huge", Type::Gauge, "d", {}, {}, 1e20); + s.write("d_zero", Type::Gauge, "d", {}, {}, 0.0); + s.write("d_small", Type::Gauge, "d", {}, {}, 0.00001); + auto out = s.finalize(); + for (auto pair : {std::make_pair("i", oss(int64_t{1234567})), + std::make_pair("neg", oss(int64_t{-5})), + std::make_pair("d_half", oss(0.5)), + std::make_pair("d_third", oss(1.0 / 3.0)), + std::make_pair("d_big", oss(1000000.0)), + std::make_pair("d_huge", oss(1e20)), + std::make_pair("d_zero", oss(0.0)), + std::make_pair("d_small", oss(0.00001))}) { + CHECK(out.find(std::string(pair.first) + "{} " + pair.second + "\n") != std::string::npos); + } +} From 1f800e53cf09c9f6cd7f1f918876444b721c39cf Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 09:45:15 -0300 Subject: [PATCH 2/9] fix(metrics): group Prometheus output by family, fixing duplicate HELP/TYPE (#716) --- src/AbstractMetricsManager.h | 11 ++- src/CoreServer.cpp | 5 +- src/Metrics.cpp | 45 ++------- src/Metrics.h | 59 ++++++------ src/Policies.cpp | 7 +- src/PrometheusSerializer.h | 2 + src/StreamHandler.h | 15 +-- src/handlers/bgp/BgpStreamHandler.cpp | 25 ++--- src/handlers/bgp/BgpStreamHandler.h | 3 +- src/handlers/bgp/tests/test_bgp_layer.cpp | 4 +- src/handlers/dhcp/DhcpStreamHandler.cpp | 37 ++++---- src/handlers/dhcp/DhcpStreamHandler.h | 3 +- src/handlers/dhcp/tests/test_dhcp_layer.cpp | 4 +- src/handlers/dns/v1/DnsStreamHandler.cpp | 91 ++++++++++--------- src/handlers/dns/v1/DnsStreamHandler.h | 3 +- src/handlers/dns/v1/tests/test_dns_layer.cpp | 4 +- src/handlers/dns/v2/DnsStreamHandler.cpp | 53 +++++------ src/handlers/dns/v2/DnsStreamHandler.h | 47 +++++----- src/handlers/dns/v2/tests/test_dns_layer.cpp | 4 +- src/handlers/flow/FlowStreamHandler.cpp | 65 ++++++------- src/handlers/flow/FlowStreamHandler.h | 3 +- src/handlers/flow/test_flows.cpp | 52 ++++++++++- .../InputResourcesStreamHandler.cpp | 17 ++-- .../InputResourcesStreamHandler.h | 3 +- .../input_resources/test_resources_layer.cpp | 8 +- src/handlers/net/v1/NetStreamHandler.cpp | 57 ++++++------ src/handlers/net/v1/NetStreamHandler.h | 3 +- src/handlers/net/v1/tests/test_net_layer.cpp | 4 +- src/handlers/net/v2/NetStreamHandler.cpp | 29 +++--- src/handlers/net/v2/NetStreamHandler.h | 19 ++-- src/handlers/net/v2/tests/test_net_layer.cpp | 8 +- .../netprobe/NetProbeStreamHandler.cpp | 25 ++--- src/handlers/netprobe/NetProbeStreamHandler.h | 3 +- src/handlers/netprobe/test_net_probe.cpp | 4 +- src/handlers/pcap/PcapStreamHandler.cpp | 9 +- src/handlers/pcap/PcapStreamHandler.h | 3 +- src/handlers/pcap/tests/test_pcap_layer.cpp | 4 +- src/tests/test_handlers.cpp | 43 ++++----- src/tests/test_metrics.cpp | 51 +++++++---- src/tests/test_prometheus_serializer.cpp | 5 + 40 files changed, 450 insertions(+), 387 deletions(-) diff --git a/src/AbstractMetricsManager.h b/src/AbstractMetricsManager.h index 29940be27..584acf62d 100644 --- a/src/AbstractMetricsManager.h +++ b/src/AbstractMetricsManager.h @@ -19,6 +19,7 @@ #endif #include "Configurable.h" #include "Metrics.h" +#include "PrometheusSerializer.h" #include #include #include @@ -217,7 +218,7 @@ class AbstractMetricsBucket } virtual void to_json(json &j) const = 0; - virtual void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const = 0; + virtual void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const = 0; virtual void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const = 0; virtual void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) = 0; }; @@ -503,7 +504,7 @@ class AbstractMetricsManager _metric_buckets.at(period)->to_json(j[key]); } - void window_single_prometheus(std::stringstream &out, uint64_t period = 0, Metric::LabelMap add_labels = {}) const + void window_single_prometheus(PrometheusSerializer &ser, uint64_t period = 0, Metric::LabelMap add_labels = {}) const { std::shared_lock rl(_base_mutex); std::shared_lock rbl(_bucket_mutex); @@ -527,7 +528,7 @@ class AbstractMetricsManager add_labels["tap"] = _tap_name; } - _metric_buckets.at(period)->to_prometheus(out, add_labels); + _metric_buckets.at(period)->to_prometheus(ser, add_labels); } void window_single_opentelemetry(metrics::v1::ScopeMetrics &scope, uint64_t period = 0, Metric::LabelMap add_labels = {}) const @@ -577,13 +578,13 @@ class AbstractMetricsManager sbucket->to_opentelemetry(scope, start_ts, end_ts, add_labels); } - void window_external_prometheus(std::stringstream &out, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) const + void window_external_prometheus(PrometheusSerializer &ser, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) const { if (_groups && _groups->none()) { return; } // static because caller guarantees only our own bucket type - static_cast(bucket)->to_prometheus(out, add_labels); + static_cast(bucket)->to_prometheus(ser, add_labels); } void window_external_json(json &j, const std::string &key, AbstractMetricsBucket *bucket) const diff --git a/src/CoreServer.cpp b/src/CoreServer.cpp index 0d31a1a9c..4054224e4 100644 --- a/src/CoreServer.cpp +++ b/src/CoreServer.cpp @@ -6,6 +6,7 @@ #include "HandlerManager.h" #include "Metrics.h" #include "Policies.h" +#include "PrometheusSerializer.h" #include "Taps.h" #include "visor_config.h" #include @@ -173,15 +174,17 @@ void CoreServer::_setup_routes(const PrometheusConfig &prom_config) } try { std::stringstream output; + PrometheusSerializer ser; auto [policy, lock] = _registry->policy_manager()->module_get_locked("default"); for (auto &mod : policy->modules()) { auto hmod = dynamic_cast(mod); if (hmod) { spdlog::stopwatch sw; - hmod->window_prometheus(output, {{"policy", "default"}}); + hmod->window_prometheus(ser, {{"policy", "default"}}); _logger->debug("{} window_prometheus elapsed time: {}", hmod->name(), sw); } } + output << ser.finalize(); res.set_content(output.str(), "text/plain"); } catch (const std::exception &e) { res.status = 500; diff --git a/src/Metrics.cpp b/src/Metrics.cpp index c08ed9234..0aba2fe73 100644 --- a/src/Metrics.cpp +++ b/src/Metrics.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "Metrics.h" +#include "PrometheusSerializer.h" #include namespace visor { @@ -12,11 +13,9 @@ void Counter::to_json(json &j) const name_json_assign(j, _value); } -void Counter::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void Counter::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " gauge" << std::endl; - out << name_snake({}, add_labels) << ' ' << _value << std::endl; + ser.write(base_name_snake(), PrometheusSerializer::Type::Gauge, _desc, {}, add_labels, _value); } void Counter::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, Metric::LabelMap add_labels) const @@ -49,10 +48,10 @@ void Rate::to_json(visor::json &j) const _quantile.to_json(j); } -void Rate::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void Rate::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { std::shared_lock lock(_sketch_mutex); - _quantile.to_prometheus(out, add_labels); + _quantile.to_prometheus(ser, add_labels); } void Rate::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, Metric::LabelMap add_labels) const @@ -72,11 +71,9 @@ void Cardinality::to_json(json &j) const { name_json_assign(j, lround(_set.get_estimate())); } -void Cardinality::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void Cardinality::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " gauge" << std::endl; - out << name_snake({}, add_labels) << ' ' << lround(_set.get_estimate()) << std::endl; + ser.write(base_name_snake(), PrometheusSerializer::Type::Gauge, _desc, {}, add_labels, lround(_set.get_estimate())); } void Cardinality::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, Metric::LabelMap add_labels) const @@ -129,32 +126,4 @@ std::string Metric::base_name_snake() const return name_text; } -std::string Metric::name_snake(std::initializer_list add_names, Metric::LabelMap add_labels) const -{ - std::string label_text{"{"}; - if (!prometheus_static_labels().empty()) { - for (const auto &[key, value] : prometheus_static_labels()) { - label_text.append(key + "=\"" + value + "\","); - } - } - if (add_labels.size()) { - for (const auto &[key, value] : add_labels) { - label_text.append(key + "=\"" + value + "\","); - } - } - if (label_text.back() == ',') { - label_text.pop_back(); - } - label_text.push_back('}'); - auto snake = [](const std::string &ss, const std::string &s) { - return ss.empty() ? s : ss + "_" + s; - }; - std::string name_text = _schema_key + "_" + std::accumulate(std::begin(_name), std::end(_name), std::string(), snake); - if (add_names.size()) { - name_text.push_back('_'); - name_text.append(std::accumulate(std::begin(add_names), std::end(add_names), std::string(), snake)); - } - return name_text + label_text; -} - } \ No newline at end of file diff --git a/src/Metrics.h b/src/Metrics.h index 2f5436caf..4d0e30bc0 100644 --- a/src/Metrics.h +++ b/src/Metrics.h @@ -4,6 +4,7 @@ #pragma once #include "MetricLabels.h" +#include "PrometheusSerializer.h" #include #include #include @@ -128,10 +129,9 @@ class Metric void name_json_assign(json &j, std::initializer_list add_names, const json &val) const; [[nodiscard]] std::string base_name_snake() const; - [[nodiscard]] std::string name_snake(std::initializer_list add_names = {}, LabelMap add_labels = {}) const; virtual void to_json(json &j) const = 0; - virtual void to_prometheus(std::stringstream &out, LabelMap add_labels = {}) const = 0; + virtual void to_prometheus(PrometheusSerializer &ser, LabelMap add_labels = {}) const = 0; virtual void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, LabelMap add_labels = {}) const = 0; }; @@ -177,7 +177,7 @@ class Counter final : public Metric // Metric void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, LabelMap add_labels = {}) const override; }; @@ -261,7 +261,7 @@ class Histogram final : public Metric name_json_assign(j, {"buckets", "+Inf"}, histogram[bins.size()] * _sketch.get_n()); } - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override { if (_sketch.is_empty()) { return; @@ -275,17 +275,16 @@ class Histogram final : public Metric } } auto histogram = _sketch.get_CDF(bins.data(), bins.size()); - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " histogram" << std::endl; + const auto base = base_name_snake(); for (std::size_t i = 0; i < bins.size(); ++i) { LabelMap le(add_labels); le["le"] = std::to_string(bins[i]); - out << name_snake({"bucket"}, le) << ' ' << histogram[i] * _sketch.get_n() << std::endl; + ser.write(base, PrometheusSerializer::Type::Histogram, _desc, {"bucket"}, le, histogram[i] * _sketch.get_n()); } LabelMap le(add_labels); le["le"] = "+Inf"; - out << name_snake({"bucket"}, le) << ' ' << histogram[bins.size()] * _sketch.get_n() << std::endl; - out << name_snake({"count"}, add_labels) << ' ' << _sketch.get_n() << std::endl; + ser.write(base, PrometheusSerializer::Type::Histogram, _desc, {"bucket"}, le, histogram[bins.size()] * _sketch.get_n()); + ser.write(base, PrometheusSerializer::Type::Histogram, _desc, {"count"}, add_labels, _sketch.get_n()); } void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, LabelMap add_labels = {}) const override @@ -413,7 +412,7 @@ class Quantile final : public Metric } } - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override { if (_quantile.is_empty()) { return; @@ -436,14 +435,13 @@ class Quantile final : public Metric l99["quantile"] = "0.99"; if (quantiles.size()) { - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " summary" << std::endl; - out << name_snake({}, l5) << ' ' << quantiles[0] << std::endl; - out << name_snake({}, l9) << ' ' << quantiles[1] << std::endl; - out << name_snake({}, l95) << ' ' << quantiles[2] << std::endl; - out << name_snake({}, l99) << ' ' << quantiles[3] << std::endl; - out << name_snake({"sum"}, add_labels) << ' ' << _quantile.get_max_item() << std::endl; - out << name_snake({"count"}, add_labels) << ' ' << _quantile.get_n() << std::endl; + const auto base = base_name_snake(); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {}, l5, quantiles[0]); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {}, l9, quantiles[1]); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {}, l95, quantiles[2]); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {}, l99, quantiles[3]); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {"sum"}, add_labels, _quantile.get_max_item()); + ser.write(base, PrometheusSerializer::Type::Summary, _desc, {"count"}, add_labels, _quantile.get_n()); } } @@ -611,7 +609,7 @@ class TopN final : public Metric name_json_assign(j, section); } - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels, std::function formatter) const + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels, std::function formatter) const { auto items = _fi.get_frequent_items(datasketches::frequent_items_error_type::NO_FALSE_NEGATIVES); if (!std::min(_top_count, items.size())) { @@ -619,19 +617,18 @@ class TopN final : public Metric } LabelMap l(add_labels); auto threshold = _get_threshold(items); - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " gauge" << std::endl; + const auto base = base_name_snake(); for (uint64_t i = 0; i < std::min(_top_count, items.size()); i++) { if (items[i].get_estimate() >= threshold) { l[_item_key] = formatter(items[i].get_item()); - out << name_snake({}, l) << ' ' << items[i].get_estimate() << std::endl; + ser.write(base, PrometheusSerializer::Type::Gauge, _desc, {}, l, items[i].get_estimate()); } else { break; } } } - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels, std::function formatter) const + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels, std::function formatter) const { auto items = _fi.get_frequent_items(datasketches::frequent_items_error_type::NO_FALSE_NEGATIVES); if (!std::min(_top_count, items.size())) { @@ -639,12 +636,11 @@ class TopN final : public Metric } LabelMap l(add_labels); auto threshold = _get_threshold(items); - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " gauge" << std::endl; + const auto base = base_name_snake(); for (uint64_t i = 0; i < std::min(_top_count, items.size()); i++) { if (items[i].get_estimate() >= threshold) { formatter(l, _item_key, items[i].get_item()); - out << name_snake({}, l) << ' ' << items[i].get_estimate() << std::endl; + ser.write(base, PrometheusSerializer::Type::Gauge, _desc, {}, l, items[i].get_estimate()); } else { break; } @@ -668,7 +664,7 @@ class TopN final : public Metric name_json_assign(j, section); } - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override { auto items = _fi.get_frequent_items(datasketches::frequent_items_error_type::NO_FALSE_NEGATIVES); if (!std::min(_top_count, items.size())) { @@ -676,14 +672,13 @@ class TopN final : public Metric } LabelMap l(add_labels); auto threshold = _get_threshold(items); - out << "# HELP " << base_name_snake() << ' ' << _desc << std::endl; - out << "# TYPE " << base_name_snake() << " gauge" << std::endl; + const auto base = base_name_snake(); for (uint64_t i = 0; i < std::min(_top_count, items.size()); i++) { if (items[i].get_estimate() >= threshold) { std::stringstream name_text; name_text << items[i].get_item(); l[_item_key] = name_text.str(); - out << name_snake({}, l) << ' ' << items[i].get_estimate() << std::endl; + ser.write(base, PrometheusSerializer::Type::Gauge, _desc, {}, l, items[i].get_estimate()); } else { break; } @@ -804,7 +799,7 @@ class Cardinality final : public Metric // Metric void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, LabelMap add_labels = {}) const override; }; @@ -905,7 +900,7 @@ class Rate final : public Metric // Metric void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, LabelMap add_labels = {}) const override; }; } \ No newline at end of file diff --git a/src/Policies.cpp b/src/Policies.cpp index b292be7f2..a0b908e0b 100644 --- a/src/Policies.cpp +++ b/src/Policies.cpp @@ -6,6 +6,7 @@ #include "CoreRegistry.h" #include "HandlerManager.h" #include "InputStreamManager.h" +#include "PrometheusSerializer.h" #include "Taps.h" #include #include @@ -381,21 +382,23 @@ void Policy::json_metrics(json &j, uint64_t period, bool merge) void Policy::prometheus_metrics(std::stringstream &out) { + PrometheusSerializer ser; if (_merge_like_handlers) { auto bucket_map = _get_merged_buckets(); for (auto &[bucket, hmod] : bucket_map) { - hmod->window_prometheus(out, bucket.get(), {{"policy", name()}, {"handler", hmod->schema_key() + "_merged"}}); + hmod->window_prometheus(ser, bucket.get(), {{"policy", name()}, {"handler", hmod->schema_key() + "_merged"}}); } } else { for (auto &mod : modules()) { auto hmod = dynamic_cast(mod); if (hmod) { spdlog::stopwatch sw; - hmod->window_prometheus(out, {{"policy", name()}, {"handler", hmod->name()}}); + hmod->window_prometheus(ser, {{"policy", name()}, {"handler", hmod->name()}}); spdlog::get("visor")->debug("{} window_prometheus elapsed time: {}", hmod->name(), sw); } } } + out << ser.finalize(); } void Policy::opentelemetry_metrics(metrics::v1::ScopeMetrics &scope) diff --git a/src/PrometheusSerializer.h b/src/PrometheusSerializer.h index fbb728316..a5bb192ee 100644 --- a/src/PrometheusSerializer.h +++ b/src/PrometheusSerializer.h @@ -45,6 +45,8 @@ class PrometheusSerializer it->second.series.push_back(std::move(line)); } + // Renders all accumulated families into the Prometheus text exposition format. + // Single-use per response: build one serializer, write all series, call finalize() once. std::string finalize() const { std::string out; diff --git a/src/StreamHandler.h b/src/StreamHandler.h index 9198d62f9..d0ada3389 100644 --- a/src/StreamHandler.h +++ b/src/StreamHandler.h @@ -8,6 +8,7 @@ #include "AbstractModule.h" #include "CoreRegistry.h" #include "InputEventProxy.h" +#include "PrometheusSerializer.h" #include #include #include @@ -71,8 +72,8 @@ class StreamHandler : public AbstractRunnableModule virtual void window_json(json &j, uint64_t period, bool merged) = 0; virtual void window_json(json &j, AbstractMetricsBucket *bucket) = 0; - virtual void window_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) = 0; - virtual void window_prometheus(std::stringstream &out, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) = 0; + virtual void window_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) = 0; + virtual void window_prometheus(PrometheusSerializer &ser, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) = 0; virtual void window_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) = 0; virtual void window_opentelemetry(metrics::v1::ScopeMetrics &scope, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) = 0; virtual std::unique_ptr merge(AbstractMetricsBucket *bucket, uint64_t period, bool prometheus, bool merged) = 0; @@ -224,18 +225,18 @@ class StreamMetricsHandler : public StreamHandler _metrics->window_external_json(j, schema_key(), bucket); } - void window_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) override + void window_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) override { if (_metrics->current_periods() > 1) { - _metrics->window_single_prometheus(out, 1, add_labels); + _metrics->window_single_prometheus(ser, 1, add_labels); } else { - _metrics->window_single_prometheus(out, 0, add_labels); + _metrics->window_single_prometheus(ser, 0, add_labels); } } - void window_prometheus(std::stringstream &out, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) override + void window_prometheus(PrometheusSerializer &ser, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) override { - _metrics->window_external_prometheus(out, bucket, add_labels); + _metrics->window_external_prometheus(ser, bucket, add_labels); }; void window_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) override diff --git a/src/handlers/bgp/BgpStreamHandler.cpp b/src/handlers/bgp/BgpStreamHandler.cpp index 3dd09b471..623747b41 100644 --- a/src/handlers/bgp/BgpStreamHandler.cpp +++ b/src/handlers/bgp/BgpStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "BgpStreamHandler.h" +#include "PrometheusSerializer.h" #include namespace visor::handler::bgp { @@ -165,28 +166,28 @@ void BgpMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Metric: _counters.filtered += other._counters.filtered; } -void BgpMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void BgpMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - _rate_total.to_prometheus(out, add_labels); + _rate_total.to_prometheus(ser, add_labels); { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); - _counters.OPEN.to_prometheus(out, add_labels); - _counters.UPDATE.to_prometheus(out, add_labels); - _counters.NOTIFICATION.to_prometheus(out, add_labels); - _counters.KEEPALIVE.to_prometheus(out, add_labels); - _counters.ROUTEREFRESH.to_prometheus(out, add_labels); - _counters.total.to_prometheus(out, add_labels); - _counters.filtered.to_prometheus(out, add_labels); + _counters.OPEN.to_prometheus(ser, add_labels); + _counters.UPDATE.to_prometheus(ser, add_labels); + _counters.NOTIFICATION.to_prometheus(ser, add_labels); + _counters.KEEPALIVE.to_prometheus(ser, add_labels); + _counters.ROUTEREFRESH.to_prometheus(ser, add_labels); + _counters.total.to_prometheus(ser, add_labels); + _counters.filtered.to_prometheus(ser, add_labels); } void BgpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const diff --git a/src/handlers/bgp/BgpStreamHandler.h b/src/handlers/bgp/BgpStreamHandler.h index cbce34155..6c9840750 100644 --- a/src/handlers/bgp/BgpStreamHandler.h +++ b/src/handlers/bgp/BgpStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "PcapInputStream.h" #include "StreamHandler.h" #ifdef __GNUC__ @@ -76,7 +77,7 @@ class BgpMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t, uint64_t) override { diff --git a/src/handlers/bgp/tests/test_bgp_layer.cpp b/src/handlers/bgp/tests/test_bgp_layer.cpp index 32b2ab5d1..2b0d56429 100644 --- a/src/handlers/bgp/tests/test_bgp_layer.cpp +++ b/src/handlers/bgp/tests/test_bgp_layer.cpp @@ -74,9 +74,9 @@ TEST_CASE("BGP to_prometheus and to_opentelemetry backends", "[pcap][bgp][backen // Counter values come from the existing parse test: total=9, OPEN=2, // UPDATE=4, KEEPALIVE=3. They must round-trip identically through both // backends. - std::stringstream prom; + visor::PrometheusSerializer prom; bgp_handler.metrics()->bucket(0)->to_prometheus(prom, {}); - auto prom_text = prom.str(); + auto prom_text = prom.finalize(); CHECK(prom_text.find("bgp_wire_packets_total{} 9") != std::string::npos); CHECK(prom_text.find("bgp_wire_packets_open{} 2") != std::string::npos); CHECK(prom_text.find("bgp_wire_packets_update{} 4") != std::string::npos); diff --git a/src/handlers/dhcp/DhcpStreamHandler.cpp b/src/handlers/dhcp/DhcpStreamHandler.cpp index 3287d109d..9680c5502 100644 --- a/src/handlers/dhcp/DhcpStreamHandler.cpp +++ b/src/handlers/dhcp/DhcpStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "DhcpStreamHandler.h" +#include "PrometheusSerializer.h" namespace visor::handler::dhcp { @@ -120,34 +121,34 @@ void DhcpMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Metric _dhcp_topServers.merge(other._dhcp_topServers); } -void DhcpMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void DhcpMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - _rate_total.to_prometheus(out, add_labels); + _rate_total.to_prometheus(ser, add_labels); { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); - _counters.DISCOVER.to_prometheus(out, add_labels); - _counters.OFFER.to_prometheus(out, add_labels); - _counters.REQUEST.to_prometheus(out, add_labels); - _counters.ACK.to_prometheus(out, add_labels); - _counters.SOLICIT.to_prometheus(out, add_labels); - _counters.ADVERTISE.to_prometheus(out, add_labels); - _counters.REQUESTV6.to_prometheus(out, add_labels); - _counters.REPLY.to_prometheus(out, add_labels); - _counters.total.to_prometheus(out, add_labels); - _counters.filtered.to_prometheus(out, add_labels); - - _dhcp_topClients.to_prometheus(out, add_labels); - _dhcp_topServers.to_prometheus(out, add_labels); + _counters.DISCOVER.to_prometheus(ser, add_labels); + _counters.OFFER.to_prometheus(ser, add_labels); + _counters.REQUEST.to_prometheus(ser, add_labels); + _counters.ACK.to_prometheus(ser, add_labels); + _counters.SOLICIT.to_prometheus(ser, add_labels); + _counters.ADVERTISE.to_prometheus(ser, add_labels); + _counters.REQUESTV6.to_prometheus(ser, add_labels); + _counters.REPLY.to_prometheus(ser, add_labels); + _counters.total.to_prometheus(ser, add_labels); + _counters.filtered.to_prometheus(ser, add_labels); + + _dhcp_topClients.to_prometheus(ser, add_labels); + _dhcp_topServers.to_prometheus(ser, add_labels); } void DhcpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const diff --git a/src/handlers/dhcp/DhcpStreamHandler.h b/src/handlers/dhcp/DhcpStreamHandler.h index a5a8363d2..e036ecc39 100644 --- a/src/handlers/dhcp/DhcpStreamHandler.h +++ b/src/handlers/dhcp/DhcpStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #ifdef __GNUC__ #pragma GCC diagnostic push #pragma GCC diagnostic ignored "-Wpedantic" @@ -98,7 +99,7 @@ class DhcpMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t, uint64_t) override { diff --git a/src/handlers/dhcp/tests/test_dhcp_layer.cpp b/src/handlers/dhcp/tests/test_dhcp_layer.cpp index 2fec8d736..0091bcda5 100644 --- a/src/handlers/dhcp/tests/test_dhcp_layer.cpp +++ b/src/handlers/dhcp/tests/test_dhcp_layer.cpp @@ -115,9 +115,9 @@ TEST_CASE("DHCP to_prometheus and to_opentelemetry backends", "[pcap][dhcp][back // Counter values come from "Parse DHCP tests": DISCOVER=1, OFFER=1, // REQUEST=3, ACK=3. Round-trip through both backends. - std::stringstream prom; + visor::PrometheusSerializer prom; dhcp_handler.metrics()->bucket(0)->to_prometheus(prom, {}); - auto prom_text = prom.str(); + auto prom_text = prom.finalize(); CHECK(prom_text.find("dhcp_wire_packets_discover{} 1") != std::string::npos); CHECK(prom_text.find("dhcp_wire_packets_offer{} 1") != std::string::npos); CHECK(prom_text.find("dhcp_wire_packets_request{} 3") != std::string::npos); diff --git a/src/handlers/dns/v1/DnsStreamHandler.cpp b/src/handlers/dns/v1/DnsStreamHandler.cpp index d027036b0..dd09fe354 100644 --- a/src/handlers/dns/v1/DnsStreamHandler.cpp +++ b/src/handlers/dns/v1/DnsStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "DnsStreamHandler.h" +#include "PrometheusSerializer.h" #include "HandlerModulePlugin.h" #include "utils.h" #include @@ -1136,91 +1137,91 @@ void DnsMetricsBucket::new_dns_transaction(bool deep, float to90th, float from90 } } } -void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void DnsMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - _rate_total.to_prometheus(out, add_labels); + _rate_total.to_prometheus(ser, add_labels); { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); if (group_enabled(group::DnsMetrics::Counters)) { - _counters.queries.to_prometheus(out, add_labels); - _counters.replies.to_prometheus(out, add_labels); - _counters.TCP.to_prometheus(out, add_labels); - _counters.UDP.to_prometheus(out, add_labels); - _counters.IPv4.to_prometheus(out, add_labels); - _counters.IPv6.to_prometheus(out, add_labels); - _counters.NX.to_prometheus(out, add_labels); - _counters.REFUSED.to_prometheus(out, add_labels); - _counters.SRVFAIL.to_prometheus(out, add_labels); - _counters.RNOERROR.to_prometheus(out, add_labels); - _counters.NODATA.to_prometheus(out, add_labels); - _counters.total.to_prometheus(out, add_labels); - _counters.filtered.to_prometheus(out, add_labels); + _counters.queries.to_prometheus(ser, add_labels); + _counters.replies.to_prometheus(ser, add_labels); + _counters.TCP.to_prometheus(ser, add_labels); + _counters.UDP.to_prometheus(ser, add_labels); + _counters.IPv4.to_prometheus(ser, add_labels); + _counters.IPv6.to_prometheus(ser, add_labels); + _counters.NX.to_prometheus(ser, add_labels); + _counters.REFUSED.to_prometheus(ser, add_labels); + _counters.SRVFAIL.to_prometheus(ser, add_labels); + _counters.RNOERROR.to_prometheus(ser, add_labels); + _counters.NODATA.to_prometheus(ser, add_labels); + _counters.total.to_prometheus(ser, add_labels); + _counters.filtered.to_prometheus(ser, add_labels); } if (group_enabled(group::DnsMetrics::Cardinality)) { - _dns_qnameCard.to_prometheus(out, add_labels); + _dns_qnameCard.to_prometheus(ser, add_labels); } if (group_enabled(group::DnsMetrics::DnsTransactions)) { - _counters.xacts_total.to_prometheus(out, add_labels); - _counters.xacts_timed_out.to_prometheus(out, add_labels); + _counters.xacts_total.to_prometheus(ser, add_labels); + _counters.xacts_timed_out.to_prometheus(ser, add_labels); - _counters.xacts_in.to_prometheus(out, add_labels); - _dns_slowXactIn.to_prometheus(out, add_labels); + _counters.xacts_in.to_prometheus(ser, add_labels); + _dns_slowXactIn.to_prometheus(ser, add_labels); if (group_enabled(group::DnsMetrics::Quantiles)) { - _dnsXactFromTimeUs.to_prometheus(out, add_labels); - _dnsXactToTimeUs.to_prometheus(out, add_labels); - _dnsXactRatio.to_prometheus(out, add_labels); + _dnsXactFromTimeUs.to_prometheus(ser, add_labels); + _dnsXactToTimeUs.to_prometheus(ser, add_labels); + _dnsXactRatio.to_prometheus(ser, add_labels); } if (group_enabled(group::DnsMetrics::Histograms)) { - _dnsXactFromHistTimeUs.to_prometheus(out, add_labels); - _dnsXactToHistTimeUs.to_prometheus(out, add_labels); + _dnsXactFromHistTimeUs.to_prometheus(ser, add_labels); + _dnsXactToHistTimeUs.to_prometheus(ser, add_labels); } - _counters.xacts_out.to_prometheus(out, add_labels); - _dns_slowXactOut.to_prometheus(out, add_labels); + _counters.xacts_out.to_prometheus(ser, add_labels); + _dns_slowXactOut.to_prometheus(ser, add_labels); } if (group_enabled(group::DnsMetrics::TopPorts)) { - _dns_topUDPPort.to_prometheus(out, add_labels, [](const uint16_t &val) { return std::to_string(val); }); + _dns_topUDPPort.to_prometheus(ser, add_labels, [](const uint16_t &val) { return std::to_string(val); }); } if (group_enabled(group::DnsMetrics::TopEcs)) { - group_enabled(group::DnsMetrics::Counters) ? _counters.queryECS.to_prometheus(out, add_labels) : void(); - _dns_topGeoLocECS.to_prometheus(out, add_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + group_enabled(group::DnsMetrics::Counters) ? _counters.queryECS.to_prometheus(ser, add_labels) : void(); + _dns_topGeoLocECS.to_prometheus(ser, add_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - _dns_topASNECS.to_prometheus(out, add_labels); - _dns_topQueryECS.to_prometheus(out, add_labels); + _dns_topASNECS.to_prometheus(ser, add_labels); + _dns_topQueryECS.to_prometheus(ser, add_labels); } if (group_enabled(group::DnsMetrics::TopQnames)) { - _dns_topQname2.to_prometheus(out, add_labels); - _dns_topQname3.to_prometheus(out, add_labels); - _dns_topNX.to_prometheus(out, add_labels); - _dns_topREFUSED.to_prometheus(out, add_labels); + _dns_topQname2.to_prometheus(ser, add_labels); + _dns_topQname3.to_prometheus(ser, add_labels); + _dns_topNX.to_prometheus(ser, add_labels); + _dns_topREFUSED.to_prometheus(ser, add_labels); - _dns_topSRVFAIL.to_prometheus(out, add_labels); - _dns_topNODATA.to_prometheus(out, add_labels); + _dns_topSRVFAIL.to_prometheus(ser, add_labels); + _dns_topNODATA.to_prometheus(ser, add_labels); if (group_enabled(group::DnsMetrics::TopQnamesDetails)) { - _dns_topSizedQnameResp.to_prometheus(out, add_labels); - _dns_topNOERROR.to_prometheus(out, add_labels); + _dns_topSizedQnameResp.to_prometheus(ser, add_labels); + _dns_topNOERROR.to_prometheus(ser, add_labels); } } - _dns_topRCode.to_prometheus(out, add_labels, [](const uint16_t &val) { + _dns_topRCode.to_prometheus(ser, add_labels, [](const uint16_t &val) { if (RCodeNames.find(val) != RCodeNames.end()) { return RCodeNames[val]; } else { @@ -1228,7 +1229,7 @@ void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad } }); - _dns_topQType.to_prometheus(out, add_labels, [](const uint16_t &val) { + _dns_topQType.to_prometheus(ser, add_labels, [](const uint16_t &val) { if (QTypeNames.find(val) != QTypeNames.end()) { return QTypeNames[val]; } else { diff --git a/src/handlers/dns/v1/DnsStreamHandler.h b/src/handlers/dns/v1/DnsStreamHandler.h index 1d1305f3b..f9390982c 100644 --- a/src/handlers/dns/v1/DnsStreamHandler.h +++ b/src/handlers/dns/v1/DnsStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "DnstapInputStream.h" #include "GeoDB.h" #include "MockInputStream.h" @@ -208,7 +209,7 @@ class DnsMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override { diff --git a/src/handlers/dns/v1/tests/test_dns_layer.cpp b/src/handlers/dns/v1/tests/test_dns_layer.cpp index 9ca14b40a..d3af8c892 100644 --- a/src/handlers/dns/v1/tests/test_dns_layer.cpp +++ b/src/handlers/dns/v1/tests/test_dns_layer.cpp @@ -1076,9 +1076,9 @@ TEST_CASE("dns to_prometheus and to_opentelemetry backends", "[pcap][dns][backen // Counter values match the existing "Parse DNS UDP IPv4 tests" case: // UDP=140, IPv4=140, queries=70, replies=70. - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - auto prom_text = prom.str(); + auto prom_text = prom.finalize(); CHECK(prom_text.find("dns_wire_packets_udp{} 140") != std::string::npos); CHECK(prom_text.find("dns_wire_packets_ipv4{} 140") != std::string::npos); CHECK(prom_text.find("dns_wire_packets_queries{} 70") != std::string::npos); diff --git a/src/handlers/dns/v2/DnsStreamHandler.cpp b/src/handlers/dns/v2/DnsStreamHandler.cpp index 1f1337841..29ca68a77 100644 --- a/src/handlers/dns/v2/DnsStreamHandler.cpp +++ b/src/handlers/dns/v2/DnsStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "DnsStreamHandler.h" +#include "PrometheusSerializer.h" #include "HandlerModulePlugin.h" #include "utils.h" #include @@ -756,53 +757,53 @@ void DnsMetricsBucket::to_json(json &j) const } } -void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void DnsMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { for (auto &dns : _dns) { auto dir_labels = add_labels; dir_labels["direction"] = _dir_str.at(dns.first); - group_enabled(group::DnsMetrics::Quantiles) ? dns.second.dnsRate.to_prometheus(out, dir_labels) : void(); + group_enabled(group::DnsMetrics::Quantiles) ? dns.second.dnsRate.to_prometheus(ser, dir_labels) : void(); } { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); - group_enabled(group::DnsMetrics::Counters) ? _filtered.to_prometheus(out, add_labels) : void(); + group_enabled(group::DnsMetrics::Counters) ? _filtered.to_prometheus(ser, add_labels) : void(); for (auto &dns : _dns) { auto dir_labels = add_labels; dir_labels["direction"] = _dir_str.at(dns.first); - group_enabled(group::DnsMetrics::Counters) ? dns.second.counters.to_prometheus(out, dir_labels) : void(); - group_enabled(group::DnsMetrics::Cardinality) ? dns.second.qnameCard.to_prometheus(out, dir_labels) : void(); - group_enabled(group::DnsMetrics::TopPorts) ? dns.second.topUDPPort.to_prometheus(out, dir_labels, [](const uint16_t &val) { return std::to_string(val); }) : void(); + group_enabled(group::DnsMetrics::Counters) ? dns.second.counters.to_prometheus(ser, dir_labels) : void(); + group_enabled(group::DnsMetrics::Cardinality) ? dns.second.qnameCard.to_prometheus(ser, dir_labels) : void(); + group_enabled(group::DnsMetrics::TopPorts) ? dns.second.topUDPPort.to_prometheus(ser, dir_labels, [](const uint16_t &val) { return std::to_string(val); }) : void(); if (group_enabled(group::DnsMetrics::TopEcs)) { - dns.second.topGeoLocECS.to_prometheus(out, dir_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + dns.second.topGeoLocECS.to_prometheus(ser, dir_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - dns.second.topASNECS.to_prometheus(out, dir_labels); - dns.second.topQueryECS.to_prometheus(out, dir_labels); + dns.second.topASNECS.to_prometheus(ser, dir_labels); + dns.second.topQueryECS.to_prometheus(ser, dir_labels); } if (group_enabled(group::DnsMetrics::TopRcodes)) { - dns.second.topNX.to_prometheus(out, dir_labels); - dns.second.topREFUSED.to_prometheus(out, dir_labels); - dns.second.topSRVFAIL.to_prometheus(out, dir_labels); - dns.second.topNODATA.to_prometheus(out, dir_labels); - dns.second.topNOERROR.to_prometheus(out, dir_labels); - dns.second.topRCode.to_prometheus(out, dir_labels, [](const uint16_t &val) { + dns.second.topNX.to_prometheus(ser, dir_labels); + dns.second.topREFUSED.to_prometheus(ser, dir_labels); + dns.second.topSRVFAIL.to_prometheus(ser, dir_labels); + dns.second.topNODATA.to_prometheus(ser, dir_labels); + dns.second.topNOERROR.to_prometheus(ser, dir_labels); + dns.second.topRCode.to_prometheus(ser, dir_labels, [](const uint16_t &val) { if (RCodeNames.find(val) != RCodeNames.end()) { return RCodeNames[val]; } else { @@ -812,17 +813,17 @@ void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad } if (group_enabled(group::DnsMetrics::TopQnames)) { - dns.second.topQname2.to_prometheus(out, dir_labels); - dns.second.topQname3.to_prometheus(out, dir_labels); + dns.second.topQname2.to_prometheus(ser, dir_labels); + dns.second.topQname3.to_prometheus(ser, dir_labels); } if (group_enabled(group::DnsMetrics::TopSize)) { - dns.second.topSizedQnameResp.to_prometheus(out, dir_labels); - dns.second.dnsRatio.to_prometheus(out, dir_labels); + dns.second.topSizedQnameResp.to_prometheus(ser, dir_labels); + dns.second.dnsRatio.to_prometheus(ser, dir_labels); } if (group_enabled(group::DnsMetrics::TopQtypes)) { - dns.second.topQType.to_prometheus(out, dir_labels, [](const uint16_t &val) { + dns.second.topQType.to_prometheus(ser, dir_labels, [](const uint16_t &val) { if (QTypeNames.find(val) != QTypeNames.end()) { return QTypeNames[val]; } else { @@ -832,9 +833,9 @@ void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad } if (group_enabled(group::DnsMetrics::XactTimes)) { - dns.second.dnsTimeUs.to_prometheus(out, dir_labels); - dns.second.dnsHistTimeUs.to_prometheus(out, dir_labels); - dns.second.topSlow.to_prometheus(out, dir_labels); + dns.second.dnsTimeUs.to_prometheus(ser, dir_labels); + dns.second.dnsHistTimeUs.to_prometheus(ser, dir_labels); + dns.second.topSlow.to_prometheus(ser, dir_labels); } } } diff --git a/src/handlers/dns/v2/DnsStreamHandler.h b/src/handlers/dns/v2/DnsStreamHandler.h index 17722652d..66f21dd97 100644 --- a/src/handlers/dns/v2/DnsStreamHandler.h +++ b/src/handlers/dns/v2/DnsStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "DnstapInputStream.h" #include "GeoDB.h" #include "MockInputStream.h" @@ -180,29 +181,29 @@ struct DnsDirection { orphan.to_json(j); } - void to_prometheus(std::stringstream &out, const Metric::LabelMap &add_labels) const + void to_prometheus(PrometheusSerializer &ser, const Metric::LabelMap &add_labels) const { - xacts.to_prometheus(out, add_labels); - UDP.to_prometheus(out, add_labels); - TCP.to_prometheus(out, add_labels); - DOT.to_prometheus(out, add_labels); - DOH.to_prometheus(out, add_labels); - cryptUDP.to_prometheus(out, add_labels); - cryptTCP.to_prometheus(out, add_labels); - DOQ.to_prometheus(out, add_labels); - IPv4.to_prometheus(out, add_labels); - IPv6.to_prometheus(out, add_labels); - NX.to_prometheus(out, add_labels); - ECS.to_prometheus(out, add_labels); - REFUSED.to_prometheus(out, add_labels); - SRVFAIL.to_prometheus(out, add_labels); - RNOERROR.to_prometheus(out, add_labels); - NODATA.to_prometheus(out, add_labels); - authData.to_prometheus(out, add_labels); - authAnswer.to_prometheus(out, add_labels); - checkDisabled.to_prometheus(out, add_labels); - timeout.to_prometheus(out, add_labels); - orphan.to_prometheus(out, add_labels); + xacts.to_prometheus(ser, add_labels); + UDP.to_prometheus(ser, add_labels); + TCP.to_prometheus(ser, add_labels); + DOT.to_prometheus(ser, add_labels); + DOH.to_prometheus(ser, add_labels); + cryptUDP.to_prometheus(ser, add_labels); + cryptTCP.to_prometheus(ser, add_labels); + DOQ.to_prometheus(ser, add_labels); + IPv4.to_prometheus(ser, add_labels); + IPv6.to_prometheus(ser, add_labels); + NX.to_prometheus(ser, add_labels); + ECS.to_prometheus(ser, add_labels); + REFUSED.to_prometheus(ser, add_labels); + SRVFAIL.to_prometheus(ser, add_labels); + RNOERROR.to_prometheus(ser, add_labels); + NODATA.to_prometheus(ser, add_labels); + authData.to_prometheus(ser, add_labels); + authAnswer.to_prometheus(ser, add_labels); + checkDisabled.to_prometheus(ser, add_labels); + timeout.to_prometheus(ser, add_labels); + orphan.to_prometheus(ser, add_labels); } void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, Metric::LabelMap add_labels) const @@ -371,7 +372,7 @@ class DnsMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override { diff --git a/src/handlers/dns/v2/tests/test_dns_layer.cpp b/src/handlers/dns/v2/tests/test_dns_layer.cpp index bb4b929ae..d6db6198f 100644 --- a/src/handlers/dns/v2/tests/test_dns_layer.cpp +++ b/src/handlers/dns/v2/tests/test_dns_layer.cpp @@ -998,9 +998,9 @@ TEST_CASE("dnsv2 to_prometheus and to_opentelemetry backends", "[pcap][dnsv2][ba // DNS v2 slices counters by `direction` label (in/out/unknown), so we // sum across all data points to get the project total. The fixture has // 70 query/reply pairs over UDP IPv4 → 70 xacts total. - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - CHECK(prom.str().find("dns_xacts{") != std::string::npos); + CHECK(prom.finalize().find("dns_xacts{") != std::string::npos); opentelemetry::proto::metrics::v1::ScopeMetrics scope; timespec start_ts{}, end_ts{}; diff --git a/src/handlers/flow/FlowStreamHandler.cpp b/src/handlers/flow/FlowStreamHandler.cpp index 0029fbdb5..416bdc09b 100644 --- a/src/handlers/flow/FlowStreamHandler.cpp +++ b/src/handlers/flow/FlowStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "FlowStreamHandler.h" +#include "PrometheusSerializer.h" #include "HandlerModulePlugin.h" #include "Tos.h" #include @@ -716,7 +717,7 @@ void FlowMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, [[mayb } } -void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void FlowMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { std::shared_lock r_lock(_mutex); @@ -733,12 +734,12 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a device_labels["device"] = deviceId; if (group_enabled(group::FlowMetrics::Counters)) { - device.second->total.to_prometheus(out, device_labels); - device.second->filtered.to_prometheus(out, device_labels); + device.second->total.to_prometheus(ser, device_labels); + device.second->filtered.to_prometheus(ser, device_labels); } if (group_enabled(group::FlowMetrics::ByBytes) && group_enabled(group::FlowMetrics::TopInterfaces)) { - device.second->topInIfIndexBytes.to_prometheus(out, device_labels, [dev](const uint32_t &val) { + device.second->topInIfIndexBytes.to_prometheus(ser, device_labels, [dev](const uint32_t &val) { if (dev) { if (auto it = dev->interfaces.find(val); it != dev->interfaces.end()) { return it->second.name; @@ -746,7 +747,7 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a } return std::to_string(val); }); - device.second->topOutIfIndexBytes.to_prometheus(out, device_labels, [dev](const uint32_t &val) { + device.second->topOutIfIndexBytes.to_prometheus(ser, device_labels, [dev](const uint32_t &val) { if (dev) { if (auto it = dev->interfaces.find(val); it != dev->interfaces.end()) { return it->second.name; @@ -757,7 +758,7 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a } if (group_enabled(group::FlowMetrics::ByPackets) && group_enabled(group::FlowMetrics::TopInterfaces)) { - device.second->topInIfIndexPackets.to_prometheus(out, device_labels, [dev](const uint32_t &val) { + device.second->topInIfIndexPackets.to_prometheus(ser, device_labels, [dev](const uint32_t &val) { if (dev) { if (auto it = dev->interfaces.find(val); it != dev->interfaces.end()) { return it->second.name; @@ -765,7 +766,7 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a } return std::to_string(val); }); - device.second->topOutIfIndexPackets.to_prometheus(out, device_labels, [dev](const uint32_t &val) { + device.second->topOutIfIndexPackets.to_prometheus(ser, device_labels, [dev](const uint32_t &val) { if (dev) { if (auto it = dev->interfaces.find(val); it != dev->interfaces.end()) { return it->second.name; @@ -787,12 +788,12 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a if (group_enabled(group::FlowMetrics::Cardinality)) { if (group_enabled(group::FlowMetrics::Conversations)) { - interface.second->conversationsCard.to_prometheus(out, device_labels); + interface.second->conversationsCard.to_prometheus(ser, device_labels); } - interface.second->srcIPCard.to_prometheus(out, interface_labels); - interface.second->dstIPCard.to_prometheus(out, interface_labels); - interface.second->srcPortCard.to_prometheus(out, interface_labels); - interface.second->dstPortCard.to_prometheus(out, interface_labels); + interface.second->srcIPCard.to_prometheus(ser, interface_labels); + interface.second->dstIPCard.to_prometheus(ser, interface_labels); + interface.second->srcPortCard.to_prometheus(ser, interface_labels); + interface.second->dstPortCard.to_prometheus(ser, interface_labels); } for (auto &count_dir : interface.second->counters) { @@ -803,12 +804,12 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a continue; } if (group_enabled(group::FlowMetrics::Counters)) { - count_dir.second.UDP.to_prometheus(out, interface_labels); - count_dir.second.TCP.to_prometheus(out, interface_labels); - count_dir.second.OtherL4.to_prometheus(out, interface_labels); - count_dir.second.IPv4.to_prometheus(out, interface_labels); - count_dir.second.IPv6.to_prometheus(out, interface_labels); - count_dir.second.total.to_prometheus(out, interface_labels); + count_dir.second.UDP.to_prometheus(ser, interface_labels); + count_dir.second.TCP.to_prometheus(ser, interface_labels); + count_dir.second.OtherL4.to_prometheus(ser, interface_labels); + count_dir.second.IPv4.to_prometheus(ser, interface_labels); + count_dir.second.IPv6.to_prometheus(ser, interface_labels); + count_dir.second.total.to_prometheus(ser, interface_labels); } } @@ -820,26 +821,26 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a continue; } if (group_enabled(group::FlowMetrics::TopIPs)) { - top_dir.second.topSrcIP.to_prometheus(out, interface_labels); - top_dir.second.topDstIP.to_prometheus(out, interface_labels); + top_dir.second.topSrcIP.to_prometheus(ser, interface_labels); + top_dir.second.topDstIP.to_prometheus(ser, interface_labels); } if (group_enabled(group::FlowMetrics::TopPorts)) { - top_dir.second.topSrcPort.to_prometheus(out, interface_labels); - top_dir.second.topDstPort.to_prometheus(out, interface_labels); + top_dir.second.topSrcPort.to_prometheus(ser, interface_labels); + top_dir.second.topDstPort.to_prometheus(ser, interface_labels); } if (group_enabled(group::FlowMetrics::TopIPPorts)) { - top_dir.second.topSrcIPPort.to_prometheus(out, interface_labels); - top_dir.second.topDstIPPort.to_prometheus(out, interface_labels); + top_dir.second.topSrcIPPort.to_prometheus(ser, interface_labels); + top_dir.second.topDstIPPort.to_prometheus(ser, interface_labels); } if (group_enabled(group::FlowMetrics::TopTos)) { - top_dir.second.topDSCP.to_prometheus(out, interface_labels, [](const uint8_t &val) { + top_dir.second.topDSCP.to_prometheus(ser, interface_labels, [](const uint8_t &val) { if (DscpNames.find(val) != DscpNames.end()) { return DscpNames[val]; } else { return std::to_string(val); } }); - top_dir.second.topECN.to_prometheus(out, interface_labels, [](const uint8_t &val) { + top_dir.second.topECN.to_prometheus(ser, interface_labels, [](const uint8_t &val) { if (EcnNames.find(val) != EcnNames.end()) { return EcnNames[val]; } else { @@ -851,33 +852,33 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a if (group_enabled(group::FlowMetrics::ByBytes)) { if (group_enabled(group::FlowMetrics::TopGeo)) { - interface.second->topN.first.topGeoLoc.to_prometheus(out, interface_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + interface.second->topN.first.topGeoLoc.to_prometheus(ser, interface_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - interface.second->topN.first.topASN.to_prometheus(out, interface_labels); + interface.second->topN.first.topASN.to_prometheus(ser, interface_labels); } if (group_enabled(group::FlowMetrics::Conversations)) { - interface.second->topN.first.topConversations.to_prometheus(out, interface_labels); + interface.second->topN.first.topConversations.to_prometheus(ser, interface_labels); } } if (group_enabled(group::FlowMetrics::ByPackets)) { if (group_enabled(group::FlowMetrics::TopGeo)) { - interface.second->topN.second.topGeoLoc.to_prometheus(out, interface_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + interface.second->topN.second.topGeoLoc.to_prometheus(ser, interface_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - interface.second->topN.second.topASN.to_prometheus(out, interface_labels); + interface.second->topN.second.topASN.to_prometheus(ser, interface_labels); } if (group_enabled(group::FlowMetrics::Conversations)) { - interface.second->topN.second.topConversations.to_prometheus(out, interface_labels); + interface.second->topN.second.topConversations.to_prometheus(ser, interface_labels); } } } diff --git a/src/handlers/flow/FlowStreamHandler.h b/src/handlers/flow/FlowStreamHandler.h index 523d25830..4061a02ad 100644 --- a/src/handlers/flow/FlowStreamHandler.h +++ b/src/handlers/flow/FlowStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "FlowInputStream.h" #include "GeoDB.h" #include "IpPort.h" @@ -273,7 +274,7 @@ class FlowMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override { diff --git a/src/handlers/flow/test_flows.cpp b/src/handlers/flow/test_flows.cpp index e2d6d3b47..fdc9a5104 100644 --- a/src/handlers/flow/test_flows.cpp +++ b/src/handlers/flow/test_flows.cpp @@ -9,6 +9,7 @@ #include "FlowInputStream.h" #include "FlowStreamHandler.h" #include "IpPort.h" +#include "PrometheusSerializer.h" using namespace visor::handler::flow; @@ -592,9 +593,9 @@ TEST_CASE("flow to_prometheus and to_opentelemetry backends", "[sflow][flow][bac handler.stop(); stream.stop(); - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - CHECK(prom.str().find("flow_") != std::string::npos); + CHECK(prom.finalize().find("flow_") != std::string::npos); opentelemetry::proto::metrics::v1::ScopeMetrics scope; timespec start_ts{}, end_ts{}; @@ -602,6 +603,49 @@ TEST_CASE("flow to_prometheus and to_opentelemetry backends", "[sflow][flow][bac CHECK(scope.metrics_size() > 0); } +TEST_CASE("flow prometheus: one HELP/TYPE per family, contiguous series (#716)", "[flow][prometheus][regression]") +{ + visor::input::flow::FlowInputStream stream{"flow-test"}; + stream.config_set("pcap_file", "tests/fixtures/ecmp.pcap"); + stream.config_set("flow_type", "sflow"); + + visor::Config c; + c.config_set("num_periods", 1); + auto stream_proxy = stream.add_event_proxy(c); + visor::handler::flow::FlowStreamHandler handler{"flow-test", stream_proxy, &c}; + // Enable every group so counters + top_ports emit per-interface families. + handler.config_set("enable", visor::Configurable::StringList({"all"})); + + handler.start(); + stream.start(); + handler.stop(); + stream.stop(); + + visor::PrometheusSerializer ser; + handler.metrics()->bucket(0)->to_prometheus(ser, {}); + const std::string out = ser.finalize(); + + const std::string family = "flow_top_out_dst_ports_bytes"; + auto count = [&](const std::string &needle) { + size_t n = 0, pos = 0; + while ((pos = out.find(needle, pos)) != std::string::npos) { + ++n; + pos += needle.size(); + } + return n; + }; + // Exactly one HELP and one TYPE for the family. + CHECK(count("# HELP " + family + " ") == 1); + CHECK(count("# TYPE " + family + " ") == 1); + // The fixture must emit this family for >=2 interfaces, else the test is vacuous. + CHECK(count(family + "{") >= 2); + // Series are contiguous: no other family's "# TYPE " appears between first and last series. + auto first = out.find(family + "{"); + auto last = out.rfind(family + "{"); + REQUIRE(first != std::string::npos); + CHECK(out.find("# TYPE ", first) > last); +} + TEST_CASE("Flow specialized_merge + to_prometheus + to_opentelemetry with all groups enabled", "[sflow][flow][unit]") { auto build = [](const std::string &name, @@ -650,11 +694,11 @@ TEST_CASE("Flow specialized_merge + to_prometheus + to_opentelemetry with all gr // After merging both runs of ecmp.pcap, the flow records counter must equal // the sum of the two input buckets' counts. - std::stringstream prom; + visor::PrometheusSerializer prom; target->to_prometheus(prom, {}); // Flow's prometheus output decorates per-device/per-interface labels, so // grep the line by name+value rather than an exact-prefix match. - CHECK(prom.str().find("flow_records_flows") != std::string::npos); + CHECK(prom.finalize().find("flow_records_flows") != std::string::npos); opentelemetry::proto::metrics::v1::ScopeMetrics scope; timespec start_ts{}, end_ts{}; diff --git a/src/handlers/input_resources/InputResourcesStreamHandler.cpp b/src/handlers/input_resources/InputResourcesStreamHandler.cpp index e0c1eb593..d5adb62d3 100644 --- a/src/handlers/input_resources/InputResourcesStreamHandler.cpp +++ b/src/handlers/input_resources/InputResourcesStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "InputResourcesStreamHandler.h" +#include "PrometheusSerializer.h" #include "Policies.h" namespace visor::handler::resources { @@ -147,22 +148,22 @@ void InputResourcesMetricsBucket::specialized_merge(const AbstractMetricsBucket } } -void InputResourcesMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void InputResourcesMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); - _cpu_usage.to_prometheus(out, add_labels); - _memory_bytes.to_prometheus(out, add_labels); - _policy_count.to_prometheus(out, add_labels); - _handler_count.to_prometheus(out, add_labels); + _cpu_usage.to_prometheus(ser, add_labels); + _memory_bytes.to_prometheus(ser, add_labels); + _policy_count.to_prometheus(ser, add_labels); + _handler_count.to_prometheus(ser, add_labels); } void InputResourcesMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const diff --git a/src/handlers/input_resources/InputResourcesStreamHandler.h b/src/handlers/input_resources/InputResourcesStreamHandler.h index 71c0c8e16..b127ed6c5 100644 --- a/src/handlers/input_resources/InputResourcesStreamHandler.h +++ b/src/handlers/input_resources/InputResourcesStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "DnstapInputStream.h" #include "FlowInputStream.h" #include "MockInputStream.h" @@ -51,7 +52,7 @@ class InputResourcesMetricsBucket final : public visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t, uint64_t) override { diff --git a/src/handlers/input_resources/test_resources_layer.cpp b/src/handlers/input_resources/test_resources_layer.cpp index 26f2f5112..ea2855583 100644 --- a/src/handlers/input_resources/test_resources_layer.cpp +++ b/src/handlers/input_resources/test_resources_layer.cpp @@ -47,7 +47,9 @@ TEST_CASE("Check resources for pcap input", "[pcap][resources]") std::stringstream output; std::string line; - resources_handler.metrics()->bucket(0)->to_prometheus(output, {{"policy", "default"}}); + visor::PrometheusSerializer ser; + resources_handler.metrics()->bucket(0)->to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP base_total Total number of events"); std::getline(output, line); @@ -129,11 +131,11 @@ TEST_CASE("input_resources to_prometheus and to_opentelemetry backends", "[pcap] handler.stop(); stream.stop(); - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); // input_resources emits cross-schema metrics (base_, cpu_usage, memory_bytes, // etc.) so just assert the backend produced something. - CHECK(!prom.str().empty()); + CHECK(!prom.finalize().empty()); opentelemetry::proto::metrics::v1::ScopeMetrics scope; timespec start_ts{}, end_ts{}; diff --git a/src/handlers/net/v1/NetStreamHandler.cpp b/src/handlers/net/v1/NetStreamHandler.cpp index 3d88a7e3d..e47fb254b 100644 --- a/src/handlers/net/v1/NetStreamHandler.cpp +++ b/src/handlers/net/v1/NetStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "NetStreamHandler.h" +#include "PrometheusSerializer.h" #include "HandlerModulePlugin.h" #include "utils.h" @@ -328,62 +329,62 @@ void NetworkMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Met _payload_size.merge(other._payload_size, agg_operator); } -void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void NetworkMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { - _rate_in.to_prometheus(out, add_labels); - _rate_out.to_prometheus(out, add_labels); - _rate_total.to_prometheus(out, add_labels); - _throughput_in.to_prometheus(out, add_labels); - _throughput_out.to_prometheus(out, add_labels); - _throughput_total.to_prometheus(out, add_labels); + _rate_in.to_prometheus(ser, add_labels); + _rate_out.to_prometheus(ser, add_labels); + _rate_total.to_prometheus(ser, add_labels); + _throughput_in.to_prometheus(ser, add_labels); + _throughput_out.to_prometheus(ser, add_labels); + _throughput_total.to_prometheus(ser, add_labels); { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); if (group_enabled(group::NetMetrics::Counters)) { - _counters.UDP.to_prometheus(out, add_labels); - _counters.TCP.to_prometheus(out, add_labels); - _counters.TCP_SYN.to_prometheus(out, add_labels); - _counters.OtherL4.to_prometheus(out, add_labels); - _counters.IPv4.to_prometheus(out, add_labels); - _counters.IPv6.to_prometheus(out, add_labels); - _counters.total_in.to_prometheus(out, add_labels); - _counters.total_out.to_prometheus(out, add_labels); - _counters.total_unk.to_prometheus(out, add_labels); - _counters.total.to_prometheus(out, add_labels); - _counters.filtered.to_prometheus(out, add_labels); + _counters.UDP.to_prometheus(ser, add_labels); + _counters.TCP.to_prometheus(ser, add_labels); + _counters.TCP_SYN.to_prometheus(ser, add_labels); + _counters.OtherL4.to_prometheus(ser, add_labels); + _counters.IPv4.to_prometheus(ser, add_labels); + _counters.IPv6.to_prometheus(ser, add_labels); + _counters.total_in.to_prometheus(ser, add_labels); + _counters.total_out.to_prometheus(ser, add_labels); + _counters.total_unk.to_prometheus(ser, add_labels); + _counters.total.to_prometheus(ser, add_labels); + _counters.filtered.to_prometheus(ser, add_labels); } if (group_enabled(group::NetMetrics::Cardinality)) { - _srcIPCard.to_prometheus(out, add_labels); - _dstIPCard.to_prometheus(out, add_labels); + _srcIPCard.to_prometheus(ser, add_labels); + _dstIPCard.to_prometheus(ser, add_labels); } if (group_enabled(group::NetMetrics::TopIps)) { - _topIPv4.to_prometheus(out, add_labels, [](const uint32_t &val) { return pcpp::IPv4Address(val).toString(); }); - _topIPv6.to_prometheus(out, add_labels); + _topIPv4.to_prometheus(ser, add_labels, [](const uint32_t &val) { return pcpp::IPv4Address(val).toString(); }); + _topIPv6.to_prometheus(ser, add_labels); } if (group_enabled(group::NetMetrics::TopGeo)) { - _topGeoLoc.to_prometheus(out, add_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + _topGeoLoc.to_prometheus(ser, add_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - _topASN.to_prometheus(out, add_labels); + _topASN.to_prometheus(ser, add_labels); } - _payload_size.to_prometheus(out, add_labels); + _payload_size.to_prometheus(ser, add_labels); } void NetworkMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const diff --git a/src/handlers/net/v1/NetStreamHandler.h b/src/handlers/net/v1/NetStreamHandler.h index 5b3ae58e2..29959221e 100644 --- a/src/handlers/net/v1/NetStreamHandler.h +++ b/src/handlers/net/v1/NetStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "DnstapInputStream.h" #include "GeoDB.h" #include "MockInputStream.h" @@ -137,7 +138,7 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override { diff --git a/src/handlers/net/v1/tests/test_net_layer.cpp b/src/handlers/net/v1/tests/test_net_layer.cpp index 3464bd812..900879ae9 100644 --- a/src/handlers/net/v1/tests/test_net_layer.cpp +++ b/src/handlers/net/v1/tests/test_net_layer.cpp @@ -600,9 +600,9 @@ TEST_CASE("net to_prometheus and to_opentelemetry backends", "[pcap][net][backen // Counter values match the existing "Parse net (dns) UDP IPv4 tests" // case for the same fixture: UDP=140, IPv4=140, IPv6=0. - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - auto prom_text = prom.str(); + auto prom_text = prom.finalize(); CHECK(prom_text.find("packets_udp{} 140") != std::string::npos); CHECK(prom_text.find("packets_ipv4{} 140") != std::string::npos); CHECK(prom_text.find("packets_ipv6{} 0") != std::string::npos); diff --git a/src/handlers/net/v2/NetStreamHandler.cpp b/src/handlers/net/v2/NetStreamHandler.cpp index bc337ab0f..d35403fc8 100644 --- a/src/handlers/net/v2/NetStreamHandler.cpp +++ b/src/handlers/net/v2/NetStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "NetStreamHandler.h" +#include "PrometheusSerializer.h" #include "HandlerModulePlugin.h" #include "utils.h" @@ -329,55 +330,55 @@ void NetworkMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Met } } -void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void NetworkMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { if (group_enabled(group::NetMetrics::Quantiles)) { for (auto &net : _net) { auto dir_labels = add_labels; dir_labels["direction"] = _dir_str.at(net.first); - net.second.rate.to_prometheus(out, dir_labels); - net.second.throughput.to_prometheus(out, dir_labels); + net.second.rate.to_prometheus(ser, dir_labels); + net.second.throughput.to_prometheus(ser, dir_labels); } } { auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe - event_rate->to_prometheus(out, add_labels); - num_events->to_prometheus(out, add_labels); - num_samples->to_prometheus(out, add_labels); + event_rate->to_prometheus(ser, add_labels); + num_events->to_prometheus(ser, add_labels); + num_samples->to_prometheus(ser, add_labels); } std::shared_lock r_lock(_mutex); - group_enabled(group::NetMetrics::Counters) ? _filtered.to_prometheus(out, add_labels) : void(); + group_enabled(group::NetMetrics::Counters) ? _filtered.to_prometheus(ser, add_labels) : void(); for (auto &net : _net) { auto dir_labels = add_labels; dir_labels["direction"] = _dir_str.at(net.first); - group_enabled(group::NetMetrics::Counters) ? net.second.counters.to_prometheus(out, dir_labels) : void(); + group_enabled(group::NetMetrics::Counters) ? net.second.counters.to_prometheus(ser, dir_labels) : void(); - group_enabled(group::NetMetrics::Cardinality) ? net.second.ipCard.to_prometheus(out, dir_labels) : void(); + group_enabled(group::NetMetrics::Cardinality) ? net.second.ipCard.to_prometheus(ser, dir_labels) : void(); if (group_enabled(group::NetMetrics::TopIps)) { - net.second.topIPv4.to_prometheus(out, dir_labels, [](const uint32_t &val) { return pcpp::IPv4Address(val).toString(); }); - net.second.topIPv6.to_prometheus(out, dir_labels); + net.second.topIPv4.to_prometheus(ser, dir_labels, [](const uint32_t &val) { return pcpp::IPv4Address(val).toString(); }); + net.second.topIPv6.to_prometheus(ser, dir_labels); } if (group_enabled(group::NetMetrics::TopGeo)) { - net.second.topGeoLoc.to_prometheus(out, dir_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { + net.second.topGeoLoc.to_prometheus(ser, dir_labels, [](Metric::LabelMap &l, const std::string &key, const visor::geo::City &val) { l[key] = val.location; if (!val.latitude.empty() && !val.longitude.empty()) { l["lat"] = val.latitude; l["lon"] = val.longitude; } }); - net.second.topASN.to_prometheus(out, dir_labels); + net.second.topASN.to_prometheus(ser, dir_labels); } - group_enabled(group::NetMetrics::Quantiles) ? net.second.payload_size.to_prometheus(out, dir_labels) : void(); + group_enabled(group::NetMetrics::Quantiles) ? net.second.payload_size.to_prometheus(ser, dir_labels) : void(); } } diff --git a/src/handlers/net/v2/NetStreamHandler.h b/src/handlers/net/v2/NetStreamHandler.h index c9f3c8834..bd50548a9 100644 --- a/src/handlers/net/v2/NetStreamHandler.h +++ b/src/handlers/net/v2/NetStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "DnstapInputStream.h" #include "GeoDB.h" #include "MockInputStream.h" @@ -99,15 +100,15 @@ struct NetworkDirection { total.to_json(j); } - void to_prometheus(std::stringstream &out, const Metric::LabelMap &add_labels) const + void to_prometheus(PrometheusSerializer &ser, const Metric::LabelMap &add_labels) const { - UDP.to_prometheus(out, add_labels); - TCP.to_prometheus(out, add_labels); - OtherL4.to_prometheus(out, add_labels); - IPv4.to_prometheus(out, add_labels); - IPv6.to_prometheus(out, add_labels); - TCP_SYN.to_prometheus(out, add_labels); - total.to_prometheus(out, add_labels); + UDP.to_prometheus(ser, add_labels); + TCP.to_prometheus(ser, add_labels); + OtherL4.to_prometheus(ser, add_labels); + IPv4.to_prometheus(ser, add_labels); + IPv6.to_prometheus(ser, add_labels); + TCP_SYN.to_prometheus(ser, add_labels); + total.to_prometheus(ser, add_labels); } void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start, timespec &end, Metric::LabelMap add_labels) const @@ -190,7 +191,7 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override { diff --git a/src/handlers/net/v2/tests/test_net_layer.cpp b/src/handlers/net/v2/tests/test_net_layer.cpp index a06611d80..83c553d8d 100644 --- a/src/handlers/net/v2/tests/test_net_layer.cpp +++ b/src/handlers/net/v2/tests/test_net_layer.cpp @@ -593,9 +593,9 @@ TEST_CASE("netv2 to_prometheus and to_opentelemetry backends", "[pcap][netv2][ba // v2 slices counters by `direction` label, so individual metric lines // in prom output look like `net_udp_packets{direction="out"} 140` — sum // across directions for the project total. - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - auto prom_text = prom.str(); + auto prom_text = prom.finalize(); CHECK(prom_text.find("net_udp_packets{") != std::string::npos); CHECK(prom_text.find("net_ipv4_packets{") != std::string::npos); @@ -645,9 +645,9 @@ TEST_CASE("Net v2 process_net_layer shallow overload + specialized_merge", "[net // individual per-direction packet counts that to_prometheus emits. REQUIRE_NOTHROW(b1->specialized_merge(*b2, visor::Metric::Aggregate::DEFAULT)); - std::stringstream prom_after; + visor::PrometheusSerializer prom_after; b1->to_prometheus(prom_after, {}); - auto prom_after_text = prom_after.str(); + auto prom_after_text = prom_after.finalize(); // 4 calls total to process_net_layer across the two buckets, verify the // summed packet count via the otel backend (more robust to label fmt). opentelemetry::proto::metrics::v1::ScopeMetrics scope_after; diff --git a/src/handlers/netprobe/NetProbeStreamHandler.cpp b/src/handlers/netprobe/NetProbeStreamHandler.cpp index b843b8c9e..ea01aff93 100644 --- a/src/handlers/netprobe/NetProbeStreamHandler.cpp +++ b/src/handlers/netprobe/NetProbeStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "NetProbeStreamHandler.h" +#include "PrometheusSerializer.h" namespace visor::handler::netprobe { @@ -136,7 +137,7 @@ void NetProbeMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, Me } } -void NetProbeMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void NetProbeMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { std::shared_lock r_lock(_mutex); @@ -146,11 +147,11 @@ void NetProbeMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelM target_labels["target"] = targetId; if (group_enabled(group::NetProbeMetrics::Counters)) { - target.second->attempts.to_prometheus(out, target_labels); - target.second->successes.to_prometheus(out, target_labels); - target.second->connect_failures.to_prometheus(out, target_labels); - target.second->dns_failures.to_prometheus(out, target_labels); - target.second->timed_out.to_prometheus(out, target_labels); + target.second->attempts.to_prometheus(ser, target_labels); + target.second->successes.to_prometheus(ser, target_labels); + target.second->connect_failures.to_prometheus(ser, target_labels); + target.second->dns_failures.to_prometheus(ser, target_labels); + target.second->timed_out.to_prometheus(ser, target_labels); } bool h_max_min{true}; @@ -161,12 +162,12 @@ void NetProbeMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelM if (group_enabled(group::NetProbeMetrics::Counters)) { target.second->minimum += target.second->h_time_us.get_min(); - target.second->minimum.to_prometheus(out, target_labels); + target.second->minimum.to_prometheus(ser, target_labels); target.second->maximum += target.second->h_time_us.get_max(); - target.second->maximum.to_prometheus(out, target_labels); + target.second->maximum.to_prometheus(ser, target_labels); } - target.second->h_time_us.to_prometheus(out, target_labels); + target.second->h_time_us.to_prometheus(ser, target_labels); } catch (const std::exception &) { h_max_min = false; } @@ -181,11 +182,11 @@ void NetProbeMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelM target.second->maximum.clear(); target.second->minimum += target.second->q_time_us.get_min(); - target.second->minimum.to_prometheus(out, target_labels); + target.second->minimum.to_prometheus(ser, target_labels); target.second->maximum += target.second->q_time_us.get_max(); - target.second->maximum.to_prometheus(out, target_labels); + target.second->maximum.to_prometheus(ser, target_labels); } - target.second->q_time_us.to_prometheus(out, target_labels); + target.second->q_time_us.to_prometheus(ser, target_labels); } catch (const std::exception &) { } } diff --git a/src/handlers/netprobe/NetProbeStreamHandler.h b/src/handlers/netprobe/NetProbeStreamHandler.h index 4cb83bf01..c46dc61c9 100644 --- a/src/handlers/netprobe/NetProbeStreamHandler.h +++ b/src/handlers/netprobe/NetProbeStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "NetProbeInputStream.h" #include "StreamHandler.h" #include "TransactionManager.h" @@ -82,7 +83,7 @@ class NetProbeMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t, uint64_t) override { diff --git a/src/handlers/netprobe/test_net_probe.cpp b/src/handlers/netprobe/test_net_probe.cpp index 84b665138..6b9a6a322 100644 --- a/src/handlers/netprobe/test_net_probe.cpp +++ b/src/handlers/netprobe/test_net_probe.cpp @@ -358,9 +358,9 @@ TEST_CASE("NetProbe to_prometheus emits configured metrics", "[netprobe][unit]") fx.manager()->process_failure(ErrorType::Timeout, "tprom"); - std::stringstream out; + visor::PrometheusSerializer out; fx.manager()->bucket(0)->to_prometheus(out, {}); - auto s = out.str(); + auto s = out.finalize(); // Counter name is registered in Target ctor as "packets_timeout". CHECK(s.find("packets_timeout") != std::string::npos); diff --git a/src/handlers/pcap/PcapStreamHandler.cpp b/src/handlers/pcap/PcapStreamHandler.cpp index faf3ae649..9bd189387 100644 --- a/src/handlers/pcap/PcapStreamHandler.cpp +++ b/src/handlers/pcap/PcapStreamHandler.cpp @@ -3,6 +3,7 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ #include "PcapStreamHandler.h" +#include "PrometheusSerializer.h" namespace visor::handler::pcap { @@ -89,13 +90,13 @@ void PcapMetricsBucket::specialized_merge(const AbstractMetricsBucket &o, [[mayb _counters.pcap_if_drop += other._counters.pcap_if_drop; } -void PcapMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const +void PcapMetricsBucket::to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels) const { std::shared_lock r_lock(_mutex); - _counters.pcap_TCP_reassembly_errors.to_prometheus(out, add_labels); - _counters.pcap_os_drop.to_prometheus(out, add_labels); - _counters.pcap_if_drop.to_prometheus(out, add_labels); + _counters.pcap_TCP_reassembly_errors.to_prometheus(ser, add_labels); + _counters.pcap_os_drop.to_prometheus(ser, add_labels); + _counters.pcap_if_drop.to_prometheus(ser, add_labels); } void PcapMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const diff --git a/src/handlers/pcap/PcapStreamHandler.h b/src/handlers/pcap/PcapStreamHandler.h index 06602ef98..196206498 100644 --- a/src/handlers/pcap/PcapStreamHandler.h +++ b/src/handlers/pcap/PcapStreamHandler.h @@ -5,6 +5,7 @@ #pragma once #include "AbstractMetricsManager.h" +#include "PrometheusSerializer.h" #include "PcapInputStream.h" #include "StreamHandler.h" #include @@ -57,7 +58,7 @@ class PcapMetricsBucket final : public visor::AbstractMetricsBucket // visor::AbstractMetricsBucket void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override; void to_json(json &j) const override; - void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override; + void to_prometheus(PrometheusSerializer &ser, Metric::LabelMap add_labels = {}) const override; void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override; void update_topn_metrics(size_t, uint64_t) override { diff --git a/src/handlers/pcap/tests/test_pcap_layer.cpp b/src/handlers/pcap/tests/test_pcap_layer.cpp index f8a3f57d9..cd72e1101 100644 --- a/src/handlers/pcap/tests/test_pcap_layer.cpp +++ b/src/handlers/pcap/tests/test_pcap_layer.cpp @@ -57,9 +57,9 @@ TEST_CASE("pcap to_prometheus and to_opentelemetry backends", "[pcap][pcap][back handler.stop(); stream.stop(); - std::stringstream prom; + visor::PrometheusSerializer prom; handler.metrics()->bucket(0)->to_prometheus(prom, {}); - CHECK(prom.str().find("pcap_") != std::string::npos); + CHECK(prom.finalize().find("pcap_") != std::string::npos); opentelemetry::proto::metrics::v1::ScopeMetrics scope; timespec start_ts{}, end_ts{}; diff --git a/src/tests/test_handlers.cpp b/src/tests/test_handlers.cpp index 6e96d3392..6a0eb33b6 100644 --- a/src/tests/test_handlers.cpp +++ b/src/tests/test_handlers.cpp @@ -16,7 +16,7 @@ class HandlerBucket : public AbstractMetricsBucket std::string type; void specialized_merge(const AbstractMetricsBucket &, Metric::Aggregate) override{}; void to_json(json &) const override{}; - void to_prometheus(std::stringstream &, + void to_prometheus(PrometheusSerializer &, Metric::LabelMap) const override{}; void to_opentelemetry(metrics::v1::ScopeMetrics &, timespec &, timespec &, Metric::LabelMap) const override{}; void update_topn_metrics(size_t, uint64_t) override{}; @@ -43,12 +43,12 @@ class TestHandlerMetricsManager : public AbstractMetricsManager { j["window"] = "single"; } - void window_single_prometheus(std::stringstream &out, uint64_t period, Metric::LabelMap) const + void window_single_prometheus(PrometheusSerializer &ser, uint64_t period, Metric::LabelMap) const { if (period) { - out << "first_window"; + ser.write("first_window", PrometheusSerializer::Type::Gauge, "", {}, {}, 1); } else { - out << "live_window"; + ser.write("live_window", PrometheusSerializer::Type::Gauge, "", {}, {}, 1); } } void window_single_opentelemetry(metrics::v1::ScopeMetrics &scope, uint64_t period, Metric::LabelMap) const @@ -63,9 +63,9 @@ class TestHandlerMetricsManager : public AbstractMetricsManager { scope.add_metrics()->set_name("external_window"); } - void window_external_prometheus(std::stringstream &out, AbstractMetricsBucket *, Metric::LabelMap) const + void window_external_prometheus(PrometheusSerializer &ser, AbstractMetricsBucket *, Metric::LabelMap) const { - out << "external_window"; + ser.write("external_window", PrometheusSerializer::Type::Gauge, "", {}, {}, 1); } void window_external_json(json &j, const std::string &, AbstractMetricsBucket *) const { @@ -170,26 +170,27 @@ TEST_CASE("StreamMetricsHandler tests", "[metrics][handler]") SECTION("Prometheus window") { - std::string line; - std::stringstream out; config.config_set("period", 0); auto handler = std::make_unique("my_handler", &config); - handler->window_prometheus(out); - std::getline(out, line); - CHECK(line == "live_window"); - out.clear(); + { + PrometheusSerializer ser; + handler->window_prometheus(ser); + CHECK(ser.finalize().find("live_window") != std::string::npos); + } config.config_set("period", 3); handler = std::make_unique("my_handler", &config); - handler->window_prometheus(out); - std::getline(out, line); - CHECK(line == "first_window"); - out.clear(); - - handler->window_prometheus(out, nullptr); - std::getline(out, line); - CHECK(line == "external_window"); - out.clear(); + { + PrometheusSerializer ser; + handler->window_prometheus(ser); + CHECK(ser.finalize().find("first_window") != std::string::npos); + } + + { + PrometheusSerializer ser; + handler->window_prometheus(ser, nullptr); + CHECK(ser.finalize().find("external_window") != std::string::npos); + } } SECTION("Opentelemetry window") diff --git a/src/tests/test_metrics.cpp b/src/tests/test_metrics.cpp index 1c263ec2b..529d8329e 100644 --- a/src/tests/test_metrics.cpp +++ b/src/tests/test_metrics.cpp @@ -14,10 +14,10 @@ class TestMetricsBucket final : public AbstractMetricsBucket void to_json([[maybe_unused]] json &j) const { } - void to_prometheus([[maybe_unused]] std::stringstream &out, + void to_prometheus([[maybe_unused]] PrometheusSerializer &ser, [[maybe_unused]] Metric::LabelMap add_labels = {}) const { - out << "test_performed" << std::endl; + ser.write("test_performed", PrometheusSerializer::Type::Gauge, "", {}, {}, 1); } void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &, timespec &, Metric::LabelMap) const { @@ -67,9 +67,9 @@ TEST_CASE("Abstract metrics manager", "[metrics][abstract]") SECTION("Abstract window single prometheus") { - manager->window_single_prometheus(output, 0, {{"policy", "default"}}); - std::getline(output, line); - CHECK(line == "test_performed"); + PrometheusSerializer ser; + manager->window_single_prometheus(ser, 0, {{"policy", "default"}}); + CHECK(ser.finalize().find("test_performed") != std::string::npos); } SECTION("Abstract window single opentelemetry") @@ -80,7 +80,8 @@ TEST_CASE("Abstract metrics manager", "[metrics][abstract]") SECTION("Abstract window single prometheus failed") { - CHECK_THROWS_WITH(manager->window_single_prometheus(output, 2, {{"policy", "default"}}), + PrometheusSerializer ser; + CHECK_THROWS_WITH(manager->window_single_prometheus(ser, 2, {{"policy", "default"}}), "invalid metrics period, specify [0, 0]"); } @@ -99,9 +100,9 @@ TEST_CASE("Abstract metrics manager", "[metrics][abstract]") SECTION("Abstract window external prometheus") { auto live = static_cast(manager->live_bucket()); - manager->window_external_prometheus(output, live, {{"policy", "default"}}); - std::getline(output, line); - CHECK(line == "test_performed"); + PrometheusSerializer ser; + manager->window_external_prometheus(ser, live, {{"policy", "default"}}); + CHECK(ser.finalize().find("test_performed") != std::string::npos); } SECTION("Abstract simple merge without bucket") @@ -160,7 +161,9 @@ TEST_CASE("Counter metrics", "[metrics][counter]") SECTION("Counter prometheus") { ++c; - c.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + c.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A counter test metric"); std::getline(output, line); @@ -219,7 +222,9 @@ TEST_CASE("Quantile metrics", "[metrics][quantile]") SECTION("Quantile prometheus") { q.update(12); - q.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + q.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A quantile test metric"); std::getline(output, line); @@ -289,7 +294,9 @@ TEST_CASE("Histogram int metrics", "[metrics][histogram]") h.update(1); h.update(8); h.update(12); - h.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + h.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A histogram test metric"); std::getline(output, line); @@ -356,7 +363,9 @@ TEST_CASE("Histogram double metrics", "[metrics][histogram]") h.update(1); h.update(8); h.update(12); - h.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + h.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A histogram test metric"); std::getline(output, line); @@ -423,7 +432,9 @@ TEST_CASE("TopN metrics", "[metrics][topn]") top_sting.update("top1"); top_sting.update("top2"); top_sting.update("top1"); - top_sting.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + top_sting.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A topn test metric"); std::getline(output, line); @@ -452,8 +463,10 @@ TEST_CASE("TopN metrics", "[metrics][topn]") top_int.update(123); top_int.update(10); top_int.update(123); - top_int.to_prometheus(output, {{"policy", "default"}}, + PrometheusSerializer ser; + top_int.to_prometheus(ser, {{"policy", "default"}}, [](const uint16_t &val) { return std::to_string(val); }); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A topn test metric"); std::getline(output, line); @@ -525,7 +538,9 @@ TEST_CASE("Cardinality metrics", "[metrics][cardinality]") SECTION("Cardinality prometheus") { c.update("metric"); - c.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + c.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); std::getline(output, line); CHECK(line == "# HELP root_test_metric A cardinality test metric"); std::getline(output, line); @@ -575,7 +590,9 @@ TEST_CASE("Rate metrics", "[metrics][rate]") SECTION("rate prometheus") { - r.to_prometheus(output, {{"policy", "default"}}); + PrometheusSerializer ser; + r.to_prometheus(ser, {{"policy", "default"}}); + output << ser.finalize(); } SECTION("rate opentelemetry") diff --git a/src/tests/test_prometheus_serializer.cpp b/src/tests/test_prometheus_serializer.cpp index f13bffa26..72cf39806 100644 --- a/src/tests/test_prometheus_serializer.cpp +++ b/src/tests/test_prometheus_serializer.cpp @@ -8,6 +8,7 @@ using Type = PrometheusSerializer::Type; TEST_CASE("single gauge family", "[prometheus][serializer]") { + Metric::reset_static_labels(); PrometheusSerializer s; s.write("net_packets_total", Type::Gauge, "Total packets", {}, {}, 42); CHECK(s.finalize() == @@ -18,6 +19,7 @@ TEST_CASE("single gauge family", "[prometheus][serializer]") TEST_CASE("repeated family writes one header, contiguous series", "[prometheus][serializer]") { + Metric::reset_static_labels(); PrometheusSerializer s; s.write("flow_bytes", Type::Gauge, "bytes", {}, {{"device", "A"}}, 10); s.write("other_metric", Type::Gauge, "x", {}, {}, 1); @@ -34,6 +36,7 @@ TEST_CASE("repeated family writes one header, contiguous series", "[prometheus][ TEST_CASE("histogram suffix + le label", "[prometheus][serializer]") { + Metric::reset_static_labels(); PrometheusSerializer s; s.write("dns_xact", Type::Histogram, "latency", {"bucket"}, {{"le", "10"}}, 3.0); s.write("dns_xact", Type::Histogram, "latency", {"count"}, {}, 7); @@ -59,6 +62,7 @@ TEST_CASE("static labels merged before passed labels", "[prometheus][serializer] TEST_CASE("label values are escaped", "[prometheus][serializer]") { + Metric::reset_static_labels(); PrometheusSerializer s; s.write("m", Type::Gauge, "d", {}, {{"desc", "a\"b\\c\nd"}}, 1); CHECK(s.finalize() == @@ -69,6 +73,7 @@ TEST_CASE("label values are escaped", "[prometheus][serializer]") TEST_CASE("numeric formatting matches ostream operator<<", "[prometheus][serializer]") { + Metric::reset_static_labels(); auto oss = [](auto v) { std::stringstream o; o << v; return o.str(); }; PrometheusSerializer s; s.write("i", Type::Gauge, "d", {}, {}, int64_t{1234567}); From 85681602406c825e5fdb4897ad9fe87dd76ee940 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 10:22:28 -0300 Subject: [PATCH 3/9] fix(metrics): share one Prometheus serializer across policies on the aggregate /metrics route (#716) --- src/CoreServer.cpp | 17 ++++--- src/Policies.cpp | 9 +++- src/Policies.h | 2 + src/tests/test_policies.cpp | 93 ++++++++++++++++++++++++++++++++++++- 4 files changed, 111 insertions(+), 10 deletions(-) diff --git a/src/CoreServer.cpp b/src/CoreServer.cpp index 4054224e4..157844594 100644 --- a/src/CoreServer.cpp +++ b/src/CoreServer.cpp @@ -434,16 +434,19 @@ void CoreServer::_setup_routes(const PrometheusConfig &prom_config) } } std::stringstream output; - for (const auto &p_mname : plist) { - try { + PrometheusSerializer ser; + try { + for (const auto &p_mname : plist) { auto [policy, lock] = _registry->policy_manager()->module_get_locked(p_mname); - policy->prometheus_metrics(output); - } catch (const std::exception &e) { - res.status = 500; - res.set_content(e.what(), "text/plain"); + policy->prometheus_metrics(ser); } - res.set_content(output.str(), "text/plain"); + } catch (const std::exception &e) { + res.status = 500; + res.set_content(e.what(), "text/plain"); + return; } + output << ser.finalize(); + res.set_content(output.str(), "text/plain"); }); if (_otel) { _otel->OnInterval([&](metrics::v1::ResourceMetrics &resource) { diff --git a/src/Policies.cpp b/src/Policies.cpp index a0b908e0b..b0b3e1b4b 100644 --- a/src/Policies.cpp +++ b/src/Policies.cpp @@ -380,9 +380,8 @@ void Policy::json_metrics(json &j, uint64_t period, bool merge) } } -void Policy::prometheus_metrics(std::stringstream &out) +void Policy::prometheus_metrics(PrometheusSerializer &ser) { - PrometheusSerializer ser; if (_merge_like_handlers) { auto bucket_map = _get_merged_buckets(); for (auto &[bucket, hmod] : bucket_map) { @@ -398,6 +397,12 @@ void Policy::prometheus_metrics(std::stringstream &out) } } } +} + +void Policy::prometheus_metrics(std::stringstream &out) +{ + PrometheusSerializer ser; + prometheus_metrics(ser); out << ser.finalize(); } diff --git a/src/Policies.h b/src/Policies.h index b20fc0cf3..e3aa3b531 100644 --- a/src/Policies.h +++ b/src/Policies.h @@ -19,6 +19,7 @@ namespace visor { class CoreRegistry; class AbstractMetricsBucket; +class PrometheusSerializer; class PolicyException : public std::runtime_error { @@ -108,6 +109,7 @@ class Policy : public AbstractRunnableModule void json_metrics(json &j, uint64_t period, bool merge); void prometheus_metrics(std::stringstream &out); + void prometheus_metrics(PrometheusSerializer &ser); void opentelemetry_metrics(metrics::v1::ScopeMetrics &scope); }; diff --git a/src/tests/test_policies.cpp b/src/tests/test_policies.cpp index abc9b121e..f99a5d2ae 100644 --- a/src/tests/test_policies.cpp +++ b/src/tests/test_policies.cpp @@ -5,6 +5,7 @@ #include "InputStream.h" #include "InputStreamManager.h" #include "Policies.h" +#include "PrometheusSerializer.h" #include #include @@ -1409,4 +1410,94 @@ TEST_CASE("Policies and Metrics", "[policies][metrics]") metrics::v1::ScopeMetrics scope; REQUIRE_NOTHROW(policy->opentelemetry_metrics(scope)); } -} \ No newline at end of file +} + +static auto policies_config_two_policies = R"( +version: "1.0" + +visor: + taps: + tap_multi: + input_type: mock + config: + iface: eth0 + policies: + policy_alpha: + kind: collection + input: + tap: tap_multi + input_type: mock + handlers: + window_config: + num_periods: 5 + deep_sample_rate: 100 + modules: + net: + type: net + policy_beta: + kind: collection + input: + tap: tap_multi + input_type: mock + handlers: + window_config: + num_periods: 5 + deep_sample_rate: 100 + modules: + net: + type: net +)"; + +// Count non-overlapping occurrences of needle in haystack. +static int count_occurrences(const std::string &haystack, const std::string &needle) +{ + int count = 0; + std::string::size_type pos = 0; + while ((pos = haystack.find(needle, pos)) != std::string::npos) { + ++count; + pos += needle.size(); + } + return count; +} + +TEST_CASE("Multi-policy Prometheus shared serializer", "[policies][prometheus][multi-policy]") +{ + // Regression test for #716: rendering two policies through a shared PrometheusSerializer + // must produce exactly one # HELP / # TYPE line per metric family, not one per policy. + CoreRegistry registry; + visor::load_builtin_plugins(registry); + registry.start(nullptr); + + YAML::Node config_file = YAML::Load(policies_config_two_policies); + REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"])); + REQUIRE_NOTHROW(registry.policy_manager()->load(config_file["visor"]["policies"])); + REQUIRE(registry.policy_manager()->module_exists("policy_alpha")); + REQUIRE(registry.policy_manager()->module_exists("policy_beta")); + + SECTION("Shared serializer produces unique # HELP and # TYPE headers") + { + PrometheusSerializer ser; + + { + auto [policy, lock] = registry.policy_manager()->module_get_locked("policy_alpha"); + REQUIRE_NOTHROW(policy->prometheus_metrics(ser)); + } + { + auto [policy, lock] = registry.policy_manager()->module_get_locked("policy_beta"); + REQUIRE_NOTHROW(policy->prometheus_metrics(ser)); + } + + std::string output = ser.finalize(); + REQUIRE(!output.empty()); + + // Every metric family must have its # HELP and # TYPE header emitted exactly once, + // even when two policies contribute series to the same family. + CHECK(count_occurrences(output, "# HELP packets_udp ") == 1); + CHECK(count_occurrences(output, "# TYPE packets_udp ") == 1); + CHECK(count_occurrences(output, "# HELP packets_tcp ") == 1); + CHECK(count_occurrences(output, "# TYPE packets_tcp ") == 1); + // Each policy must still contribute its own series (two distinct policy= label values). + CHECK(count_occurrences(output, "policy=\"policy_alpha\"") >= 1); + CHECK(count_occurrences(output, "policy=\"policy_beta\"") >= 1); + } +} From 262c2bc9a7733137cf97764bc942c276c24a7bc8 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:18:45 -0300 Subject: [PATCH 4/9] test(metrics): assert Prometheus series lines, not just family names; fill empty HELP MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses PR review (Copilot): - Migrated test assertions matched the bare metric/family name, which the new "# HELP"/"# TYPE" headers now contain — so they passed even if the sample line regressed. Assert the series-line prefix "{" instead (headers never contain it), restoring strictness. (test_handlers window markers; test_metrics test_performed.) The literal value is intentionally not pinned: process-global static labels set by sibling tests in the same binary make the exact label set non-deterministic, but "{" fully closes the headers-only-pass gap. - PrometheusSerializer::write now fills a family's HELP if it was first recorded empty and a later write supplies a non-empty one, and documents that the first write fixes the family Type (first-write-wins). Added a covering unit test. --- src/PrometheusSerializer.h | 6 ++++++ src/tests/test_handlers.cpp | 6 +++--- src/tests/test_metrics.cpp | 4 ++-- src/tests/test_prometheus_serializer.cpp | 15 +++++++++++++++ 4 files changed, 26 insertions(+), 5 deletions(-) diff --git a/src/PrometheusSerializer.h b/src/PrometheusSerializer.h index a5bb192ee..9246d6c08 100644 --- a/src/PrometheusSerializer.h +++ b/src/PrometheusSerializer.h @@ -23,10 +23,16 @@ class PrometheusSerializer void write(const std::string &base, Type type, const std::string &help, std::initializer_list suffix, const LabelMap &labels, V value) { + // The first write for a family fixes its Type and HELP; later writes only append series. + // Exception: if the family's HELP was first recorded empty, a later non-empty HELP fills it. + // Every caller uses one fixed Type+HELP per family (each metric has a fixed _desc), so a + // differing Type on a later write does not occur in practice; first-write Type wins. auto it = _families.find(base); if (it == _families.end()) { _order.push_back(base); it = _families.emplace(base, Family{type, help, {}}).first; + } else if (it->second.help.empty() && !help.empty()) { + it->second.help = help; } std::string line = base; for (const auto &s : suffix) { diff --git a/src/tests/test_handlers.cpp b/src/tests/test_handlers.cpp index 6a0eb33b6..4250df72f 100644 --- a/src/tests/test_handlers.cpp +++ b/src/tests/test_handlers.cpp @@ -175,7 +175,7 @@ TEST_CASE("StreamMetricsHandler tests", "[metrics][handler]") { PrometheusSerializer ser; handler->window_prometheus(ser); - CHECK(ser.finalize().find("live_window") != std::string::npos); + CHECK(ser.finalize().find("live_window{") != std::string::npos); } config.config_set("period", 3); @@ -183,13 +183,13 @@ TEST_CASE("StreamMetricsHandler tests", "[metrics][handler]") { PrometheusSerializer ser; handler->window_prometheus(ser); - CHECK(ser.finalize().find("first_window") != std::string::npos); + CHECK(ser.finalize().find("first_window{") != std::string::npos); } { PrometheusSerializer ser; handler->window_prometheus(ser, nullptr); - CHECK(ser.finalize().find("external_window") != std::string::npos); + CHECK(ser.finalize().find("external_window{") != std::string::npos); } } diff --git a/src/tests/test_metrics.cpp b/src/tests/test_metrics.cpp index 529d8329e..e0f9ec0d6 100644 --- a/src/tests/test_metrics.cpp +++ b/src/tests/test_metrics.cpp @@ -69,7 +69,7 @@ TEST_CASE("Abstract metrics manager", "[metrics][abstract]") { PrometheusSerializer ser; manager->window_single_prometheus(ser, 0, {{"policy", "default"}}); - CHECK(ser.finalize().find("test_performed") != std::string::npos); + CHECK(ser.finalize().find("test_performed{") != std::string::npos); } SECTION("Abstract window single opentelemetry") @@ -102,7 +102,7 @@ TEST_CASE("Abstract metrics manager", "[metrics][abstract]") auto live = static_cast(manager->live_bucket()); PrometheusSerializer ser; manager->window_external_prometheus(ser, live, {{"policy", "default"}}); - CHECK(ser.finalize().find("test_performed") != std::string::npos); + CHECK(ser.finalize().find("test_performed{") != std::string::npos); } SECTION("Abstract simple merge without bucket") diff --git a/src/tests/test_prometheus_serializer.cpp b/src/tests/test_prometheus_serializer.cpp index 72cf39806..15f18dc1d 100644 --- a/src/tests/test_prometheus_serializer.cpp +++ b/src/tests/test_prometheus_serializer.cpp @@ -96,3 +96,18 @@ TEST_CASE("numeric formatting matches ostream operator<<", "[prometheus][seriali CHECK(out.find(std::string(pair.first) + "{} " + pair.second + "\n") != std::string::npos); } } + +TEST_CASE("a later non-empty HELP fills an initially-empty one", "[prometheus][serializer]") +{ + Metric::reset_static_labels(); + PrometheusSerializer s; + s.write("m", Type::Gauge, "", {}, {}, 1); // first write: empty HELP + s.write("m", Type::Gauge, "real help", {}, {}, 2); // later write supplies HELP + auto out = s.finalize(); + // single family, one header pair, the non-empty HELP wins, both series present + CHECK(out == + "# HELP m real help\n" + "# TYPE m gauge\n" + "m{} 1\n" + "m{} 2\n"); +} From 26742aecd0feeaac7b69bec73f720cc6d389edf6 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:41:54 -0300 Subject: [PATCH 5/9] perf(metrics): buffer one string per family in PrometheusSerializer The serializer stored every series as a separate heap std::string in a std::vector per family, costing ~5 heap allocations per series. A microbenchmark on a realistic ~2400-series exposition showed this made the collect-then-emit path slower than the old reused-stringstream code it replaced, with ~6x more allocations. Append each series line directly into a single growing per-family body buffer instead; the rendered bytes are identical (the serializer's exact-output unit tests pass unchanged), allocations drop ~6x, and the path becomes ~3x faster. fmt stays on the integer/header path (cheap, allocation-free into the buffer); the float path is unchanged. --- src/PrometheusSerializer.h | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/src/PrometheusSerializer.h b/src/PrometheusSerializer.h index 9246d6c08..33146adfa 100644 --- a/src/PrometheusSerializer.h +++ b/src/PrometheusSerializer.h @@ -34,21 +34,25 @@ class PrometheusSerializer } else if (it->second.help.empty() && !help.empty()) { it->second.help = help; } - std::string line = base; + // Append the series line directly into the family's single body buffer (one growing + // allocation per family) instead of materializing a separate std::string per series — this + // avoids O(series) small heap allocations while keeping the rendered bytes identical. + std::string &body = it->second.body; + body.append(base); for (const auto &s : suffix) { - line.push_back('_'); - line.append(s); + body.push_back('_'); + body.append(s); } - append_labels(line, labels); - line.push_back(' '); + append_labels(body, labels); + body.push_back(' '); if constexpr (std::is_floating_point_v) { std::ostringstream oss; oss << value; - line.append(oss.str()); + body.append(oss.str()); } else { - fmt::format_to(std::back_inserter(line), "{}", value); + fmt::format_to(std::back_inserter(body), "{}", value); } - it->second.series.push_back(std::move(line)); + body.push_back('\n'); } // Renders all accumulated families into the Prometheus text exposition format. @@ -60,10 +64,7 @@ class PrometheusSerializer const auto &fam = _families.at(base); fmt::format_to(std::back_inserter(out), "# HELP {} {}\n", base, fam.help); fmt::format_to(std::back_inserter(out), "# TYPE {} {}\n", base, type_str(fam.type)); - for (const auto &series : fam.series) { - out.append(series); - out.push_back('\n'); - } + out.append(fam.body); } return out; } @@ -72,7 +73,7 @@ class PrometheusSerializer struct Family { Type type; std::string help; - std::vector series; + std::string body; // all series lines for this family, each terminated with '\n' }; std::vector _order; std::map _families; From d6064bf260c8f88ab622cb3b1c6f5a43c87b5aff Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:41:54 -0300 Subject: [PATCH 6/9] fix(metrics): fresh label map per TopN item to prevent conditional-label leakage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #787 review: TopN::to_prometheus (the formatter(LabelMap&,...) overload) built one LabelMap before the loop and reused it across items. The geo formatter sets lat/lon only when present, so an item without geo data inherited the previous item's lat/lon labels — emitting an incorrect label set. Construct the label map fresh per item so conditionally-set keys can't leak between series. --- src/Metrics.h | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Metrics.h b/src/Metrics.h index 4d0e30bc0..3fd1f3361 100644 --- a/src/Metrics.h +++ b/src/Metrics.h @@ -634,11 +634,13 @@ class TopN final : public Metric if (!std::min(_top_count, items.size())) { return; } - LabelMap l(add_labels); auto threshold = _get_threshold(items); const auto base = base_name_snake(); for (uint64_t i = 0; i < std::min(_top_count, items.size()); i++) { if (items[i].get_estimate() >= threshold) { + // Fresh label map per item: the formatter may set keys conditionally (e.g. geo + // lat/lon only when present), so a reused map would leak a prior item's keys. + LabelMap l(add_labels); formatter(l, _item_key, items[i].get_item()); ser.write(base, PrometheusSerializer::Type::Gauge, _desc, {}, l, items[i].get_estimate()); } else { From 6902aa72845a3811d7376e0569e05bd9a7d591d9 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 15:59:54 -0300 Subject: [PATCH 7/9] fix(metrics): default /metrics endpoint reuses Policy::prometheus_metrics for handler labels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #787 review: the default /metrics endpoint wrote every handler into one shared PrometheusSerializer but passed only {policy="default"}. Since every handler emits the shared base_* families (base_total, base_deep_samples, base_event_rate), grouping by family produced multiple samples with the same name AND identical labels — invalid Prometheus/OpenMetrics. Replace the hand-rolled module loop with policy->prometheus_metrics(ser), the same path the named/__all routes use, which tags each handler's series with a distinct {handler=...} label (and honors merge_like_handlers). Keeps the base families distinct and de-duplicates the loop. --- src/CoreServer.cpp | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/src/CoreServer.cpp b/src/CoreServer.cpp index 157844594..0a12fb883 100644 --- a/src/CoreServer.cpp +++ b/src/CoreServer.cpp @@ -176,14 +176,10 @@ void CoreServer::_setup_routes(const PrometheusConfig &prom_config) std::stringstream output; PrometheusSerializer ser; auto [policy, lock] = _registry->policy_manager()->module_get_locked("default"); - for (auto &mod : policy->modules()) { - auto hmod = dynamic_cast(mod); - if (hmod) { - spdlog::stopwatch sw; - hmod->window_prometheus(ser, {{"policy", "default"}}); - _logger->debug("{} window_prometheus elapsed time: {}", hmod->name(), sw); - } - } + // Reuse Policy::prometheus_metrics so each handler's shared base_* families carry a + // distinct {handler=...} label (and merge_like_handlers is honored consistently); + // a bare {policy="default"} would collapse them into duplicate same-label series. + policy->prometheus_metrics(ser); output << ser.finalize(); res.set_content(output.str(), "text/plain"); } catch (const std::exception &e) { From 6a7f1bbb362d24da95cb4b179efd72c81be7f9f0 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 16:11:56 -0300 Subject: [PATCH 8/9] fix(metrics): escape HELP text and de-dupe static/per-call labels in serializer PR #787 review (two spec-compliance fixes): - finalize() rendered "# HELP" with the raw description. HELP text must escape backslash and newline (but NOT quotes, unlike label values). Added a help-specific escaper so a _desc containing '\' or a newline can't produce invalid exposition. - append_labels() emitted both the static and the per-call value when a key existed in both maps, producing a duplicate label name in one sample (invalid). Static labels whose key is also provided per-call are now skipped (per-call wins); non-colliding keys keep the existing static-then-per-call order, so current output is unchanged. Added covering unit tests for both. --- src/PrometheusSerializer.h | 44 ++++++++++++++++++------ src/tests/test_prometheus_serializer.cpp | 21 +++++++++++ 2 files changed, 54 insertions(+), 11 deletions(-) diff --git a/src/PrometheusSerializer.h b/src/PrometheusSerializer.h index 33146adfa..3d53a254d 100644 --- a/src/PrometheusSerializer.h +++ b/src/PrometheusSerializer.h @@ -62,7 +62,10 @@ class PrometheusSerializer std::string out; for (const auto &base : _order) { const auto &fam = _families.at(base); - fmt::format_to(std::back_inserter(out), "# HELP {} {}\n", base, fam.help); + // HELP text escapes backslash and newline (but NOT quotes — unlike label values). + fmt::format_to(std::back_inserter(out), "# HELP {} ", base); + append_help_escaped(out, fam.help); + out.push_back('\n'); fmt::format_to(std::back_inserter(out), "# TYPE {} {}\n", base, type_str(fam.type)); out.append(fam.body); } @@ -89,6 +92,7 @@ class PrometheusSerializer } } + // Label-value escaping: backslash, double-quote, newline. static void append_escaped(std::string &out, const std::string &v) { for (char c : v) { @@ -101,21 +105,39 @@ class PrometheusSerializer } } + // HELP-text escaping: backslash and newline only (quotes are literal in HELP lines). + static void append_help_escaped(std::string &out, const std::string &v) + { + for (char c : v) { + switch (c) { + case '\\': out.append("\\\\"); break; + case '\n': out.append("\\n"); break; + default: out.push_back(c); + } + } + } + static void append_labels(std::string &line, const LabelMap &labels) { line.push_back('{'); bool any = false; - auto emit = [&](const LabelMap &m) { - for (const auto &[k, v] : m) { - line.append(k); - line.append("=\""); - append_escaped(line, v); - line.append("\","); - any = true; - } + auto append_one = [&](const std::string &k, const std::string &v) { + line.append(k); + line.append("=\""); + append_escaped(line, v); + line.append("\","); + any = true; }; - emit(prometheus_static_labels()); - emit(labels); + // Static labels first, but skip any key also given per-call so the per-call value wins — + // emitting the same label name twice in one sample is invalid exposition. + for (const auto &[k, v] : prometheus_static_labels()) { + if (labels.find(k) == labels.end()) { + append_one(k, v); + } + } + for (const auto &[k, v] : labels) { + append_one(k, v); + } if (any) { line.pop_back(); } diff --git a/src/tests/test_prometheus_serializer.cpp b/src/tests/test_prometheus_serializer.cpp index 15f18dc1d..fee40f724 100644 --- a/src/tests/test_prometheus_serializer.cpp +++ b/src/tests/test_prometheus_serializer.cpp @@ -111,3 +111,24 @@ TEST_CASE("a later non-empty HELP fills an initially-empty one", "[prometheus][s "m{} 1\n" "m{} 2\n"); } + +TEST_CASE("HELP text escapes backslash and newline but not quotes", "[prometheus][serializer]") +{ + Metric::reset_static_labels(); + PrometheusSerializer s; + s.write("m", Type::Gauge, "a\\b\nc\"d", {}, {}, 1); // help = a \ b c " d + auto out = s.finalize(); + CHECK(out.find("# HELP m a\\\\b\\nc\"d\n") != std::string::npos); +} + +TEST_CASE("per-call label overrides static label of the same key (no duplicate)", "[prometheus][serializer]") +{ + Metric::reset_static_labels(); + Metric::add_static_label("instance", "static_inst"); + PrometheusSerializer s; + s.write("m", Type::Gauge, "d", {}, {{"instance", "call_inst"}}, 1); + auto out = s.finalize(); + CHECK(out.find("m{instance=\"call_inst\"} 1\n") != std::string::npos); + CHECK(out.find("static_inst") == std::string::npos); // static value suppressed; key emitted once + Metric::reset_static_labels(); +} From 74a4c8253eb40488f4dbe4139b50b420a10659e5 Mon Sep 17 00:00:00 2001 From: Leo Parente <23251360+leoparente@users.noreply.github.com> Date: Wed, 24 Jun 2026 16:46:17 -0300 Subject: [PATCH 9/9] Revert "fix(metrics): default /metrics endpoint reuses Policy::prometheus_metrics for handler labels" MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 6902aa7. The review comment that prompted it assumed every handler emits the literal shared base_* families (base_total, base_deep_samples, base_event_rate), which would duplicate at the default endpoint. That premise is false: net/dhcp/dns/bgp rename their base metrics to schema-unique names (packets_*, dhcp_*, dns_*) via set_*_info, pcap emits no base metrics, and the only literal-base_* emitter (input_resources) is not in the default policy (net+dhcp+dns+pcap, cmd/pktvisord/main.cpp default_tap_policy). So the default /metrics endpoint never produced duplicate same-identity series, and 6902aa7 instead ADDED a {handler="..."} label to every series on the legacy "3.0.x compatible" default endpoint — a series-dimensionality change that would break existing default-endpoint scrapers/dashboards. Restore the original {policy="default"} loop (still feeding the grouping serializer, a no-op there since the default policy's families are uniquely named). The named-policy/__all routes (which always carried the handler label) and the #716 grouping fix are unaffected. --- src/CoreServer.cpp | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/CoreServer.cpp b/src/CoreServer.cpp index 0a12fb883..157844594 100644 --- a/src/CoreServer.cpp +++ b/src/CoreServer.cpp @@ -176,10 +176,14 @@ void CoreServer::_setup_routes(const PrometheusConfig &prom_config) std::stringstream output; PrometheusSerializer ser; auto [policy, lock] = _registry->policy_manager()->module_get_locked("default"); - // Reuse Policy::prometheus_metrics so each handler's shared base_* families carry a - // distinct {handler=...} label (and merge_like_handlers is honored consistently); - // a bare {policy="default"} would collapse them into duplicate same-label series. - policy->prometheus_metrics(ser); + for (auto &mod : policy->modules()) { + auto hmod = dynamic_cast(mod); + if (hmod) { + spdlog::stopwatch sw; + hmod->window_prometheus(ser, {{"policy", "default"}}); + _logger->debug("{} window_prometheus elapsed time: {}", hmod->name(), sw); + } + } output << ser.finalize(); res.set_content(output.str(), "text/plain"); } catch (const std::exception &e) {