ROX-28530: use the new collector iservice for process information #14652

Molter73 · 2025-03-18T12:20:50Z

Description

This PR enables sensor to handle process signals coming from collector through the new internal service. In order to achieve this:

The existing process signal component is split into a separate service for handling legacy signals coming from older collectors and a standalone component that receives signals through channels.
A new service is introduced for handling messages coming through the new internal service. In its current state, only process signals are handled by forwarding them to the standalone component described in the previous bullet point, additional functionality will be added in other PRs to fully flesh out the feature with reviewable changes.

In addition to the previously described changes, a couple functions for translating between signal types have been added in order for the new type to be translated to the type central is expecting. In order to prevent this translation we could update the internal service between sensor and central to also use this new type, moving the translation into central itself before dumping to the DB, but this will require additional effort for compatibility between sensor and central which has not been discussed at the moment. Should this change be needed, it will be addressed in a follow up PR.

Sibling PR stackrox/collector#2063

User-facing documentation

CHANGELOG is updated OR update is not needed
documentation PR is created and is linked above OR is not needed

Testing and quality

the change is production ready: the change is GA or otherwise the functionality is gated by a feature flag
CI results are inspected

Automated testing

How I validated my change

Run collector and sensor with the new changes and checked the process information is available in the risk tab.

openshift-ci · 2025-03-18T12:20:54Z

Skipping CI for Draft Pull Request.
If you want CI signal for your change, please convert it to an actual PR.
You can still manually trigger a test run with /test all

rhacs-bot · 2025-03-18T12:52:47Z

Images are ready for the commit at 0bf1081.

To use with deploy scripts, first export MAIN_IMAGE_TAG=4.8.x-491-g0bf1081ee2.

codecov · 2025-03-18T13:27:22Z

Codecov Report

Attention: Patch coverage is 26.73267% with 74 lines in your changes missing coverage. Please review.

Project coverage is 48.94%. Comparing base (bae0d5d) to head (ceae0a4).
Report is 5 commits behind head on master.

Files with missing lines	Patch %	Lines
sensor/common/processindicator/component.go	27.00%	73 Missing ⚠️
sensor/kubernetes/fake/fake.go	0.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #14652      +/-   ##
==========================================
- Coverage   48.95%   48.94%   -0.01%     
==========================================
  Files        2550     2551       +1     
  Lines      187238   187338     +100     
==========================================
+ Hits        91662    91698      +36     
- Misses      88327    88392      +65     
+ Partials     7249     7248       -1

Flag	Coverage Δ
go-unit-tests	`48.94% <26.73%> (-0.01%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Molter73 · 2025-03-31T13:41:30Z

/test all

Molter73 · 2025-04-01T10:49:55Z

/test all

Molter73 · 2025-04-02T08:12:25Z

/retest

vikin91 · 2025-04-02T14:12:56Z

sensor/common/collector/service_impl.go

+func newService(queue chan *sensor.ProcessSignal) Service {
+	return &serviceImpl{
+		queue: queue,
+	}
+}


Anti-pattern waring - this object writes to the queue that is provided from the outside. This makes it unclear who and when should be closing this channel.

Solution: the queue should be created in the constructor of this object and returned as read-only. This object should then be responsible for closing the channel.

vikin91 · 2025-04-02T14:13:46Z

sensor/common/signal/component/component.go

+type Component interface {
+	common.SensorComponent
+
+	GetReceiver() chan *sensor.ProcessSignal


Let's always make it read-only for external actors.

Suggested change

GetReceiver() chan *sensor.ProcessSignal

GetReceiver() <-chan *sensor.ProcessSignal

vikin91 · 2025-04-02T14:14:51Z

sensor/common/signal/component/component.go

+	common.SensorComponent
+
+	processPipeline Pipeline
+	indicators      chan *message.ExpiringMessage


I can't see any writes to this channel here, so let's make it read only.

Suggested change

indicators chan *message.ExpiringMessage

indicators <-chan *message.ExpiringMessage

vikin91 · 2025-04-02T14:17:07Z

sensor/common/signal/component/component.go

+	cmp := &componentImpl{
+		processPipeline: pipeline,
+		indicators:      indicators,
+		receiver:        make(chan *sensor.ProcessSignal, maxBufferSize),


Why do we need the buffer here?
It looks like we are only reading from this channel, so I am wondering what is the purpose of this channel?

vikin91 · 2025-04-02T14:22:05Z

sensor/kubernetes/sensor/sensor.go

+
+	var signalSrv signalService.Service
+	if cfg.signalServiceAuthFuncOverride != nil && cfg.localSensor {
+		signalSrv = signalService.NewService(signalCmp.GetReceiver(),


Okay, so you create a channel in signalCmp only to pass it to another object so that it can write to it?
This is not a good idea, we must always make sure that:

There is exactly one writer of the channel. That writer is responsible for creating the channel and closing the channel. Writer returns the channel to the outside always as read-only channel.

There can be many readers of the channel and none of them should attempt to write to it or close it.

Writer and readers must be easy to identify.

For more details check https://www.oreilly.com/library/view/concurrency-in-go/9781491941294/ch04.html

As I explained offline, my initial approach was to have a single channel owned by the component and have it read messages coming from collector from both the old and new services. Thanks for taking the time to explain why this would be a dangerous practice in Go, I've made it so now the services each own a channel and those are passed in to the component for handling.

lvalerom

Overall looks good. I shared a few thoughts, let me know what you think

lvalerom · 2025-04-07T08:21:50Z

sensor/common/collector/service.go

+}
+
+func NewService(opts ...Option) Service {
+	return newService()


The Options need to be passed here and handle in newService. Otherwise WithAuthFuncOverride will be ignored and some integration tests will fail.

lvalerom · 2025-04-07T08:42:14Z

sensor/common/signal/component/component.go

+type componentImpl struct {
+	common.SensorComponent
+
+	processPipeline Pipeline
+	indicators      <-chan *message.ExpiringMessage
+	signalMessages  <-chan *storage.ProcessSignal
+	processMessages <-chan *sensor.ProcessSignal
+	writer          io.Writer
+
+	stopper concurrency.Stopper
+}


Would it make sense to merge this new component with the Pipeline? I feel like this new component only acts as a bridge between the service and the Pipeline component.

The whole code for processing processes (haha) is messy and confusing. From what I can tell, a long long time ago in a galaxy far away, signals were (or were going to be) the generic way for stackrox to handle events, so there are all these abstractions for signals by themselves, but the only signal that ever got implemented was for processes. And the reason I mention this is that we have the same situation in collector, there are a bunch of abstract classes (essentially interfaces) for signals, but they were only ever implemented for processes as well. I'm not a huge fan of this approach of "writing the abstraction first", IMO it leads to over complicated code that can very well end up not being used or used just once, like in our case.

AFAICT, this is the only implementation of the Pipeline interface in sensor/common/signal/component/pipeline.go (and gopls agrees with me):

stackrox/sensor/common/processsignal/pipeline.go

Line 32 in 858cda1

type Pipeline struct {

So, at this point, I'm not 100% sure how I would handle this. My instinct tells me to move the entire code under sensor/common/processsignal/ into sensor/common/signal and ditch the interface altogether, but that might be messy. Another possibility would be to move the service and component into sensor/common/processsignal/ and rename the directory to something like process, also dropping the interface. WDYT?

I'd go with the second one, move everything to something like sensor/common/process (maybe processindicator), remove the interface, and probably have only the SensorComponent as the handler for processes. So:

collectorService: to handle the connection with collector.

processIndicatorComponent: to handle the processIndicator enrichment.

lvalerom · 2025-04-07T08:44:14Z

sensor/common/signal/component/component.go

+	return &storage.ProcessSignal{
+		Id:           signal.Id,
+		ContainerId:  signal.ContainerId,
+		Time:         signal.CreationTime,
+		Name:         signal.Name,
+		Args:         signal.Args,
+		ExecFilePath: signal.ExecFilePath,
+		Pid:          signal.Pid,
+		Uid:          signal.Uid,
+		Gid:          signal.Gid,
+		Scraped:      signal.Scraped,
+		LineageInfo:  lineage,
+	}


Probably out of the scope for this PR and maybe this entire effort but I would try to move away from using storage protos in the API as a general rule.

I agree, I would argue this falls under this issue that is outside of the scope of the current epic.
https://issues.redhat.com/browse/ROX-28532

We'll try to get that prioritized and implemented in the near future, I have to also do some cleanups in the near future and I can probably get this done then.

lvalerom · 2025-04-07T09:28:47Z

sensor/common/collector/service_impl.go

+
+func newService() Service {
+	return &serviceImpl{
+		queue: make(chan *sensor.ProcessSignal, maxBufferSize),


We should add metrics to this buffer. Also, consider using ScaleSizeOnNonDefault and an environment variable to configure this max size.

lvalerom · 2025-04-07T09:32:47Z

sensor/common/signal/singleton.go

-		queue:            make(chan *v1.Signal, maxBufferSize),
-		indicators:       indicators,
-		processPipeline:  pipeline,
+		queue:            make(chan *storage.ProcessSignal, maxBufferSize),


Not 100% sure why we had this queue: make(chan *v1.Signal, maxBufferSize) before because I believe it was not used 🤷 but since now queue: make(chan *storage.ProcessSignal, maxBufferSize) is used, I'd add metrics and make is configurable with ScaleSizeOnNonDefault + and env variable.

Asking for both this and the serivice_impl.go comment, could you provide an example of the metrics you'd like added? I'm not 100% sure how to look for those in the sensor code.

Sure! Here's how we do it for the BufferedStream. Basically we track how many items were added, removed, and dropped from the buffer.

- Gracefully stop services. - Remove unused indicators member. - Rename environment variable.

openshift-ci · 2025-04-11T14:30:28Z

@Molter73: The following test failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
ci/prow/ocp-4-18-qa-e2e-tests	`ceae0a4`	link	false	`/test ocp-4-18-qa-e2e-tests`

Full PR test history. Your PR dashboard.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.

JoukoVirtanen · 2025-04-22T01:27:42Z

sensor/common/processindicator/component.go

+	c.processPipeline.Process(signal)
+}
+
+// TODO(ROX-3281) this is a workaround for these collector issues


This ticket was closed years ago. I guess this was copied from signal/signal_service.go. You should delete this comment and its duplicate.

JoukoVirtanen · 2025-04-22T01:30:17Z

sensor/common/processindicator/component.go

+	return true
+}
+
+func sensorIntoStorageSignal(signal *sensor.ProcessSignal) *storage.ProcessSignal {


Suggested change

func sensorIntoStorageSignal(signal *sensor.ProcessSignal) *storage.ProcessSignal {

func convertSensorProcessSignalToStorageProcessSignal(signal *sensor.ProcessSignal) *storage.ProcessSignal {

That naming is way too verbose, I will not make this change.

vikin91

Posting my detailed comments from reading the code.
I still plan to look at the tests with local-sensor.

vikin91 · 2025-04-22T11:47:51Z

sensor/common/collector/service_impl.go

+		msg, err := server.Recv()
+		if err != nil {
+			log.Error("error dequeueing collector message, event: ", err)
+			return err
+		}


Why not placing this block at the beginning of default?

vikin91 · 2025-04-22T11:48:47Z

sensor/common/collector/service_impl.go

+			case *sensor.MsgFromCollector_Register:
+				log.Infof("got register: %+v", msg.GetRegister())
+			case *sensor.MsgFromCollector_Info:
+				log.Infof("got network info: %+v", msg.GetInfo())


Looks like debugging output. Let's remove the log statements or change them to debug level.

Changed them to debug, a follow up PR will implement the actual logic that needs to go here.

vikin91 · 2025-04-22T11:51:06Z

sensor/common/collector/service_impl.go

+			metrics.CollectorChannelInc(msg)
+			switch msg.GetMsg().(type) {
+			case *sensor.MsgFromCollector_ProcessSignal:
+				s.queue <- msg.GetProcessSignal()


This may block, so we need another switch here:

Suggested change

s.queue <- msg.GetProcessSignal()

select {

case s.queue <- msg.GetProcessSignal():

case case <-server.Context().Done():

return nil

}

vikin91 · 2025-04-22T11:51:54Z

sensor/common/collector/service_impl.go

+		select {
+		case <-server.Context().Done():
+			return nil
+		default:


The body of this default block may look better in a separate function (provided so many nested selects).

vikin91 · 2025-04-22T11:53:16Z

sensor/common/metrics/metrics.go

+	collectorChannelMessagesCount = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace: metrics.PrometheusNamespace,
+		Subsystem: metrics.SensorSubsystem.String(),
+		Name:      "num_messages_received_collector",


Please check the naming convention for prometheus metrics. The counter should not use num but rather total as postfix.

vikin91 · 2025-04-22T14:38:46Z

sensor/common/processindicator/component.go

+}
+
+func (c *componentImpl) Notify(e common.SensorComponentEvent) {
+	log.Info(common.LogSensorComponentEvent(e))


LogSensorComponentEvent accepts a component name as a second parameter - let's use it as it may help in some exotic debugging sessions.

vikin91 · 2025-04-22T14:42:28Z

sensor/common/processindicator/component.go

+	if c.writer != nil {
+		if data, err := signal.MarshalVT(); err == nil {
+			if _, err := c.writer.Write(data); err != nil {
+				log.Warnf("Error writing msg: %v", err)


Let's make this a bit more informative

Suggested change

log.Warnf("Error writing msg: %v", err)

log.Warnf("Error writing ProcessSignal data: %v", err)

also, we could return those errors from this func (instead of only logging) and make this code easier to test.

Do we want to return the errors? It seems these are problems with the writer which is only used for testing AFAICT. Returning the error would prevent the process signal from properly being processed.

vikin91 · 2025-04-22T14:46:09Z

sensor/common/processindicator/component.go

+// TODO(ROX-3281) this is a workaround for these collector issues
+func isProcessSignalValid(signal *storage.ProcessSignal) bool {
+	// Example: <NA> or sometimes a truncated variant
+	if signal.GetExecFilePath() == "" || signal.GetExecFilePath()[0] == '<' {


Although accessing the 0-th element should be safe here, it looks like a case for strings.HasPrefix

vikin91 · 2025-04-22T14:55:30Z

sensor/kubernetes/sensor/sensor.go

-	processPipeline := processsignal.NewProcessPipeline(indicators, storeProvider.Entities(), processfilter.Singleton(), policyDetector)
-	var processSignals signalService.Service
+	var signalSrv signalService.Service
+	if cfg.signalServiceAuthFuncOverride != nil && cfg.localSensor {


It looks like this if is duplicated in line 123 - can we maybe merge those two?

vikin91 · 2025-04-22T15:01:03Z

sensor/kubernetes/sensor/sensor.go

 	}
+
+	// Create Process Pipeline
+	indicators := make(chan *message.ExpiringMessage, queue.ScaleSizeOnNonDefault(env.ProcessIndicatorBufferSize))


This channel is used only in line 131 - let's create it inside the NewProcessPipeline

lvalerom · 2025-04-23T07:54:16Z

sensor/common/processindicator/component.go

+
+func (c *componentImpl) Stop(_ error) {
+	c.processPipeline.Shutdown()
+	c.stopper.Client().Stop()


Should we wait for the component to stop?

Not 100% sure how to go about this one, I think I just copy pasted the code from the existing component. Could you elaborate a bit?

lvalerom · 2025-04-23T08:08:49Z

sensor/common/signal/signal_service.go

-			// Currently eat unhandled signals
-			continue
+			metrics.IncrementTotalProcessesReceivedCounter()
+			s.queue <- signal


This could block. It should be in a select statement.

lvalerom · 2025-04-23T08:12:44Z

sensor/common/signal/signal_service.go

-			continue
+		signal, err := unwrapSignal(signalStreamMsg)
+		if err != nil {
+			metrics.IncrementTotalProcessesDroppedCounter()


🤔 I'm not sure about this metric. Normally these SomethingDroppedCounter are used to record when we drop a message not because it's not valid but because the buffers are full.

I think this should be used when we are pushing messages to the queue:

select { case s.queue <- signal: metrics.IncrementTotalProcessesReceivedCounter() default: metrics.IncrementTotalProcessesDroppedCounter() }

- Unify process metrics - Refactor Communicate method - Minor cleanups - Prevent some locks when writing to channels

openshift-ci bot added the do-not-merge/work-in-progress label Mar 18, 2025

github-actions bot added the area/sensor label Mar 18, 2025

github-actions bot added the ci-all-qa-tests Tells CI to run all API tests (not just BAT). label Mar 31, 2025

Molter73 force-pushed the mauro/ROX-28530-use-collector-iservice branch 2 times, most recently from c4eaff1 to 627fe7b Compare March 31, 2025 15:35

Molter73 changed the title ~~[WIP] Use the new collector internal service~~ feat: use the new collector internal service Apr 1, 2025

Molter73 marked this pull request as ready for review April 1, 2025 16:13

Molter73 requested a review from a team as a code owner April 1, 2025 16:13

openshift-ci bot removed the do-not-merge/work-in-progress label Apr 1, 2025

Molter73 mentioned this pull request Apr 1, 2025

Use the new collector internal service for process information stackrox/collector#2063

Open

5 tasks

Molter73 changed the title ~~feat: use the new collector internal service~~ feat: use the new collector internal service for process information Apr 1, 2025

vikin91 reviewed Apr 2, 2025

View reviewed changes

Molter73 changed the title ~~feat: use the new collector internal service for process information~~ ROX-28530: use the new collector internal service for process information Apr 2, 2025

Molter73 changed the title ~~ROX-28530: use the new collector internal service for process information~~ ROX-28530: use the new collector iservice for process information Apr 2, 2025

lvalerom reviewed Apr 7, 2025

View reviewed changes

Molter73 force-pushed the mauro/ROX-28530-use-collector-iservice branch from 2c0619b to 291f3a2 Compare April 7, 2025 15:40

Molter73 requested review from vikin91 and lvalerom April 7, 2025 16:23

Molter73 force-pushed the mauro/ROX-28530-use-collector-iservice branch 2 times, most recently from ffa406e to 332ef39 Compare April 8, 2025 11:04

Molter73 added 20 commits April 11, 2025 12:42

Fix style issues

d9ecf16

Bump collector version

0cb4185

Fix some tests that rely on a collector logs message

f3d635f

Add unit tests for sensor to storage signal translators

4814dff

Add authFuncOverride to collector service

ce2ad20

Fix style errors

91c71af

Improve unit test coverage

212f6d2

Remove lineage conversion functions

0feab64

Invert channel ownership for signal component and services

3a3e42a

Remove unused function

bb5c8e3

Fix CI lint error

44ccda1

Use errors.New instead of fmt.Errorf

134c2fe

Fix opts handling in collector iservice and autoscale queues

c16664a

Remove unused const

e5584d8

Move signal/component to processsignal

84cbcf8

Add metrics

8b5042f

Minor cleanups

a1153f8

- Gracefully stop services. - Remove unused indicators member. - Rename environment variable.

Fix sensor integration test

472291b

test(sensor-integration): use the new collector iservice for testing

b893bdf

Remove close statements

ceae0a4

Molter73 force-pushed the mauro/ROX-28530-use-collector-iservice branch from d461686 to ceae0a4 Compare April 11, 2025 10:43

JoukoVirtanen reviewed Apr 22, 2025

View reviewed changes

vikin91 reviewed Apr 22, 2025

View reviewed changes

lvalerom reviewed Apr 23, 2025

View reviewed changes

Address PR comments

0bf1081

- Unify process metrics - Refactor Communicate method - Minor cleanups - Prevent some locks when writing to channels

Molter73 requested review from vikin91, JoukoVirtanen and lvalerom April 23, 2025 10:22

	GetReceiver() chan *sensor.ProcessSignal
	GetReceiver() <-chan *sensor.ProcessSignal

	indicators chan *message.ExpiringMessage
	indicators <-chan *message.ExpiringMessage

	func sensorIntoStorageSignal(signal sensor.ProcessSignal) storage.ProcessSignal {
	func convertSensorProcessSignalToStorageProcessSignal(signal sensor.ProcessSignal) storage.ProcessSignal {

	log.Warnf("Error writing msg: %v", err)
	log.Warnf("Error writing ProcessSignal data: %v", err)

ROX-28530: use the new collector iservice for process information #14652

Are you sure you want to change the base?

ROX-28530: use the new collector iservice for process information #14652

Conversation

Molter73 commented Mar 18, 2025 • edited Loading

Description

User-facing documentation

Testing and quality

Automated testing

How I validated my change

openshift-ci bot commented Mar 18, 2025

rhacs-bot commented Mar 18, 2025 • edited Loading

codecov bot commented Mar 18, 2025 • edited Loading

Codecov Report

Molter73 commented Mar 31, 2025

Molter73 commented Apr 1, 2025

Molter73 commented Apr 2, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvalerom left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lvalerom Apr 8, 2025 • edited Loading

Choose a reason for hiding this comment

openshift-ci bot commented Apr 11, 2025

JoukoVirtanen Apr 22, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vikin91 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Molter73 commented Mar 18, 2025 •

edited

Loading

rhacs-bot commented Mar 18, 2025 •

edited

Loading

codecov bot commented Mar 18, 2025 •

edited

Loading

lvalerom Apr 8, 2025 •

edited

Loading

JoukoVirtanen Apr 22, 2025 •

edited

Loading