Rename and reorganize output namespace

Simplify naming for classes in the output namespace.
Create dedicated namespaces for gprc and log implementations of IClient.

Author: Molter73

collector/lib/CMakeLists.txt

@@ -2,6 +2,7 @@ file(GLOB COLLECTOR_LIB_SRC_FILES
     ${CMAKE_CURRENT_SOURCE_DIR}/*.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/system-inspector/*.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/output/*.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/output/*/*.cpp
 )
 
 add_library(collector_lib ${DRIVER_HEADERS} ${COLLECTOR_LIB_SRC_FILES})

collector/lib/ProcessSignalHandler.h

@@ -6,7 +6,7 @@
 #include "ProcessSignalFormatter.h"
 #include "RateLimit.h"
 #include "SignalHandler.h"
-#include "output/SensorClientFormatter.h"
+#include "output/Formatter.h"
 #include "system-inspector/Service.h"
 
 // forward declarations
@@ -50,7 +50,7 @@ class ProcessSignalHandler : public SignalHandler {
 
   output::Output* client_;
   ProcessSignalFormatter signal_formatter_;
-  output::SensorClientFormatter sensor_formatter_;
+  output::Formatter sensor_formatter_;
   system_inspector::Stats* stats_;
   RateLimitCache rate_limiter_;
 };

collector/lib/output/SensorClientFormatter.cpp renamed to collector/lib/output/Formatter.cpp

@@ -1,4 +1,4 @@
-#include "SensorClientFormatter.h"
+#include "Formatter.h"
 
 #include <uuid/uuid.h>
 
@@ -16,9 +16,9 @@
 namespace collector::output {
 
 using SignalStreamMessage = sensor::SignalStreamMessage;
-using Signal = SensorClientFormatter::Signal;
-using ProcessSignal = SensorClientFormatter::ProcessSignal;
-using LineageInfo = SensorClientFormatter::LineageInfo;
+using Signal = Formatter::Signal;
+using ProcessSignal = Formatter::ProcessSignal;
+using LineageInfo = Formatter::LineageInfo;
 
 using Timestamp = google::protobuf::Timestamp;
 using TimeUtil = google::protobuf::util::TimeUtil;
@@ -57,17 +57,17 @@ std::string extract_proc_args(sinsp_threadinfo* tinfo) {
 
 }  // namespace
 
-SensorClientFormatter::SensorClientFormatter(sinsp* inspector, const CollectorConfig& config)
+Formatter::Formatter(sinsp* inspector, const CollectorConfig& config)
     : event_names_(EventNames::GetInstance()),
       event_extractor_(std::make_unique<system_inspector::EventExtractor>()),
       container_metadata_(inspector),
       config_(config) {
   event_extractor_->Init(inspector);
 }
 
-SensorClientFormatter::~SensorClientFormatter() = default;
+Formatter::~Formatter() = default;
 
-const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_evt* event) {
+const sensor::ProcessSignal* Formatter::ToProtoMessage(sinsp_evt* event) {
   if (process_signals[event->get_type()] == ProcessSignalType::UNKNOWN_PROCESS_TYPE) {
     return nullptr;
   }
@@ -82,7 +82,7 @@ const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_evt* ev
   return CreateProcessSignal(event);
 }
 
-const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_threadinfo* tinfo) {
+const sensor::ProcessSignal* Formatter::ToProtoMessage(sinsp_threadinfo* tinfo) {
   Reset();
   if (!ValidateProcessDetails(tinfo)) {
     CLOG(INFO) << "Dropping process event: " << tinfo;
@@ -92,7 +92,7 @@ const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_threadi
   return CreateProcessSignal(tinfo);
 }
 
-ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_evt* event) {
+ProcessSignal* Formatter::CreateProcessSignal(sinsp_evt* event) {
   auto signal = AllocateRoot();
 
   // set id
@@ -174,7 +174,7 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_evt* event) {
   return signal;
 }
 
-ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_threadinfo* tinfo) {
+ProcessSignal* Formatter::CreateProcessSignal(sinsp_threadinfo* tinfo) {
   auto signal = AllocateRoot();
 
   // set id
@@ -237,7 +237,7 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_threadinfo* tinf
   return signal;
 }
 
-std::string SensorClientFormatter::ToString(sinsp_evt* event) {
+std::string Formatter::ToString(sinsp_evt* event) {
   std::stringstream ss;
   const std::string* path = event_extractor_->get_exepath(event);
   const std::string* name = event_extractor_->get_comm(event);
@@ -254,7 +254,7 @@ std::string SensorClientFormatter::ToString(sinsp_evt* event) {
   return ss.str();
 }
 
-bool SensorClientFormatter::ValidateProcessDetails(const sinsp_threadinfo* tinfo) {
+bool Formatter::ValidateProcessDetails(const sinsp_threadinfo* tinfo) {
   if (tinfo == nullptr) {
     return false;
   }
@@ -266,11 +266,11 @@ bool SensorClientFormatter::ValidateProcessDetails(const sinsp_threadinfo* tinfo
   return true;
 }
 
-bool SensorClientFormatter::ValidateProcessDetails(sinsp_evt* event) {
+bool Formatter::ValidateProcessDetails(sinsp_evt* event) {
   return ValidateProcessDetails(event->get_thread_info());
 }
 
-void SensorClientFormatter::UpdateLineageStats(const std::vector<LineageInfo>& lineage) {
+void Formatter::UpdateLineageStats(const std::vector<LineageInfo>& lineage) {
   int string_total = std::accumulate(lineage.cbegin(), lineage.cend(), 0, [](int acc, const auto& l) {
     return acc + l.parent_exec_file_path().size();
   });
@@ -281,7 +281,7 @@ void SensorClientFormatter::UpdateLineageStats(const std::vector<LineageInfo>& l
   COUNTER_ADD(CollectorStats::process_lineage_string_total, string_total);
 }
 
-std::vector<LineageInfo> SensorClientFormatter::GetProcessLineage(sinsp_threadinfo* tinfo) {
+std::vector<LineageInfo> Formatter::GetProcessLineage(sinsp_threadinfo* tinfo) {
   std::vector<LineageInfo> lineage;
   if (tinfo == nullptr) {
     return lineage;

collector/lib/output/SensorClientFormatter.h renamed to collector/lib/output/Formatter.h

@@ -21,15 +21,15 @@ class EventExtractor;
 
 namespace collector::output {
 
-class SensorClientFormatter : public ProtoSignalFormatter<sensor::ProcessSignal> {
+class Formatter : public ProtoSignalFormatter<sensor::ProcessSignal> {
  public:
-  SensorClientFormatter(const SensorClientFormatter&) = delete;
-  SensorClientFormatter(SensorClientFormatter&&) = delete;
-  SensorClientFormatter& operator=(const SensorClientFormatter&) = delete;
-  SensorClientFormatter& operator=(SensorClientFormatter&&) = delete;
-  virtual ~SensorClientFormatter();
+  Formatter(const Formatter&) = delete;
+  Formatter(Formatter&&) = delete;
+  Formatter& operator=(const Formatter&) = delete;
+  Formatter& operator=(Formatter&&) = delete;
+  virtual ~Formatter();
 
-  SensorClientFormatter(sinsp* inspector, const CollectorConfig& config);
+  Formatter(sinsp* inspector, const CollectorConfig& config);
 
   using Signal = v1::Signal;
   using ProcessSignal = sensor::ProcessSignal;

collector/lib/output/IClient.h

@@ -0,0 +1,42 @@
+#ifndef OUTPUT_ICLIENT_H
+#define OUTPUT_ICLIENT_H
+
+#include "internalapi/sensor/collector_iservice.grpc.pb.h"
+
+#include "SignalHandler.h"
+
+namespace collector::output {
+
+class IClient {
+ public:
+  using Service = sensor::CollectorService;
+
+  IClient() = default;
+  IClient(const IClient&) = default;
+  IClient(IClient&&) = delete;
+  IClient& operator=(const IClient&) = default;
+  IClient& operator=(IClient&&) = delete;
+  virtual ~IClient() = default;
+
+  /**
+   * Recreate the internal state of the object to allow communication.
+   *
+   * Mostly useful for handling gRPC reconnections.
+   *
+   * @returns true if the refresh was succesful, false otherwise.
+   */
+  virtual bool Recreate() = 0;
+
+  /**
+   * Send a message to sensor through the iservice.
+   *
+   * @param msg The message to be sent to sensor.
+   * @returns A SignalHandler::Result with the outcome of the send
+   *          operation.
+   */
+  virtual SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) = 0;
+};
+
+}  // namespace collector::output
+
+#endif  // OUTPUT_ICLIENT_H

collector/lib/output/Output.cpp

@@ -1,7 +1,8 @@
 #include "Output.h"
 
 #include "GRPCUtil.h"
-#include "HostInfo.h"
+#include "output/grpc/Client.h"
+#include "output/log/Client.h"
 
 namespace collector::output {
 
@@ -11,7 +12,7 @@ Output::Output(const CollectorConfig& config)
     channel_ = config.grpc_channel;
 
     if (use_sensor_client_) {
-      auto sensor_client = std::make_unique<SensorClient>(channel_);
+      auto sensor_client = std::make_unique<grpc::Client>(channel_);
       sensor_clients_.emplace_back(std::move(sensor_client));
     } else {
       auto signal_client = std::make_unique<SignalServiceClient>(channel_);
@@ -21,7 +22,7 @@ Output::Output(const CollectorConfig& config)
 
   if (config.grpc_channel == nullptr || config.UseStdout()) {
     if (use_sensor_client_) {
-      auto sensor_client = std::make_unique<SensorClientStdout>();
+      auto sensor_client = std::make_unique<log::Client>();
       sensor_clients_.emplace_back(std::move(sensor_client));
     } else {
       auto signal_client = std::make_unique<StdoutSignalServiceClient>();

collector/lib/output/Output.h

@@ -5,7 +5,7 @@
 #include "internalapi/sensor/signal_iservice.pb.h"
 
 #include "CollectorConfig.h"
-#include "SensorClient.h"
+#include "IClient.h"
 #include "SignalHandler.h"
 #include "SignalServiceClient.h"
 #include "StoppableThread.h"
@@ -31,7 +31,7 @@ class Output {
   }
 
   // Constructor for tests
-  Output(std::unique_ptr<ISensorClient>&& sensor_client,
+  Output(std::unique_ptr<IClient>&& sensor_client,
          std::unique_ptr<ISignalServiceClient>&& signal_client) {
     sensor_clients_.emplace_back(std::move(sensor_client));
     signal_clients_.emplace_back(std::move(signal_client));
@@ -64,7 +64,7 @@ class Output {
   SignalHandler::Result SensorOutput(const sensor::ProcessSignal& msg);
   SignalHandler::Result SignalOutput(const sensor::SignalStreamMessage& msg);
 
-  std::vector<std::unique_ptr<ISensorClient>> sensor_clients_;
+  std::vector<std::unique_ptr<IClient>> sensor_clients_;
   std::vector<std::unique_ptr<ISignalServiceClient>> signal_clients_;
 
   bool use_sensor_client_ = true;

collector/lib/output/SensorClient.cpp renamed to collector/lib/output/grpc/Client.cpp

@@ -1,10 +1,10 @@
-#include "SensorClient.h"
+#include "Client.h"
 
 #include "Logging.h"
 
-namespace collector::output {
-bool SensorClient::Recreate() {
-  context_ = std::make_unique<grpc::ClientContext>();
+namespace collector::output::grpc {
+bool Client::Recreate() {
+  context_ = std::make_unique<::grpc::ClientContext>();
   writer_ = DuplexClient::CreateWithReadsIgnored(&sensor::CollectorService::Stub::AsyncPushProcesses, channel_, context_.get());
   if (!writer_->WaitUntilStarted(std::chrono::seconds(30))) {
     CLOG(ERROR) << "Signal stream not ready after 30 seconds. Retrying ...";
@@ -18,7 +18,7 @@ bool SensorClient::Recreate() {
   return true;
 }
 
-SignalHandler::Result SensorClient::SendMsg(const sensor::ProcessSignal& msg) {
+SignalHandler::Result Client::SendMsg(const sensor::ProcessSignal& msg) {
   if (!stream_active_.load(std::memory_order_acquire)) {
     CLOG_THROTTLED(ERROR, std::chrono::seconds(10))
         << "GRPC stream is not established";
@@ -44,4 +44,4 @@ SignalHandler::Result SensorClient::SendMsg(const sensor::ProcessSignal& msg) {
 
   return SignalHandler::PROCESSED;
 }
-}  // namespace collector::output
+}  // namespace collector::output::grpc

collector/lib/output/grpc/Client.h

@@ -0,0 +1,46 @@
+#ifndef OUTPUT_GRPC_CLIENT
+#define OUTPUT_GRPC_CLIENT
+
+#include <grpcpp/channel.h>
+
+#include "DuplexGRPC.h"
+#include "output/IClient.h"
+
+namespace collector::output::grpc {
+
+using Channel = ::grpc::Channel;
+
+class Client : public IClient {
+ public:
+  using Service = sensor::CollectorService;
+
+  Client(const Client&) = delete;
+  Client(Client&&) = delete;
+  Client& operator=(const Client&) = delete;
+  Client& operator=(Client&&) = delete;
+  ~Client() override {
+    context_->TryCancel();
+  }
+
+  explicit Client(std::shared_ptr<Channel> channel)
+      : channel_(std::move(channel)) {
+  }
+
+  bool Recreate() override;
+
+  SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) override;
+
+ private:
+  std::shared_ptr<Channel> channel_;
+
+  std::atomic<bool> stream_active_ = false;
+
+  // This needs to have the same lifetime as the class.
+  std::unique_ptr<::grpc::ClientContext> context_;
+  std::unique_ptr<IDuplexClientWriter<sensor::ProcessSignal>> writer_;
+
+  bool first_write_ = false;
+};
+}  // namespace collector::output::grpc
+
+#endif

collector/lib/output/log/Client.h

@@ -0,0 +1,19 @@
+#ifndef OUTPUT_LOG_CLIENT_H
+#define OUTPUT_LOG_CLIENT_H
+
+#include <google/protobuf/message.h>
+
+#include "Utility.h"
+#include "output/IClient.h"
+
+namespace collector::output::log {
+class Client : public IClient {
+  bool Recreate() override { return true; }
+
+  SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) override {
+    LogProtobufMessage(msg);
+    return SignalHandler::PROCESSED;
+  }
+};
+}  // namespace collector::output::log
+#endif