ROX-28981: Split methods for collector iservice

Due to a limitation on collector side with the C++ gRPC API, we have
decided to split the RPC method into one per message type collector can
send.

This patch changes and regenerates the protobuf definitions and reworks
the collector service to comply with the new definition.

Author: Molter73

collector/lib/CollectorOutput.cpp

@@ -1,7 +1,5 @@
 #include "CollectorOutput.h"
 
-#include "internalapi/sensor/collector_iservice.pb.h"
-
 #include "GRPCUtil.h"
 #include "HostInfo.h"
 
@@ -40,7 +38,7 @@ void CollectorOutput::HandleOutputError() {
   stream_interrupted_.notify_one();
 }
 
-SignalHandler::Result CollectorOutput::SensorOutput(const sensor::MsgFromCollector& msg) {
+SignalHandler::Result CollectorOutput::SensorOutput(const sensor::ProcessSignal& msg) {
   for (auto& client : sensor_clients_) {
     auto res = client->SendMsg(msg);
     switch (res) {
@@ -83,7 +81,7 @@ SignalHandler::Result CollectorOutput::SignalOutput(const sensor::SignalStreamMe
 SignalHandler::Result CollectorOutput::SendMsg(const MessageType& msg) {
   auto visitor = [this](auto&& m) {
     using T = std::decay_t<decltype(m)>;
-    if constexpr (std::is_same_v<T, sensor::MsgFromCollector>) {
+    if constexpr (std::is_same_v<T, sensor::ProcessSignal>) {
       return SensorOutput(m);
     } else if constexpr (std::is_same_v<T, sensor::SignalStreamMessage>) {
       return SignalOutput(m);

collector/lib/CollectorOutput.h

@@ -2,7 +2,6 @@
 
 #include <variant>
 
-#include "internalapi/sensor/collector_iservice.pb.h"
 #include "internalapi/sensor/signal_iservice.pb.h"
 
 #include "CollectorConfig.h"
@@ -13,7 +12,7 @@
 
 namespace collector {
 
-using MessageType = std::variant<sensor::MsgFromCollector, sensor::SignalStreamMessage>;
+using MessageType = std::variant<sensor::ProcessSignal, sensor::SignalStreamMessage>;
 
 class CollectorOutput {
  public:
@@ -62,7 +61,7 @@ class CollectorOutput {
   bool EstablishGrpcStreamSingle();
 
   void HandleOutputError();
-  SignalHandler::Result SensorOutput(const sensor::MsgFromCollector& msg);
+  SignalHandler::Result SensorOutput(const sensor::ProcessSignal& msg);
   SignalHandler::Result SignalOutput(const sensor::SignalStreamMessage& msg);
 
   std::vector<std::unique_ptr<ISensorClient>> sensor_clients_;

collector/lib/ProcessSignalHandler.cpp

@@ -104,9 +104,9 @@ SignalHandler::Result ProcessSignalHandler::HandleSensorSignal(sinsp_evt* evt) {
     return IGNORED;
   }
 
-  dtrace_probe(signal_msg->process_signal());
+  dtrace_probe(*signal_msg);
 
-  if (!rate_limiter_.Allow(compute_process_key(signal_msg->process_signal()))) {
+  if (!rate_limiter_.Allow(compute_process_key(*signal_msg))) {
     ++(stats_->nProcessRateLimitCount);
     return IGNORED;
   }
@@ -128,7 +128,7 @@ SignalHandler::Result ProcessSignalHandler::HandleExistingProcessSensor(sinsp_th
     return IGNORED;
   }
 
-  if (!rate_limiter_.Allow(compute_process_key(signal_msg->process_signal()))) {
+  if (!rate_limiter_.Allow(compute_process_key(*signal_msg))) {
     ++(stats_->nProcessRateLimitCount);
     return IGNORED;
   }

collector/lib/SensorClient.cpp

@@ -5,7 +5,7 @@
 namespace collector {
 bool SensorClient::Recreate() {
   context_ = std::make_unique<grpc::ClientContext>();
-  writer_ = DuplexClient::CreateWithReadsIgnored(&sensor::CollectorService::Stub::AsyncCommunicate, channel_, context_.get());
+  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 ...";
     CLOG(ERROR) << "Error message: " << writer_->FinishNow().error_message();
@@ -18,14 +18,14 @@ bool SensorClient::Recreate() {
   return true;
 }
 
-SignalHandler::Result SensorClient::SendMsg(const sensor::MsgFromCollector& msg) {
+SignalHandler::Result SensorClient::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";
     return SignalHandler::ERROR;
   }
 
-  if (first_write_ && msg.has_process_signal()) {
+  if (first_write_) {
     first_write_ = false;
     return SignalHandler::NEEDS_REFRESH;
   }

collector/lib/SensorClient.h

@@ -5,7 +5,6 @@
 #include <grpcpp/channel.h>
 
 #include "internalapi/sensor/collector_iservice.grpc.pb.h"
-#include "internalapi/sensor/collector_iservice.pb.h"
 
 #include "DuplexGRPC.h"
 #include "SignalHandler.h"
@@ -39,7 +38,7 @@ class ISensorClient {
    * @returns A SignalHandler::Result with the outcome of the send
    *          operation.
    */
-  virtual SignalHandler::Result SendMsg(const sensor::MsgFromCollector& msg) = 0;
+  virtual SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) = 0;
 };
 
 class SensorClient : public ISensorClient {
@@ -60,7 +59,7 @@ class SensorClient : public ISensorClient {
 
   bool Recreate() override;
 
-  SignalHandler::Result SendMsg(const sensor::MsgFromCollector& msg) override;
+  SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) override;
 
  private:
   std::shared_ptr<grpc::Channel> channel_;
@@ -69,15 +68,15 @@ class SensorClient : public ISensorClient {
 
   // This needs to have the same lifetime as the class.
   std::unique_ptr<grpc::ClientContext> context_;
-  std::unique_ptr<IDuplexClientWriter<sensor::MsgFromCollector>> writer_;
+  std::unique_ptr<IDuplexClientWriter<sensor::ProcessSignal>> writer_;
 
   bool first_write_ = false;
 };
 
 class SensorClientStdout : public ISensorClient {
   bool Recreate() override { return true; }
 
-  SignalHandler::Result SendMsg(const sensor::MsgFromCollector& msg) override {
+  SignalHandler::Result SendMsg(const sensor::ProcessSignal& msg) override {
     LogProtobufMessage(msg);
     return SignalHandler::PROCESSED;
   }

collector/lib/SensorClientFormatter.cpp

@@ -4,7 +4,7 @@
 
 #include <google/protobuf/util/time_util.h>
 
-#include "internalapi/sensor/collector_iservice.pb.h"
+#include "internalapi/sensor/collector.pb.h"
 #include "internalapi/sensor/signal_iservice.pb.h"
 
 #include "CollectorStats.h"
@@ -67,7 +67,7 @@ SensorClientFormatter::SensorClientFormatter(sinsp* inspector, const CollectorCo
 
 SensorClientFormatter::~SensorClientFormatter() = default;
 
-const sensor::MsgFromCollector* SensorClientFormatter::ToProtoMessage(sinsp_evt* event) {
+const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_evt* event) {
   if (process_signals[event->get_type()] == ProcessSignalType::UNKNOWN_PROCESS_TYPE) {
     return nullptr;
   }
@@ -79,39 +79,21 @@ const sensor::MsgFromCollector* SensorClientFormatter::ToProtoMessage(sinsp_evt*
     return nullptr;
   }
 
-  ProcessSignal* process_signal = CreateProcessSignal(event);
-  if (process_signal == nullptr) {
-    return nullptr;
-  }
-
-  auto* msg = AllocateRoot();
-  msg->clear_info();
-  msg->clear_register_();
-  msg->set_allocated_process_signal(process_signal);
-  return msg;
+  return CreateProcessSignal(event);
 }
 
-const sensor::MsgFromCollector* SensorClientFormatter::ToProtoMessage(sinsp_threadinfo* tinfo) {
+const sensor::ProcessSignal* SensorClientFormatter::ToProtoMessage(sinsp_threadinfo* tinfo) {
   Reset();
   if (!ValidateProcessDetails(tinfo)) {
     CLOG(INFO) << "Dropping process event: " << tinfo;
     return nullptr;
   }
 
-  ProcessSignal* signal = CreateProcessSignal(tinfo);
-  if (signal == nullptr) {
-    return nullptr;
-  }
-
-  auto* msg = AllocateRoot();
-  msg->clear_register_();
-  msg->clear_info();
-  msg->set_allocated_process_signal(signal);
-  return msg;
+  return CreateProcessSignal(tinfo);
 }
 
 ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_evt* event) {
-  auto signal = Allocate<ProcessSignal>();
+  auto signal = AllocateRoot();
 
   // set id
   signal->set_id(UUIDStr());
@@ -193,7 +175,7 @@ ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_evt* event) {
 }
 
 ProcessSignal* SensorClientFormatter::CreateProcessSignal(sinsp_threadinfo* tinfo) {
-  auto signal = Allocate<ProcessSignal>();
+  auto signal = AllocateRoot();
 
   // set id
   signal->set_id(UUIDStr());

collector/lib/SensorClientFormatter.h

@@ -5,7 +5,6 @@
 #include <gtest/gtest_prod.h>
 
 #include "api/v1/signal.pb.h"
-#include "internalapi/sensor/collector_iservice.pb.h"
 
 #include "CollectorConfig.h"
 #include "ContainerMetadata.h"
@@ -22,7 +21,7 @@ class EventExtractor;
 
 namespace collector {
 
-class SensorClientFormatter : public ProtoSignalFormatter<sensor::MsgFromCollector> {
+class SensorClientFormatter : public ProtoSignalFormatter<sensor::ProcessSignal> {
  public:
   SensorClientFormatter(const SensorClientFormatter&) = delete;
   SensorClientFormatter(SensorClientFormatter&&) = delete;
@@ -35,10 +34,9 @@ class SensorClientFormatter : public ProtoSignalFormatter<sensor::MsgFromCollect
   using Signal = v1::Signal;
   using ProcessSignal = sensor::ProcessSignal;
   using LineageInfo = sensor::ProcessSignal_LineageInfo;
-  using MsgFromCollector = sensor::MsgFromCollector;
 
-  const MsgFromCollector* ToProtoMessage(sinsp_evt* event) override;
-  const MsgFromCollector* ToProtoMessage(sinsp_threadinfo* tinfo) override;
+  const ProcessSignal* ToProtoMessage(sinsp_evt* event) override;
+  const ProcessSignal* ToProtoMessage(sinsp_threadinfo* tinfo) override;
 
   /**
    * Get the list of processes that spawned the current one.

collector/proto/third_party/stackrox

@@ -1,8 +1,6 @@
 #include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
-#include "internalapi/sensor/collector_iservice.pb.h"
-
 #include "CollectorOutput.h"
 #include "SensorClient.h"
 #include "SignalServiceClient.h"
@@ -11,7 +9,7 @@ namespace collector {
 class MockSensorClient : public ISensorClient {
  public:
   MOCK_METHOD(bool, Recreate, ());
-  MOCK_METHOD(SignalHandler::Result, SendMsg, (const sensor::MsgFromCollector&));
+  MOCK_METHOD(SignalHandler::Result, SendMsg, (const sensor::ProcessSignal&));
 };
 
 class MockSignalClient : public ISignalServiceClient {
@@ -32,7 +30,7 @@ class CollectorOutputTest : public testing::Test {
 };
 
 TEST_F(CollectorOutputTest, SensorClient) {
-  sensor::MsgFromCollector msg;
+  sensor::ProcessSignal msg;
 
   EXPECT_CALL(*sensor_client, SendMsg).Times(1).WillOnce(testing::Return(SignalHandler::PROCESSED));
 

collector/test/CollectorOutputTest.cpp

@@ -16,10 +16,10 @@ require (
 	github.com/stackrox/rox v0.0.0-20210914215712-9ac265932e28
 	github.com/stretchr/testify v1.10.0
 	github.com/thoas/go-funk v0.9.3
-	go.opentelemetry.io/otel/trace v1.34.0
+	go.opentelemetry.io/otel/trace v1.35.0
 	golang.org/x/exp v0.0.0-20241217172543-b2144cdd0a67
-	golang.org/x/sys v0.31.0
-	google.golang.org/grpc v1.71.0
+	golang.org/x/sys v0.32.0
+	google.golang.org/grpc v1.71.1
 	gopkg.in/yaml.v3 v3.0.1
 	k8s.io/api v0.32.3
 	k8s.io/apimachinery v0.32.3
@@ -46,7 +46,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
 	github.com/go-openapi/jsonreference v0.21.0 // indirect
-	github.com/go-openapi/swag v0.23.0 // indirect
+	github.com/go-openapi/swag v0.23.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/gonum/blas v0.0.0-20181208220705-f22b278b28ac // indirect
@@ -67,7 +67,6 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
-	github.com/klauspost/compress v1.18.0 // indirect
 	github.com/mailru/easyjson v0.9.0 // indirect
 	github.com/mitchellh/go-wordwrap v1.0.1 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
@@ -81,7 +80,7 @@ require (
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240409071808-615f978279ca // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_golang v1.21.1 // indirect
+	github.com/prometheus/client_golang v1.22.0 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
@@ -93,24 +92,24 @@ require (
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.59.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.59.0 // indirect
-	go.opentelemetry.io/otel v1.34.0 // indirect
+	go.opentelemetry.io/otel v1.35.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.32.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.32.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.28.0 // indirect
-	go.opentelemetry.io/otel/metric v1.34.0 // indirect
-	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/sdk v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
-	golang.org/x/net v0.37.0 // indirect
-	golang.org/x/oauth2 v0.28.0 // indirect
-	golang.org/x/term v0.30.0 // indirect
-	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/net v0.39.0 // indirect
+	golang.org/x/oauth2 v0.29.0 // indirect
+	golang.org/x/term v0.31.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
 	golang.org/x/time v0.11.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
-	google.golang.org/protobuf v1.36.5 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
@@ -129,7 +128,7 @@ replace (
 	github.com/fullsailor/pkcs7 => github.com/stackrox/pkcs7 v0.0.0-20220914154527-cfdb0aa47179
 	github.com/heroku/docker-registry-client => github.com/stackrox/docker-registry-client v0.0.0-20230714151239-78b1f5f70b8a
 	github.com/operator-framework/helm-operator-plugins => github.com/stackrox/helm-operator v0.0.10-0.20220919093109-89f9785764c6
-	github.com/stackrox/rox => github.com/stackrox/stackrox v0.0.0-20250325114232-8e68c6a26ed1
+	github.com/stackrox/rox => github.com/stackrox/stackrox v0.0.0-20250415091607-159cf2aea233
 	go.uber.org/zap => github.com/stackrox/zap v1.18.2-0.20240314134248-5f932edd0404
 
 )