Address PR comments

- Rollback no grpc channel behaviour.
- Add documentation to new classes.
- Remove unneeded workaround for old kernels.
- Minor refactorings in SensorClientFormatter.

Author: Molter73

collector/lib/CollectorOutput.cpp

@@ -7,9 +7,8 @@
 
 namespace collector {
 
-CollectorOutput::CollectorOutput(const CollectorConfig& config) {
-  use_sensor_client_ = !config.UseLegacyServices();
-
+CollectorOutput::CollectorOutput(const CollectorConfig& config)
+    : use_sensor_client_(!config.UseLegacyServices()) {
   if (config.grpc_channel != nullptr) {
     channel_ = config.grpc_channel;
 
@@ -22,7 +21,7 @@ CollectorOutput::CollectorOutput(const CollectorConfig& config) {
     }
   }
 
-  if (config.UseStdout()) {
+  if (config.grpc_channel == nullptr || config.UseStdout()) {
     if (use_sensor_client_) {
       auto sensor_client = std::make_unique<SensorClientStdout>();
       sensor_clients_.emplace_back(std::move(sensor_client));
@@ -32,10 +31,6 @@ CollectorOutput::CollectorOutput(const CollectorConfig& config) {
     }
   }
 
-  if (sensor_clients_.empty() && signal_clients_.empty()) {
-    CLOG(FATAL) << "No available output configured";
-  }
-
   thread_.Start([this] { EstablishGrpcStream(); });
 }
 
@@ -125,7 +120,7 @@ bool CollectorOutput::EstablishGrpcStreamSingle() {
   }
 
   // Refresh all clients
-  auto success = true;
+  bool success = true;
   for (const auto& client : signal_clients_) {
     success &= client->Refresh();
   }

collector/lib/CollectorOutput.h

@@ -38,8 +38,23 @@ class CollectorOutput {
     signal_clients_.emplace_back(std::move(signal_client));
   }
 
+  /**
+   * Send a message to sensor.
+   *
+   * @param msg One of sensor::MsgFromCollector or
+   *            sensor::SignalStreamMessage, the proper service to be
+   *            used will be determined from the type held in msg.
+   * @returns A SignalHandler::Result with the outcome of the send
+   *          operation
+   */
   SignalHandler::Result SendMsg(const MessageType& msg);
 
+  /**
+   * Whether we should use the new iservice or not.
+   *
+   * @returns true if configuration indicates we should use the new
+   *          iservice, false otherwise.
+   */
   bool UseSensorClient() const { return use_sensor_client_; }
 
  private:
@@ -53,10 +68,10 @@ class CollectorOutput {
   std::vector<std::unique_ptr<ISensorClient>> sensor_clients_;
   std::vector<std::unique_ptr<ISignalServiceClient>> signal_clients_;
 
-  bool use_sensor_client_{true};
+  bool use_sensor_client_ = true;
 
   StoppableThread thread_;
-  std::atomic<bool> stream_active_{false};
+  std::atomic<bool> stream_active_ = false;
   std::condition_variable stream_interrupted_;
   std::shared_ptr<grpc::Channel> channel_;
 };

collector/lib/ProtoSignalFormatter.h

@@ -30,7 +30,26 @@ class ProtoSignalFormatter : public BaseProtoSignalFormatter, protected ProtoAll
  public:
   ProtoSignalFormatter() = default;
 
+  /**
+   * Turn a sinsp_evt into a protobuf message to be sent to sensor.
+   *
+   * @param event The event to be translated.
+   * @returns A pointer to the message to be sent or nullptr if no
+   *          message should be sent. Ownership of the message is kept
+   *          in the implementing class, caller must not attempt to free
+   *          it.
+   */
   const Message* ToProtoMessage(sinsp_evt* event) override = 0;
+
+  /**
+   * Turn a sinsp_threadinfo into a protobuf message to be sent to sensor.
+   *
+   * @param event The event to be translated.
+   * @returns A pointer to the message to be sent or nullptr if no
+   *          message should be sent. Ownership of the message is kept
+   *          in the implementing class, caller must not attempt to free
+   *          it.
+   */
   const Message* ToProtoMessage(sinsp_threadinfo* tinfo) override = 0;
 };
 

collector/lib/SensorClient.h

@@ -26,8 +26,22 @@ class ISensorClient {
   ISensorClient& operator=(ISensorClient&&) = delete;
   virtual ~ISensorClient() = 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 Refresh() = 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::MsgFromCollector& msg) = 0;
 };
 

collector/lib/SensorClientFormatter.cpp

@@ -76,7 +76,7 @@ const sensor::MsgFromCollector* SensorClientFormatter::ToProtoMessage(sinsp_evt*
   Reset();
 
   if (!ValidateProcessDetails(event)) {
-    CLOG(INFO) << "Dropping process event: " << ProcessDetails(event);
+    CLOG(INFO) << "Dropping process event: " << ToString(event);
     return nullptr;
   }
 
@@ -178,10 +178,9 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_evt* event) {
   }
 
   // set process lineage
-  std::vector<LineageInfo> lineage;
-  this->GetProcessLineage(event->get_thread_info(), lineage);
+  auto lineage = GetProcessLineage(event->get_thread_info());
   for (const auto& p : lineage) {
-    auto signal_lineage = signal->add_lineage_info();
+    auto* signal_lineage = signal->add_lineage_info();
     signal_lineage->set_parent_exec_file_path(p.parent_exec_file_path());
     signal_lineage->set_parent_uid(p.parent_uid());
   }
@@ -241,11 +240,10 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_threadinfo* tinf
   signal->set_container_id(tinfo->m_container_id);
 
   // set process lineage
-  std::vector<LineageInfo> lineage;
-  GetProcessLineage(tinfo, lineage);
+  auto lineage = GetProcessLineage(tinfo);
 
   for (const auto& p : lineage) {
-    auto signal_lineage = signal->add_lineage_info();
+    auto* signal_lineage = signal->add_lineage_info();
     signal_lineage->set_parent_exec_file_path(p.parent_exec_file_path());
     signal_lineage->set_parent_uid(p.parent_uid());
   }
@@ -258,7 +256,7 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_threadinfo* tinf
   return signal;
 }
 
-std::string SensorClientFormatter::ProcessDetails(sinsp_evt* event) {
+std::string SensorClientFormatter::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);
@@ -293,37 +291,28 @@ bool SensorClientFormatter::ValidateProcessDetails(sinsp_evt* event) {
   return ValidateProcessDetails(tinfo);
 }
 
-int SensorClientFormatter::GetTotalStringLength(const std::vector<LineageInfo>& lineage) {
-  int totalStringLength = 0;
-  for (LineageInfo l : lineage) {
-    totalStringLength += l.parent_exec_file_path().size();
-  }
-
-  return totalStringLength;
-}
+void SensorClientFormatter::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();
+  });
 
-void SensorClientFormatter::CountLineage(const std::vector<LineageInfo>& lineage) {
-  int totalStringLength = GetTotalStringLength(lineage);
   COUNTER_INC(CollectorStats::process_lineage_counts);
   COUNTER_ADD(CollectorStats::process_lineage_total, lineage.size());
   COUNTER_ADD(CollectorStats::process_lineage_sqr_total, lineage.size() * lineage.size());
-  COUNTER_ADD(CollectorStats::process_lineage_string_total, totalStringLength);
+  COUNTER_ADD(CollectorStats::process_lineage_string_total, string_total);
 }
 
-void SensorClientFormatter::GetProcessLineage(sinsp_threadinfo* tinfo,
-                                              std::vector<LineageInfo>& lineage) {
+std::vector<LineageInfo> SensorClientFormatter::GetProcessLineage(sinsp_threadinfo* tinfo) {
+  std::vector<LineageInfo> lineage;
   if (tinfo == nullptr) {
-    return;
+    return lineage;
   }
-  sinsp_threadinfo* mt = nullptr;
-  if (tinfo->is_main_thread()) {
-    mt = tinfo;
-  } else {
-    mt = tinfo->get_main_thread();
-    if (mt == nullptr) {
-      return;
-    }
+
+  sinsp_threadinfo* mt = tinfo->get_main_thread();
+  if (mt == nullptr) {
+    return lineage;
   }
+
   sinsp_threadinfo::visitor_func_t visitor = [&lineage](sinsp_threadinfo* pt) {
     if (pt == nullptr) {
       return false;
@@ -332,22 +321,9 @@ void SensorClientFormatter::GetProcessLineage(sinsp_threadinfo* tinfo,
       return false;
     }
 
-    //
     // Collection of process lineage information should stop at the container
     // boundary to avoid collecting host process information.
-    //
-    // In back-ported eBPF probes, `m_vpid` will not be set for containers
-    // running when collector comes online because /proc/{pid}/status does
-    // not contain namespace information, so `m_container_id` is checked
-    // instead. `m_container_id` is not enough on its own to identify
-    // containerized processes, because it is not guaranteed to be set on
-    // all platforms.
-    //
-    if (pt->m_vpid == 0) {
-      if (pt->m_container_id.empty()) {
-        return false;
-      }
-    } else if (pt->m_pid == pt->m_vpid) {
+    if (pt->m_pid == pt->m_vpid) {
       return false;
     }
 
@@ -371,7 +347,9 @@ void SensorClientFormatter::GetProcessLineage(sinsp_threadinfo* tinfo,
     return true;
   };
   mt->traverse_parent_state(visitor);
-  CountLineage(lineage);
+  UpdateLineageStats(lineage);
+
+  return lineage;
 }
 
 }  // namespace collector

collector/lib/SensorClientFormatter.h

@@ -41,22 +41,68 @@ class SensorClientFormatter : public ProtoSignalFormatter<sensor::MsgFromCollect
   const MsgFromCollector* ToProtoMessage(sinsp_evt* event) override;
   const MsgFromCollector* ToProtoMessage(sinsp_threadinfo* tinfo) override;
 
-  void GetProcessLineage(sinsp_threadinfo* tinfo, std::vector<LineageInfo>& lineage);
+  /**
+   * Get the list of processes that spawned the current one.
+   *
+   * The list will be limited to processes inside the same container as
+   * the one being processed.
+   *
+   * @param tinfo The sinsp_threadinfo for which we should generate
+   *              lineage.
+   * @returns A vector with the lineage information.
+   */
+  static std::vector<LineageInfo> GetProcessLineage(sinsp_threadinfo* tinfo);
 
  private:
   FRIEND_TEST(SensorClientFormatterTest, NoProcessArguments);
   FRIEND_TEST(SensorClientFormatterTest, ProcessArguments);
 
+  /**
+   * Allocate and fill in a ProcessSignal message.
+   *
+   * @param event A Falco event to be translated into a ProcessSignal.
+   * @returns A non-owning pointer to the ProcessSignal.
+   */
   ProcessSignal* CreateProcessSignal(sinsp_evt* event);
-  Signal* CreateSignal(sinsp_evt* event);
+
+  /**
+   * Check if the provided threadinfo has the required fields.
+   *
+   * @param tinfo the Falco threadinfo to validate.
+   * @returns true if the threadinfo is valid, false otherwise.
+   */
   bool ValidateProcessDetails(const sinsp_threadinfo* tinfo);
+
+  /**
+   * Check if the provided event has the required fields.
+   *
+   * @param event the Falco event to validate.
+   * @returns true if the threadinfo is valid, false otherwise.
+   */
   bool ValidateProcessDetails(sinsp_evt* event);
-  std::string ProcessDetails(sinsp_evt* event);
 
-  Signal* CreateSignal(sinsp_threadinfo* tinfo);
+  /**
+   * Translate a Falco event to a printable string.
+   *
+   * @param event The Falco event to be translated.
+   * @returns A printable string.
+   */
+  std::string ToString(sinsp_evt* event);
+
+  /**
+   * Allocate and fill in a ProcessSignal message.
+   *
+   * @param tinfo A Falco threadinfo to be translated into a ProcessSignal.
+   * @returns A non-owning pointer to the ProcessSignal.
+   */
   ProcessSignal* CreateProcessSignal(sinsp_threadinfo* tinfo);
-  int GetTotalStringLength(const std::vector<LineageInfo>& lineage);
-  void CountLineage(const std::vector<LineageInfo>& lineage);
+
+  /**
+   * Update lineage related prometheus stats.
+   *
+   * @param lineage The lineage used for updating the stats.
+   */
+  static void UpdateLineageStats(const std::vector<LineageInfo>& lineage);
 
   const EventNames& event_names_;
   std::unique_ptr<system_inspector::EventExtractor> event_extractor_;

collector/lib/SignalServiceClient.h

@@ -61,7 +61,7 @@ class SignalServiceClient : public ISignalServiceClient {
   std::unique_ptr<grpc::ClientContext> context_;
   std::unique_ptr<IDuplexClientWriter<SignalStreamMessage>> writer_;
 
-  bool first_write_{};
+  bool first_write_ = false;
 };
 
 class StdoutSignalServiceClient : public ISignalServiceClient {