From a67b3cb3a6294ae3e2a2a529ce922261cd94dc85 Mon Sep 17 00:00:00 2001 From: Vadim Saranov Date: Tue, 13 Oct 2026 18:00:26 +0300 Subject: [PATCH] core: statistics DI interface Expose the registry over the DI channel as the "statistics" interface, so the metrics can be read without a Prometheus pipeline. dump returns every current series as one flat map, get returns one group or series, meta returns type and help text, and reset calls the reset hook of the groups whose owner registered one. The interface is registered between AmPlugIn::init() and load(), so loaded modules can register their own metrics and find the interface already exported over xmlrpc2di, which enumerates DI interfaces at module load time. Signed-off-by: Vadim Saranov --- core/AmSessionContainer.cpp | 2 + core/AmStatistics.cpp | 145 +++++++++++++++++++++++++++++++++++- core/AmStatistics.h | 29 ++++++++ core/sems.cpp | 5 ++ 4 files changed, 179 insertions(+), 2 deletions(-) diff --git a/core/AmSessionContainer.cpp b/core/AmSessionContainer.cpp index 1d956429..fa18b0e6 100644 --- a/core/AmSessionContainer.cpp +++ b/core/AmSessionContainer.cpp @@ -62,6 +62,7 @@ AmSessionContainer::AmSessionContainer() stat_group(Gauge, "core", "max_session_num") .setHelp("maximum number of running sessions") + .setReset([]() { AmSession::updateMaxSessionNum(); }) .addFunctionCounter([]() -> unsigned long long { return AmSession::getMaxSessionNum(); }); stat_group(Gauge, "core", "dead_sessions_count") @@ -78,6 +79,7 @@ AmSessionContainer::AmSessionContainer() stat_group(Gauge, "core", "cps_max") .setHelp("maximum calls per second since the last reset") + .setReset([]() { AmSessionContainer::instance()->resetMaxCPS(); }) .addFunctionCounter([]() -> unsigned long long { return AmSessionContainer::instance()->peekMaxCPS(); }); diff --git a/core/AmStatistics.cpp b/core/AmStatistics.cpp index 9783e2c1..ad697d14 100644 --- a/core/AmStatistics.cpp +++ b/core/AmStatistics.cpp @@ -29,9 +29,13 @@ /** @file AmStatistics.cpp - metrics registry */ #include "AmStatistics.h" +#include "AmApi.h" +#include "AmPlugIn.h" #include "log.h" #include +#include +#include StatCounterInterface::~StatCounterInterface() {} @@ -277,7 +281,7 @@ StatCountersSingleGroup& AmStatistics::group(StatCountersSingleGroup::Type type, if (!it.second && existent_group->getType() != type) { ERROR("attempt to add counter '%s' with type '%s' to existing counters group with another type '%s'", - name.data(), StatCountersGroupsInterface::type2str(type), + name.c_str(), StatCountersGroupsInterface::type2str(type), StatCountersGroupsInterface::type2str(existent_group->getType())); throw std::logic_error("attempt to redefine existent StatCountersSingleGroup"); } @@ -291,7 +295,144 @@ void AmStatistics::addGroupsContainer(const string& name, StatsCountersGroupsCon // not owned: the registering code keeps the container alive, see the header auto it = counters_groups_containers.try_emplace(name, container, false); if (!it.second) { - ERROR("attempt to add groups container %p by existing name: %s", container, name.data()); + ERROR("attempt to add groups container %p by existing name: %s", container, name.c_str()); throw std::logic_error("attempt to redefine existent StatCountersGroup"); } +} + +/** read the metrics registry over the DI command channels: the UDP + * stats socket ("DI statistics ..."), jsonrpc ("statistics....") + * and xmlrpc2di ("di", "statistics", ...). + * + * Replies are flat {metric: number} maps: one entry per series, its + * labels embedded in the key (core_codec_usage{codec=pcmu,...}). + * Type and description of a metric are read with meta. */ +class AmStatisticsDI : public AmDynInvoke, public AmDynInvokeFactory { +public: + AmStatisticsDI() : AmDynInvoke(), AmDynInvokeFactory("statistics") {} + ~AmStatisticsDI() override {} + + int onLoad() override { return 0; } + AmDynInvoke* getInstance() override { return this; } + + void invoke(const string& method, const AmArg& args, AmArg& ret) override + { + if (method == "_list") { + ret.push(AmArg("dump")); + ret.push(AmArg("get")); + ret.push(AmArg("meta")); + ret.push(AmArg("reset")); + } else if (method == "dump") { + AmStatistics::instance()->iterateGroups([&ret](const string& name, + StatCountersGroupsInterface& group) { + for (auto& s : collect_series(group, name)) + ret[s.first.c_str()] = AmArg(static_cast(s.second)); + }); + } else if (method == "get") { + args.assertArrayFmt("s"); + get_metric(string(args[0].asCStr()), ret); + } else if (method == "meta") { + args.assertArrayFmt("s"); + meta_metric(string(args[0].asCStr()), ret); + } else if (method == "reset") { + args.assertArrayFmt("s"); + reset_metric(string(args[0].asCStr())); + } else { + throw AmDynInvoke::NotImplemented(method); + } + } + +private: + static StatCountersGroupsInterface* find_group(const string& name) + { + StatCountersGroupsInterface* found = nullptr; + AmStatistics::instance()->iterateGroups([&name, &found](const string& group_name, + StatCountersGroupsInterface& group) { + if (group_name == name) found = &group; + }); + return found; + } + + /** rendered series name and value */ + typedef std::pair series_entry; + + /** labels go into the key: name{k1=v1,k2=v2} */ + static string render_name(const string& base, const map& labels) + { + if (labels.empty()) + return base; + + string s = base + "{"; + bool first = true; + for (auto& l : labels) { + if (!first) + s += ","; + s += l.first + "=" + l.second; + first = false; + } + return s + "}"; + } + + /** the series have to be collected before rendering: iterate callbacks + * run under the group lock */ + static vector collect_series(StatCountersGroupsInterface& group, const string& base) + { + vector, unsigned long long>> series; + group.iterateCounters([&series](unsigned long long value, const map& labels) { + series.emplace_back(labels, value); + }); + + vector entries; + for (auto& s : series) + entries.emplace_back(render_name(base, s.first), s.second); + return entries; + } + + /** a base name selects all series of the metric, a rendered series name + * selects one; a metric with no series right now renders an empty map */ + static void get_metric(const string& name, AmArg& ret) + { + string base = name.substr(0, name.find('{')); + StatCountersGroupsInterface* group = find_group(base); + if (!group) throw string("no such counters group: '") + base + "'"; + + ret.assertStruct(); + size_t matched = 0; + for (auto& s : collect_series(*group, base)) { + if (name == base || name == s.first) { + ret[s.first.c_str()] = AmArg(static_cast(s.second)); + ++matched; + } + } + if (!matched && name != base) throw string("no such series: '") + name + "'"; + } + + /** {metric: {type, help}} */ + static void meta_metric(const string& name, AmArg& ret) + { + StatCountersGroupsInterface* group = find_group(name); + if (!group) throw string("no such counters group: '") + name + "'"; + + AmArg& info = ret[name.c_str()]; + info["type"] = AmArg(StatCountersGroupsInterface::type2str(group->getType())); + info["help"] = AmArg(group->getHelp()); + } + + /** reset through the callback registered by the group owner, if any */ + static void reset_metric(const string& name) + { + StatCountersGroupsInterface* group = find_group(name); + if (!group) throw string("no such counters group: '") + name + "'"; + if (!group->hasReset()) throw string("counters group '") + name + "' is not resettable"; + group->reset(); + } +}; + +void register_statistics() +{ + AmStatisticsDI* di = new AmStatisticsDI(); + if (!AmPlugIn::instance()->registerDIInterface("statistics", di)) { + ERROR("registering 'statistics' DI interface\n"); + delete di; + } } \ No newline at end of file diff --git a/core/AmStatistics.h b/core/AmStatistics.h index f4565d90..0ded8fc7 100644 --- a/core/AmStatistics.h +++ b/core/AmStatistics.h @@ -205,11 +205,29 @@ public: } const string& getHelp() { return help_; } + /** reset callback for read-to-reset metrics (maximums), invoked by the + * statistics DI interface; counter callbacks must stay side-effect free. + * A function pointer, not a std::function: the registry is disposed after + * the plug-ins are unloaded, so a std::function built in a plug-in would + * be destroyed through code that is no longer mapped. */ + using reset_function_type = void (*)(); + + StatCountersGroupsInterface& setReset(reset_function_type func) + { + reset_ = func; + return *this; + } + bool hasReset() const { return static_cast(reset_); } + void reset() { if (reset_) reset_(); } + Type getType() { return type_; } private: Type type_; string help_; + /** initialized explicitly: a group that registers no reset callback must + * read as not resettable (a plain pointer has no default constructor) */ + reset_function_type reset_ = nullptr; }; /** provider of metric groups, registered into the registry by name */ @@ -262,6 +280,12 @@ public: return *this; } + StatCountersSingleGroup& setReset(reset_function_type func) + { + StatCountersGroupsInterface::setReset(func); + return *this; + } + private: /** under counters_lock: check the labels against the name set fixed by * the first series (or fix it) */ @@ -327,4 +351,9 @@ private: #define stat_group(type, grouping_name, name) \ AmStatistics::instance()->group(StatCountersSingleGroup::type, grouping_name, name) +// Registers the "statistics" DI interface. Must be called after +// AmPlugIn::init() and before the modules are loaded, so that modules can +// register their metrics and find the interface already exported. +void register_statistics(); + #endif \ No newline at end of file diff --git a/core/sems.cpp b/core/sems.cpp index 54eaf257..5b63b501 100644 --- a/core/sems.cpp +++ b/core/sems.cpp @@ -685,6 +685,11 @@ int main(int argc, char* argv[]) INFO("Loading plug-ins\n"); AmPlugIn::instance()->init(); + + // before the modules are loaded, so that they can use the metrics + // registry and the statistics interface + register_statistics(); + if(AmPlugIn::instance()->load(AmConfig::PlugInPath, AmConfig::LoadPlugins)) goto error;