Refactor NetworkStatusNotifier

This change started out from the effort to unify the services used to
communicate with sensor. While attempting to add an output object to the
notifier, it became apparent this class has some issues with ownership
of its members, which in turn makes it a bit odd to extend and modify.

In order to alleviate the problem the following changes are introduced:
- The connection scraper is now owned by NetworkStatusNotifier.
- The communication channel for network info is now owned by
  NetworkStatusNotifier.
- Connection stats are now owned by NetworkStatusNotifier.
- Move ownership of the prometheus registry into CollectorStatsExporter.
- Refactor CollectorConnectionStats into its own file.

When stating "X is now owned by Y" it stands for X now is part of Y and
its lifecycle is fully handled by Y (i.e X is created with Y and
destroyed with Y, preferably following RAII).

Author: Molter73

collector/CMakeLists.txt

@@ -68,8 +68,6 @@ add_subdirectory(lib)
 add_executable(collector collector.cpp)
 target_link_libraries(collector collector_lib)
 
-target_link_libraries(collector prometheus-cpp::core prometheus-cpp::pull)
-
 add_executable(connscrape connscrape.cpp)
 target_link_libraries(connscrape collector_lib)
 

collector/lib/CMakeLists.txt

@@ -14,6 +14,7 @@ target_link_libraries(collector_lib uuid)
 target_link_libraries(collector_lib gRPC::grpc++)
 target_link_libraries(collector_lib civetweb::civetweb-cpp)
 target_link_libraries(collector_lib yaml-cpp::yaml-cpp)
+target_link_libraries(collector_lib prometheus-cpp::core prometheus-cpp::pull)
 
 target_link_libraries(collector_lib rox-proto)
 

collector/lib/CollectorConnectionStats.h

@@ -0,0 +1,61 @@
+#ifndef _COLLECTOR_CONNECTION_STATS_
+#define _COLLECTOR_CONNECTION_STATS_
+
+#include <algorithm>
+
+#include "prometheus/gauge.h"
+#include "prometheus/registry.h"
+#include "prometheus/summary.h"
+
+namespace collector {
+template <typename T>
+class CollectorConnectionStats {
+ public:
+  CollectorConnectionStats(
+      prometheus::Registry* registry,
+      const std::string& name,
+      const std::string& help,
+      std::chrono::milliseconds max_age,
+      const std::vector<double>& quantiles,
+      double error) {
+    auto& family = prometheus::BuildSummary()
+                       .Name(name)
+                       .Help(help)
+                       .Register(*registry);
+    auto q = MakeQuantiles(quantiles, error);
+    inbound_private_summary_ = &family.Add({{"dir", "in"}, {"peer", "private"}}, q, max_age);
+    inbound_public_summary_ = &family.Add({{"dir", "in"}, {"peer", "public"}}, q, max_age);
+    outbound_private_summary_ = &family.Add({{"dir", "out"}, {"peer", "private"}}, q, max_age);
+    outbound_public_summary_ = &family.Add({{"dir", "out"}, {"peer", "public"}}, q, max_age);
+  }
+
+  void Observe(T inbound_private, T inbound_public, T outbound_private, T outbound_public) {
+    inbound_private_summary_->Observe(inbound_private);
+    inbound_public_summary_->Observe(inbound_public);
+    outbound_private_summary_->Observe(outbound_private);
+    outbound_public_summary_->Observe(outbound_public);
+  }
+
+ private:
+  prometheus::Summary* inbound_private_summary_;
+  prometheus::Summary* inbound_public_summary_;
+  prometheus::Summary* outbound_private_summary_;
+  prometheus::Summary* outbound_public_summary_;
+
+  prometheus::Summary::Quantiles MakeQuantiles(std::vector<double> quantiles, double error) {
+    prometheus::Summary::Quantiles result;
+
+    result.reserve(quantiles.size());
+
+    auto make_quantile = [error](double q) -> prometheus::detail::CKMSQuantiles::Quantile {
+      return {q, error};
+    };
+
+    std::transform(quantiles.begin(), quantiles.end(), std::back_inserter(result), make_quantile);
+
+    return result;
+  }
+};
+}  // namespace collector
+
+#endif

collector/lib/CollectorService.cpp

@@ -32,9 +32,8 @@ CollectorService::CollectorService(CollectorConfig& config, std::atomic<ControlV
       control_(control),
       signum_(*signum),
       server_(OPTIONS),
-      registry_(std::make_shared<prometheus::Registry>()),
       exposer_(PROMETHEUS_PORT),
-      exporter_(registry_, &config_, &system_inspector_),
+      exporter_(&config_, &system_inspector_),
       config_loader_(config_) {
   CLOG(INFO) << "Config: " << config_;
 
@@ -43,28 +42,17 @@ CollectorService::CollectorService(CollectorConfig& config, std::atomic<ControlV
     // In case if no GRPC is used, continue to setup networking infrasturcture
     // with empty grpc_channel. NetworkConnectionInfoServiceComm will pick it
     // up and use stdout instead.
-    if (config_.IsProcessesListeningOnPortsEnabled()) {
-      process_store_ = std::make_shared<ProcessStore>(&system_inspector_);
-    }
-    conn_scraper_ = std::make_shared<ConnScraper>(config_.HostProc(), process_store_);
     conn_tracker_ = std::make_shared<ConnectionTracker>();
     UnorderedSet<L4ProtoPortPair> ignored_l4proto_port_pairs(config_.IgnoredL4ProtoPortPairs());
     conn_tracker_->UpdateIgnoredL4ProtoPortPairs(std::move(ignored_l4proto_port_pairs));
     conn_tracker_->UpdateIgnoredNetworks(config_.IgnoredNetworks());
     conn_tracker_->UpdateNonAggregatedNetworks(config_.NonAggregatedNetworks());
 
-    network_connection_info_service_comm_ = std::make_shared<NetworkConnectionInfoServiceComm>(config_.grpc_channel);
-
-    auto total_reporter = config_.EnableConnectionStats() ? exporter_.GetConnectionsTotalReporter() : nullptr;
-    auto rate_reporter = config_.EnableConnectionStats() ? exporter_.GetConnectionsRateReporter() : nullptr;
-
     net_status_notifier_ = std::make_unique<NetworkStatusNotifier>(
-        conn_scraper_,
         conn_tracker_,
-        network_connection_info_service_comm_,
         config_,
-        total_reporter,
-        rate_reporter);
+        &system_inspector_,
+        exporter_.GetRegistry().get());
 
     auto network_signal_handler = std::make_unique<NetworkSignalHandler>(system_inspector_.GetInspector(), conn_tracker_, system_inspector_.GetUserspaceStats());
     network_signal_handler->SetCollectConnectionStatus(config_.CollectConnectionStatus());
@@ -88,17 +76,18 @@ CollectorService::CollectorService(CollectorConfig& config, std::atomic<ControlV
   }
 
   // Prometheus
-  exposer_.RegisterCollectable(registry_);
+  exposer_.RegisterCollectable(exporter_.GetRegistry());
 }
 
 CollectorService::~CollectorService() {
   config_loader_.Stop();
   server_.close();
-  exporter_.stop();
   if (net_status_notifier_) {
     net_status_notifier_->Stop();
   }
 
+  exporter_.stop();
+
   // system_inspector_ needs to be last, since other components relay on it.
   system_inspector_.CleanUp();
 }

collector/lib/CollectorService.h

@@ -45,15 +45,13 @@ class CollectorService {
   std::vector<std::unique_ptr<CivetWrapper>> civet_endpoints_;
 
   // Prometheus
-  std::shared_ptr<prometheus::Registry> registry_;
   prometheus::Exposer exposer_;
   CollectorStatsExporter exporter_;
 
   ConfigLoader config_loader_;
 
   // Network monitoring
   std::shared_ptr<ConnectionTracker> conn_tracker_;
-  std::shared_ptr<IConnScraper> conn_scraper_;
   std::unique_ptr<NetworkStatusNotifier> net_status_notifier_;
   std::shared_ptr<ProcessStore> process_store_;
   std::shared_ptr<NetworkConnectionInfoServiceComm> network_connection_info_service_comm_;

collector/lib/CollectorStats.h

@@ -96,14 +96,6 @@ class CollectorStats {
   CollectorStats() {};
 };
 
-template <typename T>
-class CollectorConnectionStats {
- public:
-  virtual void Observe(T inbound_private, T inbound_public, T outbound_private, T outbound_public) = 0;
-
-  virtual ~CollectorConnectionStats() {}
-};
-
 namespace internal {
 
 template <typename T>

collector/lib/CollectorStatsExporter.cpp

@@ -9,73 +9,14 @@
 #include "Logging.h"
 #include "Utility.h"
 #include "prometheus/gauge.h"
-#include "prometheus/summary.h"
 #include "system-inspector/Service.h"
 
 namespace collector {
 
-template <typename T>
-class CollectorConnectionStatsPrometheus : public CollectorConnectionStats<T> {
- public:
-  CollectorConnectionStatsPrometheus(
-      std::shared_ptr<prometheus::Registry> registry,
-      std::string name,
-      std::string help,
-      std::chrono::milliseconds max_age,
-      const std::vector<double>& quantiles,
-      double error) : family_(prometheus::BuildSummary().Name(name).Help(help).Register(*registry)),
-                      inbound_private_summary_(family_.Add({{"dir", "in"}, {"peer", "private"}}, MakeQuantiles(quantiles, error), max_age)),
-                      inbound_public_summary_(family_.Add({{"dir", "in"}, {"peer", "public"}}, MakeQuantiles(quantiles, error), max_age)),
-                      outbound_private_summary_(family_.Add({{"dir", "out"}, {"peer", "private"}}, MakeQuantiles(quantiles, error), max_age)),
-                      outbound_public_summary_(family_.Add({{"dir", "out"}, {"peer", "public"}}, MakeQuantiles(quantiles, error), max_age)) {}
-
-  void Observe(T inbound_private, T inbound_public, T outbound_private, T outbound_public) override {
-    inbound_private_summary_.Observe(inbound_private);
-    inbound_public_summary_.Observe(inbound_public);
-    outbound_private_summary_.Observe(outbound_private);
-    outbound_public_summary_.Observe(outbound_public);
-  }
-
- private:
-  prometheus::Family<prometheus::Summary>& family_;
-  prometheus::Summary& inbound_private_summary_;
-  prometheus::Summary& inbound_public_summary_;
-  prometheus::Summary& outbound_private_summary_;
-  prometheus::Summary& outbound_public_summary_;
-
-  prometheus::Summary::Quantiles MakeQuantiles(std::vector<double> quantiles, double error) {
-    prometheus::Summary::Quantiles result;
-
-    result.reserve(quantiles.size());
-
-    auto make_quantile = [error](double q) -> prometheus::detail::CKMSQuantiles::Quantile {
-      return {q, error};
-    };
-
-    std::transform(quantiles.begin(), quantiles.end(), std::back_inserter(result), make_quantile);
-
-    return result;
-  }
-};
-
-CollectorStatsExporter::CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, system_inspector::Service* si)
-    : registry_(std::move(registry)),
+CollectorStatsExporter::CollectorStatsExporter(const CollectorConfig* config, system_inspector::Service* si)
+    : registry_(std::make_shared<prometheus::Registry>()),
       config_(config),
-      system_inspector_(si),
-      connections_total_reporter_(std::make_shared<CollectorConnectionStatsPrometheus<unsigned int>>(
-          registry_,
-          "rox_connections_total",
-          "Amount of stored connections over time",
-          std::chrono::minutes{config->GetConnectionStatsWindow()},
-          config->GetConnectionStatsQuantiles(),
-          config->GetConnectionStatsError())),
-      connections_rate_reporter_(std::make_shared<CollectorConnectionStatsPrometheus<float>>(
-          registry_,
-          "rox_connections_rate",
-          "Rate of connections over time",
-          std::chrono::minutes{config->GetConnectionStatsWindow()},
-          config->GetConnectionStatsQuantiles(),
-          config->GetConnectionStatsError())) {}
+      system_inspector_(si) {}
 
 bool CollectorStatsExporter::start() {
   if (!thread_.Start(&CollectorStatsExporter::run, this)) {

collector/lib/CollectorStatsExporter.h

@@ -4,30 +4,27 @@
 #include <memory>
 
 #include "CollectorConfig.h"
+#include "CollectorConnectionStats.h"
 #include "CollectorStats.h"
 #include "StoppableThread.h"
 #include "prometheus/registry.h"
 #include "system-inspector/Service.h"
 
 namespace collector {
-
 class CollectorStatsExporter {
  public:
-  CollectorStatsExporter(std::shared_ptr<prometheus::Registry> registry, const CollectorConfig* config, system_inspector::Service* si);
+  CollectorStatsExporter(const CollectorConfig* config, system_inspector::Service* si);
 
   bool start();
   void run();
   void stop();
 
-  std::shared_ptr<CollectorConnectionStats<unsigned int>> GetConnectionsTotalReporter() { return connections_total_reporter_; }
-  std::shared_ptr<CollectorConnectionStats<float>> GetConnectionsRateReporter() { return connections_rate_reporter_; }
+  std::shared_ptr<prometheus::Registry>& GetRegistry() { return registry_; }
 
  private:
   std::shared_ptr<prometheus::Registry> registry_;
   const CollectorConfig* config_;
   system_inspector::Service* system_inspector_;
-  std::shared_ptr<CollectorConnectionStats<unsigned int>> connections_total_reporter_;
-  std::shared_ptr<CollectorConnectionStats<float>> connections_rate_reporter_;
   StoppableThread thread_;
 };