Create a dedicated output namespace

Molter73 · Molter73 · commit 266274d944bc · 2025-04-15T15:13:14.000+02:00
diff --git a/collector/lib/CMakeLists.txt b/collector/lib/CMakeLists.txt
@@ -1,6 +1,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
 )
 
 add_library(collector_lib ${DRIVER_HEADERS} ${COLLECTOR_LIB_SRC_FILES})
diff --git a/collector/lib/CollectorConfig.h b/collector/lib/CollectorConfig.h
@@ -199,8 +199,6 @@ class CollectorConfig {
   }
 
  protected:
-  FRIEND_TEST(SensorClientFormatterTest, NoProcessArguments);
-
   int scrape_interval_;
   CollectionMethod collection_method_;
   bool turn_off_scrape_;
diff --git a/collector/lib/CollectorService.h b/collector/lib/CollectorService.h
@@ -6,12 +6,12 @@
 
 #include "CivetWrapper.h"
 #include "CollectorConfig.h"
-#include "CollectorOutput.h"
 #include "CollectorStatsExporter.h"
 #include "ConfigLoader.h"
 #include "Control.h"
 #include "NetworkStatusInspector.h"
 #include "NetworkStatusNotifier.h"
+#include "output/Output.h"
 #include "system-inspector/Service.h"
 
 namespace collector {
@@ -34,7 +34,7 @@ class CollectorService {
   bool WaitForGRPCServer();
 
   CollectorConfig& config_;
-  CollectorOutput output_;
+  output::Output output_;
   system_inspector::Service system_inspector_;
 
   std::atomic<ControlValue>* control_;
diff --git a/collector/lib/ProcessSignalHandler.h b/collector/lib/ProcessSignalHandler.h
@@ -6,8 +6,8 @@
 #include "CollectorConfig.h"
 #include "ProcessSignalFormatter.h"
 #include "RateLimit.h"
-#include "SensorClientFormatter.h"
 #include "SignalHandler.h"
+#include "output/SensorClientFormatter.h"
 #include "system-inspector/Service.h"
 
 // forward declarations
@@ -21,7 +21,7 @@ class ProcessSignalHandler : public SignalHandler {
  public:
   ProcessSignalHandler(
       sinsp* inspector,
-      CollectorOutput* client,
+      output::Output* client,
       system_inspector::Stats* stats,
       const CollectorConfig& config)
       : client_(client),
@@ -49,9 +49,9 @@ class ProcessSignalHandler : public SignalHandler {
   Result HandleSensorSignal(sinsp_evt* evt);
   Result HandleExistingProcessSensor(sinsp_threadinfo* tinfo);
 
-  CollectorOutput* client_;
+  output::Output* client_;
   ProcessSignalFormatter signal_formatter_;
-  SensorClientFormatter sensor_formatter_;
+  output::SensorClientFormatter sensor_formatter_;
   system_inspector::Stats* stats_;
   RateLimitCache rate_limiter_;
 };
diff --git a/collector/lib/output/Output.cpp b/collector/lib/output/Output.cpp
@@ -1,10 +1,10 @@
-#include "CollectorOutput.h"
+#include "Output.h"
 
 #include "GRPCUtil.h"
 
-namespace collector {
+namespace collector::output {
 
-CollectorOutput::CollectorOutput(const CollectorConfig& config)
+Output::Output(const CollectorConfig& config)
     : use_sensor_client_(!config.UseLegacyServices()) {
   if (config.grpc_channel != nullptr) {
     channel_ = config.grpc_channel;
@@ -31,13 +31,13 @@ CollectorOutput::CollectorOutput(const CollectorConfig& config)
   thread_.Start([this] { EstablishGrpcStream(); });
 }
 
-void CollectorOutput::HandleOutputError() {
+void Output::HandleOutputError() {
   CLOG(ERROR) << "GRPC stream interrupted";
   stream_active_.store(false, std::memory_order_release);
   stream_interrupted_.notify_one();
 }
 
-SignalHandler::Result CollectorOutput::SensorOutput(const sensor::ProcessSignal& msg) {
+SignalHandler::Result Output::SensorOutput(const sensor::ProcessSignal& msg) {
   for (auto& client : sensor_clients_) {
     auto res = client->SendMsg(msg);
     switch (res) {
@@ -57,7 +57,7 @@ SignalHandler::Result CollectorOutput::SensorOutput(const sensor::ProcessSignal&
   return SignalHandler::PROCESSED;
 }
 
-SignalHandler::Result CollectorOutput::SignalOutput(const sensor::SignalStreamMessage& msg) {
+SignalHandler::Result Output::SignalOutput(const sensor::SignalStreamMessage& msg) {
   for (auto& client : signal_clients_) {
     auto res = client->PushSignals(msg);
     switch (res) {
@@ -77,7 +77,7 @@ SignalHandler::Result CollectorOutput::SignalOutput(const sensor::SignalStreamMe
   return SignalHandler::PROCESSED;
 }
 
-SignalHandler::Result CollectorOutput::SendMsg(const MessageType& msg) {
+SignalHandler::Result Output::SendMsg(const MessageType& msg) {
   auto visitor = [this](auto&& m) {
     using T = std::decay_t<decltype(m)>;
     if constexpr (std::is_same_v<T, sensor::ProcessSignal>) {
@@ -93,13 +93,13 @@ SignalHandler::Result CollectorOutput::SendMsg(const MessageType& msg) {
   return std::visit(visitor, msg);
 }
 
-void CollectorOutput::EstablishGrpcStream() {
+void Output::EstablishGrpcStream() {
   while (EstablishGrpcStreamSingle()) {
   }
   CLOG(INFO) << "Service client terminating.";
 }
 
-bool CollectorOutput::EstablishGrpcStreamSingle() {
+bool Output::EstablishGrpcStreamSingle() {
   std::mutex mtx;
   std::unique_lock<std::mutex> lock(mtx);
   stream_interrupted_.wait(lock, [this]() { return !stream_active_.load(std::memory_order_acquire) || thread_.should_stop(); });
@@ -134,4 +134,4 @@ bool CollectorOutput::EstablishGrpcStreamSingle() {
   }
   return true;
 }
-}  // namespace collector
+}  // namespace collector::output
diff --git a/collector/lib/output/Output.h b/collector/lib/output/Output.h
@@ -11,29 +11,29 @@
 #include "SignalServiceClient.h"
 #include "StoppableThread.h"
 
-namespace collector {
+namespace collector::output {
 
 using MessageType = std::variant<sensor::ProcessSignal, sensor::SignalStreamMessage>;
 
-class CollectorOutput {
+class Output {
  public:
-  CollectorOutput(const CollectorOutput&) = delete;
-  CollectorOutput(CollectorOutput&&) = delete;
-  CollectorOutput& operator=(const CollectorOutput&) = delete;
-  CollectorOutput& operator=(CollectorOutput&&) = delete;
+  Output(const Output&) = delete;
+  Output(Output&&) = delete;
+  Output& operator=(const Output&) = delete;
+  Output& operator=(Output&&) = delete;
 
-  CollectorOutput(const CollectorConfig& config);
+  Output(const CollectorConfig& config);
 
-  ~CollectorOutput() {
+  ~Output() {
     stream_interrupted_.notify_one();
     if (thread_.running()) {
       thread_.Stop();
     }
   }
 
   // Constructor for tests
-  CollectorOutput(std::unique_ptr<ISensorClient>&& sensor_client,
-                  std::unique_ptr<ISignalServiceClient>&& signal_client) {
+  Output(std::unique_ptr<ISensorClient>&& sensor_client,
+         std::unique_ptr<ISignalServiceClient>&& signal_client) {
     sensor_clients_.emplace_back(std::move(sensor_client));
     signal_clients_.emplace_back(std::move(signal_client));
   }
@@ -76,6 +76,6 @@ class CollectorOutput {
   std::shared_ptr<grpc::Channel> channel_;
 };
 
-}  // namespace collector
+}  // namespace collector::output
 
 #endif  // COLLECTOR_OUTPUT_H
diff --git a/collector/lib/output/SensorClient.cpp b/collector/lib/output/SensorClient.cpp
@@ -2,7 +2,7 @@
 
 #include "Logging.h"
 
-namespace collector {
+namespace collector::output {
 bool SensorClient::Recreate() {
   context_ = std::make_unique<grpc::ClientContext>();
   writer_ = DuplexClient::CreateWithReadsIgnored(&sensor::CollectorService::Stub::AsyncPushProcesses, channel_, context_.get());
@@ -44,4 +44,4 @@ SignalHandler::Result SensorClient::SendMsg(const sensor::ProcessSignal& msg) {
 
   return SignalHandler::PROCESSED;
 }
-}  // namespace collector
+}  // namespace collector::output
diff --git a/collector/lib/output/SensorClient.h b/collector/lib/output/SensorClient.h
@@ -10,7 +10,7 @@
 #include "DuplexGRPC.h"
 #include "SignalHandler.h"
 
-namespace collector {
+namespace collector::output {
 
 class ISensorClient {
  public:
@@ -83,6 +83,6 @@ class SensorClientStdout : public ISensorClient {
   }
 };
 
-}  // namespace collector
+}  // namespace collector::output
 
 #endif  //_SENSOR_CLIENT_H_
diff --git a/collector/lib/output/SensorClientFormatter.cpp b/collector/lib/output/SensorClientFormatter.cpp
@@ -13,7 +13,7 @@
 #include "Utility.h"
 #include "system-inspector/EventExtractor.h"
 
-namespace collector {
+namespace collector::output {
 
 using SignalStreamMessage = sensor::SignalStreamMessage;
 using Signal = SensorClientFormatter::Signal;
@@ -331,4 +331,4 @@ std::vector<LineageInfo> SensorClientFormatter::GetProcessLineage(sinsp_threadin
   return lineage;
 }
 
-}  // namespace collector
+}  // namespace collector::output
diff --git a/collector/lib/output/SensorClientFormatter.h b/collector/lib/output/SensorClientFormatter.h
@@ -20,7 +20,7 @@ namespace collector::system_inspector {
 class EventExtractor;
 }
 
-namespace collector {
+namespace collector::output {
 
 class SensorClientFormatter : public ProtoSignalFormatter<sensor::ProcessSignal> {
  public:
@@ -109,6 +109,6 @@ class SensorClientFormatter : public ProtoSignalFormatter<sensor::ProcessSignal>
   const CollectorConfig& config_;
 };
 
-}  // namespace collector
+}  // namespace collector::output
 
 #endif  // SENSOR_CLIENT_FORMATTER_H
diff --git a/collector/lib/system-inspector/Service.cpp b/collector/lib/system-inspector/Service.cpp
@@ -12,9 +12,7 @@
 
 #include <google/protobuf/util/time_util.h>
 
-#include "CollectionMethod.h"
 #include "CollectorException.h"
-#include "CollectorOutput.h"
 #include "CollectorStats.h"
 #include "ContainerEngine.h"
 #include "ContainerMetadata.h"
@@ -35,7 +33,7 @@ namespace collector::system_inspector {
 
 Service::~Service() = default;
 
-Service::Service(const CollectorConfig& config, CollectorOutput* client)
+Service::Service(const CollectorConfig& config, output::Output* client)
     : inspector_(std::make_unique<sinsp>(true)),
       container_metadata_inspector_(std::make_unique<ContainerMetadata>(inspector_.get())),
       default_formatter_(std::make_unique<sinsp_evt_formatter>(
diff --git a/collector/lib/system-inspector/Service.h b/collector/lib/system-inspector/Service.h
@@ -8,11 +8,11 @@
 
 #include <gtest/gtest_prod.h>
 
-#include "CollectorOutput.h"
 #include "ContainerMetadata.h"
 #include "Control.h"
 #include "SignalHandler.h"
 #include "SystemInspector.h"
+#include "output/Output.h"
 
 // forward declarations
 class sinsp;
@@ -30,7 +30,7 @@ class Service : public SystemInspector {
   Service& operator=(Service&&) = delete;
   ~Service() override;
 
-  Service(const CollectorConfig& config, CollectorOutput* client);
+  Service(const CollectorConfig& config, output::Output* client);
   void Start() override;
   void Run(const std::atomic<ControlValue>& control) override;
   void CleanUp() override;
diff --git a/collector/test/OutputTest.cpp b/collector/test/OutputTest.cpp
@@ -1,11 +1,11 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
-#include "CollectorOutput.h"
-#include "SensorClient.h"
 #include "SignalServiceClient.h"
+#include "output/Output.h"
+#include "output/SensorClient.h"
 
-namespace collector {
+namespace collector::output {
 class MockSensorClient : public ISensorClient {
  public:
   MOCK_METHOD(bool, Recreate, ());
@@ -34,7 +34,7 @@ TEST_F(CollectorOutputTest, SensorClient) {
 
   EXPECT_CALL(*sensor_client, SendMsg).Times(1).WillOnce(testing::Return(SignalHandler::PROCESSED));
 
-  CollectorOutput output{std::move(sensor_client), std::move(signal_client)};
+  Output output{std::move(sensor_client), std::move(signal_client)};
   auto result = output.SendMsg(msg);
 
   EXPECT_EQ(result, SignalHandler::PROCESSED);
@@ -45,10 +45,10 @@ TEST_F(CollectorOutputTest, SignalClient) {
 
   EXPECT_CALL(*signal_client, PushSignals).Times(1).WillOnce(testing::Return(SignalHandler::PROCESSED));
 
-  CollectorOutput output{std::move(sensor_client), std::move(signal_client)};
+  Output output{std::move(sensor_client), std::move(signal_client)};
 
   auto result = output.SendMsg(msg);
 
   EXPECT_EQ(result, SignalHandler::PROCESSED);
 }
-}  // namespace collector
+}  // namespace collector::output
diff --git a/collector/test/SensorClientFormatterTest.cpp b/collector/test/SensorClientFormatterTest.cpp
@@ -6,10 +6,10 @@
 #include "libsinsp/sinsp.h"
 
 #include "CollectorStats.h"
-#include "SensorClientFormatter.h"
 #include "Utility.h"
+#include "output/SensorClientFormatter.h"
 
-namespace collector {
+namespace collector::output {
 
 using LineageInfo = SensorClientFormatter::LineageInfo;
 
@@ -23,6 +23,15 @@ struct ThreadInfoParams {
   std::string exepath;
 };
 
+class MockCollectorConfig : public CollectorConfig {
+ public:
+  MockCollectorConfig() = default;
+
+  void SetDisableProcessArguments(bool value) {
+    disable_process_arguments_ = value;
+  }
+};
+
 #define EXPECT_STATS_COUNTER(index, expected) \
   EXPECT_EQ(CollectorStats::GetOrCreate().GetCounter(index), expected)
 
@@ -47,7 +56,7 @@ class SensorClientFormatterTest : public testing::Test {
   }
 
   std::unique_ptr<sinsp> inspector;
-  CollectorConfig config;
+  MockCollectorConfig config;
   SensorClientFormatter formatter;
 };
 
@@ -317,7 +326,7 @@ TEST_F(SensorClientFormatterTest, ProcessArguments) {
 }
 
 TEST_F(SensorClientFormatterTest, NoProcessArguments) {
-  config.disable_process_arguments_ = true;
+  config.SetDisableProcessArguments(true);
 
   // {pid, tid, ptid, vpid, uid, container_id, exepath},
   auto tinfo = build_threadinfo({3, 3, -1, 0, 42, "", "qwerty"});
@@ -336,4 +345,4 @@ TEST_F(SensorClientFormatterTest, NoProcessArguments) {
   EXPECT_TRUE(signal->args().empty());
 }
 
-}  // namespace collector
+}  // namespace collector::output

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`file(GLOB COLLECTOR_LIB_SRC_FILES`
`2`	`2`	`${CMAKE_CURRENT_SOURCE_DIR}/*.cpp`
`3`	`3`	`${CMAKE_CURRENT_SOURCE_DIR}/system-inspector/*.cpp`
	`4`	`+ ${CMAKE_CURRENT_SOURCE_DIR}/output/*.cpp`
`4`	`5`	`)`
`5`	`6`
`6`	`7`	`add_library(collector_lib ${DRIVER_HEADERS} ${COLLECTOR_LIB_SRC_FILES})`
Original file line number	Diff line number	Diff line change
`@@ -199,8 +199,6 @@ class CollectorConfig {`
`199`	`199`	`}`
`200`	`200`
`201`	`201`	`protected:`
`202`		`- FRIEND_TEST(SensorClientFormatterTest, NoProcessArguments);`
`203`		`-`
`204`	`202`	`int scrape_interval_;`
`205`	`203`	`CollectionMethod collection_method_;`
`206`	`204`	`bool turn_off_scrape_;`