ROX-28527: Send networkflows through the new iservice

collector/lib/Channel.h

@@ -0,0 +1,78 @@
+#ifndef COLLECTOR_CHANNEL_H
+#define COLLECTOR_CHANNEL_H
+
+#include <condition_variable>
+#include <cstddef>
+#include <mutex>
+#include <optional>
+#include <queue>
+
+namespace collector {
+template <typename T>
+class ChannelImpl {
+ public:
+  ChannelImpl(std::size_t capacity = 0)
+      : capacity_(capacity) {
+  }
+
+  ChannelImpl& Send(const T& data) {
+    std::unique_lock<std::mutex> lock{mutex_};
+    if (capacity_ > 0 && queue_.size() >= capacity_) {
+      cv_.wait(lock, [this] { return queue_.size() < capacity_; });
+    }
+
+    queue_.push(data);
+    cv_.notify_one();
+
+    return *this;
+  }
+
+  ChannelImpl& Send(T&& data) {
+    std::unique_lock<std::mutex> lock{mutex_};
+    if (capacity_ > 0 && queue_.size() >= capacity_) {
+      cv_.wait(lock, [this] { return queue_.size() < capacity_; });
+    }
+
+    queue_.push(std::move(data));
+    cv_.notify_one();
+
+    return *this;
+  }
+
+  T Recv() {
+    std::unique_lock<std::mutex> lock{mutex_};
+    cv_.wait(lock, [this] { return queue_.size() > 0; });
+
+    T out = std::move(queue_.front());
+    queue_.pop();
+
+    cv_.notify_one();
+    return out;
+  }
+
+  std::optional<T> TryRecv() {
+    std::unique_lock<std::mutex> lock{mutex_};
+    if (queue_.empty()) {
+      return {};
+    }
+
+    return Recv();
+  }
+
+  bool Empty() {
+    std::unique_lock<std::mutex> lock{mutex_};
+    return queue_.empty();
+  }
+
+ private:
+  std::size_t capacity_;
+  std::queue<T> queue_;
+  std::mutex mutex_;
+  std::condition_variable cv_;
+};
+
+template <typename T>
+using Channel = std::shared_ptr<ChannelImpl<T>>;
+}  // 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/NetworkStatusNotifier.cpp

@@ -57,15 +57,12 @@ std::vector<IPNet> readNetworks(const std::string& networks, Address::Family fam
   return ip_nets;
 }
 
-void NetworkStatusNotifier::OnRecvControlMessage(const sensor::NetworkFlowsControlMessage* msg) {
-  if (!msg) {
-    return;
-  }
-  if (msg->has_ip_networks()) {
-    ReceivePublicIPs(msg->public_ip_addresses());
+void NetworkStatusNotifier::OnRecvControlMessage(const sensor::NetworkFlowsControlMessage& msg) {
+  if (msg.has_ip_networks()) {
+    ReceivePublicIPs(msg.public_ip_addresses());
   }
-  if (msg->has_ip_networks()) {
-    ReceiveIPNetworks(msg->ip_networks());
+  if (msg.has_ip_networks()) {
+    ReceiveIPNetworks(msg.ip_networks());
   }
 }
 
@@ -114,24 +111,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);
   }
 
@@ -140,12 +123,13 @@ void NetworkStatusNotifier::Run() {
 
 void NetworkStatusNotifier::Start() {
   thread_.Start([this] { Run(); });
+  receiver_.Start();
   CLOG(INFO) << "Started network status notifier.";
 }
 
 void NetworkStatusNotifier::Stop() {
-  comm_->TryCancel();
   thread_.Stop();
+  receiver_.Stop();
 }
 
 void NetworkStatusNotifier::WaitUntilWriterStarted(IDuplexClientWriter<sensor::NetworkConnectionInfoMessage>* writer, int wait_time_seconds) {
@@ -217,15 +201,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 +254,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;
       }