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 <vadim.saranov@gmail.com>

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();
     });

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 <stdexcept>
+#include <utility>
+#include <vector>
 
 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<long long>(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<string, unsigned long long> series_entry;
+
+  /** labels go into the key: name{k1=v1,k2=v2} */
+  static string render_name(const string& base, const map<string, string>& 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<series_entry> collect_series(StatCountersGroupsInterface& group, const string& base)
+  {
+    vector<std::pair<map<string, string>, unsigned long long>> series;
+    group.iterateCounters([&series](unsigned long long value, const map<string, string>& labels) {
+      series.emplace_back(labels, value);
+    });
+
+    vector<series_entry> 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<long long>(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;
+  }
 }

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<bool>(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

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;