Use the new internal service for network connections

This is achieved by replacing the NetworkConnectionInfoServiceComm
component with the Output component.

TODO:
- Receiving messages from sensor is not fully implemented, but the
  mechanisms are already in place for it.
- NetworkStatusNotifierTest is broken.

Author: Molter73

collector/lib/Channel.h

@@ -0,0 +1,58 @@
+#ifndef COLLECTOR_CHANNEL_H
+#define COLLECTOR_CHANNEL_H
+
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <queue>
+
+namespace collector {
+template <typename T>
+class Channel {
+ public:
+  Channel(std::size_t capacity = 0)
+      : capacity_(capacity) {
+  }
+
+  friend Channel<T>& operator<<(Channel<T>& ch, const T& data) {
+    std::unique_lock<std::mutex> lock{ch.mutex_};
+    if (ch.capacity_ > 0 && ch.queue_.size() >= ch.capacity_) {
+      ch.cv_.wait(lock, [&ch]() { return ch.queue_.size() < ch.capacity_; });
+    }
+
+    ch.queue_.push(data);
+    ch.cv_.notify_one();
+    return ch;
+  }
+
+  friend Channel<T>& operator<<(Channel<T>& ch, T&& data) {
+    std::unique_lock<std::mutex> lock{ch.mutex_};
+    if (ch.capacity_ > 0 && ch.queue_.size() >= ch.capacity_) {
+      ch.cv_.wait(lock, [&ch]() { return ch.queue_.size() < ch.capacity_; });
+    }
+
+    ch.queue_.emplace(std::move(data));
+    ch.cv_.notify_one();
+    return ch;
+  }
+
+  friend Channel<T>& operator>>(Channel<T>& ch, T& out) {
+    std::unique_lock<std::mutex> lock{ch.mutex_};
+    ch.cv_.wait(lock, [&ch]() { return ch.queue_.size() > 0; });
+
+    out = std::move(ch.queue_.front());
+    ch.queue_.pop();
+
+    ch.cv_.notify_one();
+    return ch;
+  }
+
+ private:
+  std::size_t capacity_;
+  std::queue<T> queue_;
+  std::mutex mutex_;
+  std::condition_variable cv_;
+};
+}  // namespace collector
+
+#endif

collector/lib/CollectorService.cpp

@@ -52,6 +52,7 @@ CollectorService::CollectorService(CollectorConfig& config, std::atomic<ControlV
         conn_tracker_,
         config_,
         &system_inspector_,
+        &output_,
         exporter_.GetRegistry().get());
 
     auto network_signal_handler = std::make_unique<NetworkSignalHandler>(system_inspector_.GetInspector(), conn_tracker_, system_inspector_.GetUserspaceStats());

collector/lib/CollectorService.h

@@ -53,7 +53,6 @@ class CollectorService {
   std::shared_ptr<ConnectionTracker> conn_tracker_;
   std::unique_ptr<NetworkStatusNotifier> net_status_notifier_;
   std::shared_ptr<ProcessStore> process_store_;
-  std::shared_ptr<NetworkConnectionInfoServiceComm> network_connection_info_service_comm_;
 };
 
 }  // namespace collector

collector/lib/NetworkConnectionInfoServiceComm.cpp

@@ -114,24 +114,10 @@ void NetworkStatusNotifier::Run() {
   auto next_attempt = std::chrono::system_clock::now();
 
   while (thread_.PauseUntil(next_attempt)) {
-    comm_->ResetClientContext();
-
-    if (!comm_->WaitForConnectionReady([this] { return thread_.should_stop(); })) {
-      break;
-    }
-
-    auto client_writer = comm_->PushNetworkConnectionInfoOpenStream([this](const sensor::NetworkFlowsControlMessage* msg) { OnRecvControlMessage(msg); });
-
-    RunSingle(client_writer.get());
+    RunSingle();
     if (thread_.should_stop()) {
       return;
     }
-    auto status = client_writer->Finish(std::chrono::seconds(5));
-    if (status.ok()) {
-      CLOG(ERROR) << "Error streaming network connection info: server hung up unexpectedly";
-    } else {
-      CLOG(ERROR) << "Error streaming network connection info: " << status.error_message();
-    }
     next_attempt = std::chrono::system_clock::now() + std::chrono::seconds(10);
   }
 
@@ -144,7 +130,6 @@ void NetworkStatusNotifier::Start() {
 }
 
 void NetworkStatusNotifier::Stop() {
-  comm_->TryCancel();
   thread_.Stop();
 }
 
@@ -217,15 +202,13 @@ bool NetworkStatusNotifier::UpdateAllConnsAndEndpoints() {
   return true;
 }
 
-void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnectionInfoMessage>* writer) {
-  WaitUntilWriterStarted(writer, 10);
-
+void NetworkStatusNotifier::RunSingle() {
   ConnMap old_conn_state;
   AdvertisedEndpointMap old_cep_state;
   auto next_scrape = std::chrono::system_clock::now();
   int64_t time_at_last_scrape = NowMicros();
 
-  while (writer->Sleep(next_scrape)) {
+  while (thread_.PauseUntil(next_scrape)) {
     CLOG(DEBUG) << "Starting network status notification";
     next_scrape = std::chrono::system_clock::now() + std::chrono::seconds(config_.ScrapeInterval());
 
@@ -272,7 +255,7 @@ void NetworkStatusNotifier::RunSingle(IDuplexClientWriter<sensor::NetworkConnect
     }
 
     WITH_TIMER(CollectorStats::net_write_message) {
-      if (!writer->Write(*msg, next_scrape)) {
+      if (output_->SendMsg(*msg) != SignalHandler::PROCESSED) {
         CLOG(ERROR) << "Failed to write network connection info";
         return;
       }

collector/lib/NetworkConnectionInfoServiceComm.h

@@ -8,10 +8,11 @@
 #include "CollectorConfig.h"
 #include "CollectorConnectionStats.h"
 #include "ConnTracker.h"
-#include "NetworkConnectionInfoServiceComm.h"
+#include "DuplexGRPC.h"
 #include "ProcfsScraper.h"
 #include "ProtoAllocator.h"
 #include "StoppableThread.h"
+#include "output/Output.h"
 
 namespace collector {
 
@@ -20,11 +21,12 @@ class NetworkStatusNotifier : protected ProtoAllocator<sensor::NetworkConnection
   NetworkStatusNotifier(std::shared_ptr<ConnectionTracker> conn_tracker,
                         const CollectorConfig& config,
                         system_inspector::Service* inspector,
+                        output::Output* output,
                         prometheus::Registry* registry)
       : conn_scraper_(std::make_unique<ConnScraper>(config, inspector)),
         conn_tracker_(std::move(conn_tracker)),
         config_(config),
-        comm_(std::make_unique<NetworkConnectionInfoServiceComm>(config.grpc_channel)) {
+        output_(output) {
     if (config_.EnableConnectionStats()) {
       connections_total_reporter_ = {{registry,
                                       "rox_connections_total",
@@ -55,18 +57,6 @@ class NetworkStatusNotifier : protected ProtoAllocator<sensor::NetworkConnection
     conn_scraper_ = std::move(cs);
   }
 
-  /**
-   * Replace the communications object.
-   *
-   * This is meant to make testing easier by swapping in a mock object.
-   *
-   * @params comm A unique pointer to the new instance of communications
-   *              to use.
-   */
-  void ReplaceComm(std::unique_ptr<INetworkConnectionInfoServiceComm>&& comm) {
-    comm_ = std::move(comm);
-  }
-
  private:
   FRIEND_TEST(NetworkStatusNotifierTest, RateLimitedConnections);
 
@@ -84,7 +74,7 @@ class NetworkStatusNotifier : protected ProtoAllocator<sensor::NetworkConnection
   void Run();
   void WaitUntilWriterStarted(IDuplexClientWriter<sensor::NetworkConnectionInfoMessage>* writer, int wait_time);
   bool UpdateAllConnsAndEndpoints();
-  void RunSingle(IDuplexClientWriter<sensor::NetworkConnectionInfoMessage>* writer);
+  void RunSingle();
   void ReceivePublicIPs(const sensor::IPAddressList& public_ips);
   void ReceiveIPNetworks(const sensor::IPNetworkList& networks);
 
@@ -96,7 +86,7 @@ class NetworkStatusNotifier : protected ProtoAllocator<sensor::NetworkConnection
   std::shared_ptr<ConnectionTracker> conn_tracker_;
 
   const CollectorConfig& config_;
-  std::unique_ptr<INetworkConnectionInfoServiceComm> comm_;
+  output::Output* output_;
 
   std::optional<CollectorConnectionStats<unsigned int>> connections_total_reporter_;
   std::optional<CollectorConnectionStats<float>> connections_rate_reporter_;