prometheus: first stab

Add libprometheus-cpp as a submodule.  Use it to publish information
about the number of connected peers in the DHT.

To experiment with it, run prometheus (e.g., `prometheus --config.file
prometheus/infinit-prometheus.yml`) and monitor
`infinit_overlay_peers` while, for instance running `tests/overlay` or
`tests/prometheus`.

.gitignore

@@ -3,3 +3,6 @@ TAGS
 __pycache__
 *node_modules*
 /_build/*/.drake
+
+# when running Prometheus from here.
+/data

.gitmodules

@@ -19,3 +19,6 @@
 [submodule "src/infinit/grpc/grpc"]
 	path = src/infinit/grpc/grpc
 	url = https://github.com/infinit/grpc.git
+[submodule "prometheus-cpp"]
+	path = prometheus/prometheus-cpp
+	url = https://github.com/infinit/prometheus-cpp.git

drakefile

@@ -287,6 +287,25 @@ def configure(
     extra_libs.append(protobuf_lib)
   else:
     infinit_sources += grpc.grpc.sources
+
+
+  ## ------------ ##
+  ## Prometheus.  ##
+  ## ------------ ##
+  prometheus = drake.include('prometheus/prometheus-cpp',
+                             cxx_toolkit = cxx_toolkit,
+                             in_cxx_config = cxx_config,
+                             protoc = grpc.grpc.protoc,
+                             protobuf_include = grpc.grpc.protobuf_include,
+                             protobuf_lib = grpc.grpc.protobuf_lib)
+  for p in prometheus.prometheus_include_path:
+    cxx_config_infinit.add_system_include_path(p)
+  extra_libs.append(drake.copy(prometheus.prometheus_lib,
+                               'bin' if windows else 'lib',
+                               strip_prefix = True))
+  drake.node('src/infinit/model/prometheus.hh')\
+       .dependencies_add(drake.nodes('prometheus/prometheus-cpp/lib/cpp/metrics.pb.h'))
+
   if windows:
     cxx_config_infinit.lib('shlwapi')
     cxx_config_infinit.lib('ws2_32')
@@ -337,6 +356,12 @@ def configure(
   cli_cxx_config = drake.cxx.Config(cxx_config)
   cli_cxx_config += elle.das.config
   cli_cxx_config.add_local_include_path('crash_reporting/src')
+
+  # Needed for Prometheus.
+  for p in prometheus.prometheus_include_path:
+    cli_cxx_config.add_system_include_path(p)
+  cli_cxx_config.add_system_include_path(grpc.grpc.protobuf_include)
+
   cli_cxx_config.lib_path_runtime('../lib')
   if not windows:
     cli_cxx_config.library_add(
@@ -1015,6 +1040,7 @@ def configure(
     ('grpc', []),
     ('kelips', []),
     ('overlay', []),
+    ('prometheus', []),
     ('rpc', []),
     ('storage', [aws_lib]),
   ]

prometheus/infinit-prometheus.yml

@@ -0,0 +1,10 @@
+# my global config
+global:
+  scrape_interval:     15s # Default is every 1 minute.
+  evaluation_interval: 15s # Default is every 1 minute.
+  # scrape_timeout is set to the global default (10s).
+
+scrape_configs:
+  - job_name: 'infinit'
+    static_configs:
+      - targets: ['localhost:8080']

src/infinit/model/doughnut/Doughnut.cc

@@ -38,13 +38,49 @@
 #include <infinit/model/doughnut/Cache.hh>
 #include <infinit/model/doughnut/consensus/Paxos.hh>
 #include <infinit/model/doughnut/conflict/UBUpserter.hh>
+#include <infinit/model/prometheus.hh>
 #include <infinit/model/MonitoringServer.hh>
 #include <infinit/storage/MissingKey.hh>
 
 ELLE_LOG_COMPONENT("infinit.model.doughnut.Doughnut");
 
 namespace
 {
+  /// Create the family of counters used to count the number of
+  /// connected members for this DHT.
+  ::prometheus::Family<prometheus::Gauge>*
+  member_gauge_family()
+  {
+    static auto* res = [] () -> ::prometheus::Family<prometheus::Gauge>*
+    {
+      // Add a new member gauge family to the registry.
+      if (auto reg = infinit::prometheus::registry())
+      {
+        static auto& res = ::prometheus::BuildGauge()
+          .Name("infinit_overlay_peers")
+          .Help("How many overlay peers this overlay member is connected to")
+          .Register(*infinit::prometheus::registry());
+        return &res;
+      }
+      else
+        return {};
+    }();
+    return res;
+  }
+
+  /// Build a new Prometheus gauge.
+  ///
+  /// May return nullptr if set up failed.
+  ::prometheus::Gauge*
+  make_member_gauge(infinit::model::Address const& id)
+  {
+    if (auto* f = member_gauge_family())
+      return &f->Add({{"id", elle::sprintf("%f", id)}});
+    else
+      return nullptr;
+  }
+
+
   std::chrono::milliseconds
   _connect_timeout_val(elle::Defaulted<std::chrono::milliseconds> arg)
   {
@@ -155,6 +191,7 @@ namespace infinit
         , _soft_fail_running(
           _soft_fail_running_val(std::move(init.soft_fail_running)))
         , _id(std::move(init.id))
+        , _member_gauge(make_member_gauge(this->id()))
         , _keys(std::move(init.keys))
         , _owner(std::move(init.owner))
         , _passport(std::move(init.passport))
@@ -283,6 +320,8 @@ namespace infinit
           }
           this->_local.reset();
         }
+        if (this->_member_gauge)
+          member_gauge_family()->Remove(this->_member_gauge);
       }
 
       void

src/infinit/model/doughnut/Doughnut.hh

@@ -23,6 +23,11 @@
 #include <infinit/model/doughnut/Passport.hh>
 #include <infinit/overlay/Overlay.hh>
 
+namespace prometheus
+{
+  class Gauge;
+}
+
 namespace infinit
 {
   namespace model
@@ -160,6 +165,7 @@ namespace infinit
         int
         ensure_key(std::shared_ptr<elle::cryptography::rsa::PublicKey> const& k);
         ELLE_ATTRIBUTE_R(Address, id);
+        prometheus::Gauge* _member_gauge;
         ELLE_ATTRIBUTE(std::shared_ptr<elle::cryptography::rsa::KeyPair>, keys);
         ELLE_ATTRIBUTE_R(std::shared_ptr<elle::cryptography::rsa::PublicKey>, owner);
         ELLE_ATTRIBUTE_R(Passport, passport);

src/infinit/model/drakefile

@@ -76,6 +76,8 @@ sources = drake.nodes(
   'faith/Faith.hh',
   'paranoid/Paranoid.cc',
   'paranoid/Paranoid.hh',
+  'prometheus.cc',
+  'prometheus.hh',
 )
 
 # Local Variables:

src/infinit/model/prometheus.cc

@@ -0,0 +1,86 @@
+#include <infinit/model/prometheus.hh>
+
+#include <boost/exception/diagnostic_information.hpp>
+
+#include <elle/log.hh>
+#include <elle/os/environ.hh>
+
+#include <prometheus/exposer.h>
+#include <prometheus/registry.h>
+
+ELLE_LOG_COMPONENT("infinit.prometheus");
+
+namespace
+{
+  auto prometheus_endpoint
+    = elle::os::getenv("INFINIT_PROMETHEUS_ENDPOINT", "127.0.0.1:8080");
+}
+
+namespace infinit
+{
+  namespace prometheus
+  {
+    void prometheus_endpoint(std::string e)
+    {
+      ::prometheus_endpoint = std::move(e);
+    }
+
+    std::string const& prometheus_endpoint()
+    {
+      return ::prometheus_endpoint;
+    }
+
+    /// An HTTP server to answer Prometheus's requests.
+    ///
+    /// Maybe nullptr if set up failed.
+    ::prometheus::Exposer*
+    exposer()
+    {
+      static auto res = []() -> std::unique_ptr<::prometheus::Exposer> {
+        auto const addr = prometheus_endpoint();
+        if (addr != "no" || addr != "0")
+          try
+          {
+            ELLE_LOG("exposer: create: %s", addr);
+            auto res = std::make_unique<::prometheus::Exposer>(addr);
+            ELLE_LOG("exposer: creation succeeded");
+            return res;
+          }
+          catch (std::exception const& e)
+          {
+            ELLE_LOG("exposer: creation failed,"
+                     " metrics will not be exposed: %s", e);
+          }
+          catch (...)
+          {
+            ELLE_LOG("exposer: creation failed with unknown"
+                     " exception type: %s",
+                     boost::current_exception_diagnostic_information());
+          }
+        return {};
+      }();
+      return res.get();
+    }
+
+
+    /// Where to register the measurements to expose to Prometheus.
+    std::shared_ptr<::prometheus::Registry>
+    registry()
+    {
+      // Build and ask the exposer to scrape the registry on incoming
+      // scrapes.
+      static auto res = []() -> std::shared_ptr<::prometheus::Registry>
+      {
+        if (auto e = exposer())
+        {
+          auto res = std::make_shared<::prometheus::Registry>();
+          e->RegisterCollectable(res);
+          return res;
+        }
+        else
+          return nullptr;
+      }();
+      return res;
+    }
+  }
+}

src/infinit/model/prometheus.hh

@@ -0,0 +1,35 @@
+#pragma once
+
+#include <memory>
+#include <string>
+
+#include <prometheus/registry.h>
+
+namespace prometheus
+{
+  class Exposer;
+}
+
+namespace infinit
+{
+  namespace prometheus
+  {
+    /// Set the Prometheus publishing address.
+    void prometheus_endpoint(std::string e);
+
+    /// Get the Prometheus publishing address.
+    std::string const& prometheus_endpoint();
+
+    /// An HTTP server to answer Prometheus's requests.
+    ///
+    /// Maybe nullptr if set up failed.
+    ::prometheus::Exposer*
+    exposer();
+
+    /// Where to register the measurements to expose to Prometheus.
+    ///
+    /// Maybe nullptr if set up failed.
+    std::shared_ptr<::prometheus::Registry>
+    registry();
+  }
+}

src/infinit/overlay/Overlay.cc

@@ -8,6 +8,7 @@
 #include <infinit/overlay/Overlay.hh>
 #include <infinit/model/MissingBlock.hh>
 #include <infinit/model/doughnut/DummyPeer.hh>
+#include <infinit/model/prometheus.hh>
 
 ELLE_LOG_COMPONENT("infinit.overlay.Overlay");
 
@@ -23,7 +24,23 @@ namespace infinit
                      std::shared_ptr<infinit::model::doughnut::Local> local)
       : _doughnut(dht)
       , _local(local)
-    {}
+    {
+      if (auto* g = _doughnut->_member_gauge)
+      {
+        ELLE_LOG_COMPONENT("infinit.overlay.Overlay.prometheus");
+        ELLE_TRACE("%s: construct, gauge: %s", this, g->Value());
+        this->on_discovery().connect([this, g](NodeLocation, bool){
+            ELLE_DEBUG("%s: signaling one more than %s",
+                     this, g->Value());
+            g->Increment();
+          });
+        this->on_disappearance().connect([this, g](model::Address, bool){
+            ELLE_DEBUG("%s: signaling one less than %s",
+                     this, g->Value());
+            g->Decrement();
+          });
+      }
+    }
 
     Overlay::~Overlay()
     {

src/infinit/overlay/Overlay.hh

@@ -15,6 +15,11 @@
 #include <infinit/model/doughnut/protocol.hh>
 #include <infinit/serialization.hh>
 
+namespace prometheus
+{
+  class Gauge;
+}
+
 namespace infinit
 {
   namespace overlay
@@ -100,11 +105,11 @@ namespace infinit
     | Hooks |
     `------*/
     public:
-      /// Announcing discovered nodes.
+      /// Announcing discovered/connected nodes.
       using DiscoveryEvent =
         boost::signals2::signal<auto (NodeLocation id, bool observer) -> void>;
       ELLE_ATTRIBUTE_RX(DiscoveryEvent, on_discovery);
-      /// Announcing disconnected nodes.
+      /// Announcing disappeared/disconnected nodes.
       using DisappearanceEvent =
         boost::signals2::signal<auto (model::Address id, bool observer) -> void>;
       ELLE_ATTRIBUTE_RX(DisappearanceEvent, on_disappearance);

tests/overlay.cc

@@ -1596,7 +1596,6 @@ ELLE_TEST_SCHEDULED(churn, (TestConfiguration, config),
   }
 }
 
-
 void
 test_churn_socket(TestConfiguration config, bool pasv)
 {