Add connection scraper and tracker metrics (#412)

collector/lib/CollectorService.cpp

@@ -31,7 +31,6 @@ extern "C" {
 
 #include "civetweb/CivetServer.h"
 #include "prometheus/exposer.h"
-#include "prometheus/registry.h"
 
 #include "ConnTracker.h"
 #include "Containers.h"
@@ -92,14 +91,15 @@ void CollectorService::RunForever() {
       net_status_notifier = MakeUnique<NetworkStatusNotifier>(config_.Hostname(), config_.HostProc(),
                                                               config_.ScrapeInterval(), config_.ScrapeListenEndpoints(),
                                                               config_.TurnOffScrape(),
-                                                              conn_tracker, config_.grpc_channel);
+                                                              conn_tracker, config_.grpc_channel,
+                                                              &collector_stats_);
       net_status_notifier->Start();
     }
   }
 
   sysdig.Init(config_, conn_tracker);
 
-  CollectorStatsExporter exporter(registry, &config_, &sysdig);
+  CollectorStatsExporter exporter(registry, &config_, &sysdig, &collector_stats_);
   if (!exporter.start()) {
     CLOG(FATAL) << "Unable to start collector stats exporter";
   }

collector/lib/CollectorService.h

@@ -27,6 +27,7 @@ You should have received a copy of the GNU General Public License along with thi
 #include <vector>
 
 #include "CollectorConfig.h"
+#include "CollectorStats.h"
 
 namespace collector {
 
@@ -54,6 +55,7 @@ class CollectorService {
 
   std::atomic<ControlValue>* control_;
   const std::atomic<int>& signum_;
+  CollectorStats collector_stats_;
 };
 
 }  // namespace collector

collector/lib/CollectorStats.cpp

@@ -0,0 +1,17 @@
+#include "CollectorStats.h"
+
+namespace collector {
+
+#define X(n) #n,
+std::array<std::string, CollectorStats::timer_type_max> CollectorStats::timer_type_to_name = {
+  TIMER_NAMES
+};
+#undef X
+
+#define X(n) #n,
+std::array<std::string, CollectorStats::counter_type_max> CollectorStats::counter_type_to_name = {
+  COUNTER_NAMES
+};
+#undef X
+
+} // namespace collector

collector/lib/CollectorStats.h

@@ -0,0 +1,107 @@
+#ifndef COLLECTOR_COLLECTORSTATS_H
+#define COLLECTOR_COLLECTORSTATS_H
+
+#include <array>
+#include <atomic>
+#include <unordered_map>
+#include <vector>
+
+#include "TimeUtil.h"
+
+#define TIMER_NAMES   \
+X(net_scrape_read)    \
+X(net_scrape_update)  \
+X(net_fetch_state)    \
+X(net_create_message) \
+X(net_write_message)
+
+#define COUNTER_NAMES \
+X(net_conn_updates) \
+X(net_conn_deltas) \
+X(net_conn_inactive) \
+X(net_cep_updates) \
+X(net_cep_deltas) \
+X(net_cep_inactive) \
+X(net_known_ip_networks) \
+X(net_known_public_ips)
+
+namespace collector {
+
+class CollectorStats {
+public:
+# define X(n) n,
+  enum TimerType {
+    TIMER_NAMES
+    X(timer_type_max)
+  };
+# undef X
+  static std::array<std::string, timer_type_max> timer_type_to_name;
+
+  inline int64_t GetTimerCount(size_t index) const { return timer_count_[index]; }
+  inline int64_t GetTimerDurationMicros(size_t index) const { return timer_total_us_[index]; }
+  inline void EndTimerAt(size_t index, int64_t duration_us) {
+    ++timer_count_[index];
+    timer_total_us_[index] += duration_us;
+  }
+
+# define X(n) n,
+  enum CounterType {
+    COUNTER_NAMES
+    X(counter_type_max)
+  };
+# undef X
+  static std::array<std::string, counter_type_max> counter_type_to_name;
+
+  inline int64_t GetCounter(size_t index) const { return counter_[index]; }
+  inline void CounterSet(size_t index, int64_t val) {
+    counter_[index] = val;
+  }
+  inline void CounterAdd(size_t index, int64_t val) {
+    counter_[index] += val;
+  }
+
+private:
+  std::array<std::atomic<int64_t>, timer_type_max> timer_count_ = {{}};
+  std::array<std::atomic<int64_t>, timer_type_max> timer_total_us_ = {{}};
+
+  std::array<std::atomic<int64_t>, counter_type_max> counter_ = {{}};
+};
+
+namespace internal {
+
+template<typename T>
+class ScopedTimer {
+public:
+  ScopedTimer(T* timer_array, size_t index)
+    : timer_array_(timer_array), index_(index), start_time_(NowMicros()) {}
+  ~ScopedTimer() {
+    if (timer_array_) {
+      timer_array_->EndTimerAt(index_, NowMicros() - start_time_);
+    }
+  }
+  constexpr operator bool() const { return true; }
+
+private:
+  T* timer_array_;
+  size_t index_;
+  int64_t start_time_;
+};
+
+template<typename T>
+ScopedTimer<T> scoped_timer(T* timer_array, size_t index) {
+  return ScopedTimer<T>(timer_array, index);
+}
+
+} // namespace internal
+
+#define SCOPED_TIMER(s,i) auto __scoped_timer_ ## __LINE__ = internal::scoped_timer(s,i)
+#define WITH_TIMER(s,i) if (SCOPED_TIMER(s,i))
+
+#define COUNTER_SET(s,i,v) do { if (s != nullptr) { s->CounterSet(i, static_cast<int64_t>(v)); } } while(0)
+#define COUNTER_ADD(s,i,v) do { if (s != nullptr) { s->CounterAdd(i, static_cast<int64_t>(v)); } } while(0)
+#define COUNTER_INC(s,i) COUNTER_ADD(s,i,1)
+#define COUNTER_ZERO(s,i) COUNTER_SET(s,i,0)
+
+} // namespace collector
+
+#endif //COLLECTOR_COLLECTORSTATS_H

collector/lib/CollectorStatsExporter.cpp

@@ -23,26 +23,20 @@ You should have received a copy of the GNU General Public License along with thi
 
 #include <iostream>
 #include <chrono>
-#include <string>
 
 #include "Containers.h"
 #include "EventNames.h"
 #include "SysdigService.h"
 #include "CollectorStatsExporter.h"
 #include "Logging.h"
+#include "Utility.h"
 
-#include "prometheus/registry.h"
 #include "prometheus/gauge.h"
 
-extern "C" {
-    #include <pthread.h>
-    #include <string.h>
-}
-
 namespace collector {
 
-CollectorStatsExporter::CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, SysdigService* sysdig)
-    : registry_(std::move(registry)), config_(config), sysdig_(sysdig)
+CollectorStatsExporter::CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, SysdigService* sysdig, CollectorStats* collector_stats)
+    : registry_(std::move(registry)), config_(config), sysdig_(sysdig), collector_stats_(collector_stats)
 {}
 
 bool CollectorStatsExporter::start() {
@@ -53,6 +47,26 @@ bool CollectorStatsExporter::start() {
     return true;
 }
 
+class CollectorTimerGauge {
+public:
+  CollectorTimerGauge(prometheus::Family<prometheus::Gauge>& g, const std::string& timer_name)
+    : events_(&g.Add({{"type", timer_name + "_events"}})),
+      times_us_total_(&g.Add({{"type", timer_name + "_times_us_total"}})),
+      times_us_avg_(&g.Add({{"type", timer_name + "_times_us_avg"}}))
+  {}
+
+  void Update(int64_t count, int64_t total_us) {
+    events_->Set(count);
+    times_us_total_->Set(total_us);
+    times_us_avg_->Set(count ? total_us / count : 0);
+  }
+
+private:
+  prometheus::Gauge* events_;
+  prometheus::Gauge* times_us_total_;
+  prometheus::Gauge* times_us_avg_;
+};
+
 void CollectorStatsExporter::run() {
     auto& collectorEventCounters = prometheus::BuildGauge()
         .Name("rox_collector_events")
@@ -74,6 +88,26 @@ void CollectorStatsExporter::run() {
     auto& processResolutionFailuresByTinfo = collectorEventCounters.Add({{"type", "processResolutionFailuresByTinfo"}});
     auto& processRateLimitCount = collectorEventCounters.Add({{"type", "processRateLimitCount"}});
 
+    auto& collector_timers_gauge = prometheus::BuildGauge()
+            .Name("rox_collector_timers")
+            .Help("Collector timers")
+            .Register(*registry_);
+    std::array<unique_ptr<CollectorTimerGauge>, CollectorStats::timer_type_max> collector_timers;
+    for (int i=0; i<CollectorStats::timer_type_max; i++) {
+      auto tt = (CollectorStats::TimerType)(i);
+      collector_timers[tt] = MakeUnique<CollectorTimerGauge>(collector_timers_gauge,
+                                                             CollectorStats::timer_type_to_name[tt]);
+    }
+    auto& collector_counters_gauge = prometheus::BuildGauge()
+            .Name("rox_collector_counters")
+            .Help("Collector counters")
+            .Register(*registry_);
+    std::array<prometheus::Gauge*, CollectorStats::counter_type_max> collector_counters;
+    for (int i=0; i<CollectorStats::counter_type_max; i++) {
+      auto ct = (CollectorStats::CounterType)(i);
+      collector_counters[i] = &(collector_counters_gauge.Add({{"type", CollectorStats::counter_type_to_name[ct]}}));
+    }
+
     auto& collectorTypedEventCounters = prometheus::BuildGauge()
             .Name("rox_collector_events_typed")
             .Help("Collector events by event type")
@@ -186,6 +220,16 @@ void CollectorStatsExporter::run() {
         processResolutionFailuresByEvt.Set(stats.nProcessResolutionFailuresByEvt);
         processResolutionFailuresByTinfo.Set(stats.nProcessResolutionFailuresByTinfo);
         processRateLimitCount.Set(stats.nProcessRateLimitCount);
+
+        for (int i = 0; i < CollectorStats::timer_type_max; i++) {
+          auto tt = (CollectorStats::TimerType)(i);
+          collector_timers[tt]->Update(collector_stats_->GetTimerCount(tt),
+                                       collector_stats_->GetTimerDurationMicros(tt));
+        }
+        for (int i = 0; i < CollectorStats::counter_type_max; i++) {
+          auto ct = (CollectorStats::CounterType)(i);
+          collector_counters[ct]->Set(collector_stats_->GetCounter(ct));
+        }
     }
 }
 

collector/lib/CollectorStatsExporter.h

@@ -27,6 +27,7 @@ You should have received a copy of the GNU General Public License along with thi
 #include <memory>
 
 #include "CollectorConfig.h"
+#include "CollectorStats.h"
 #include "StoppableThread.h"
 #include "SysdigService.h"
 
@@ -36,7 +37,7 @@ namespace collector {
 
 class CollectorStatsExporter {
   public:
-    CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, SysdigService* sysdig);
+    CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, SysdigService* sysdig, CollectorStats* collector_stats);
 
     bool start();
     void run();
@@ -47,6 +48,7 @@ class CollectorStatsExporter {
     const CollectorConfig* config_;
     SysdigService *sysdig_;
     StoppableThread thread_;
+    CollectorStats* collector_stats_;
 };
 
 }  // namespace collector