I wonder why this has been put in to the default case in the first place 🤔 Having it kept here, the code would have been called 1 or more times. Now we would call the initialization only once. Not sure whether this is a problem, but we should keep an eye on the stability of the stream.

For example: would we be able to reestablish the connection if we crash sensor while compliance is still working? That is something to check.

Yes, the crash of sensor causes the connection to break and there is no self-healing. We need to fix that by reverting the change and merging the two loops by adding a new case to the for-switch loop.

It seems to me, the code before the change will shut down Compliance when the connection to Sensor is lost.
Piotr, in your experiment did you check behavior of new or of unchanged code?

I checked the behavior from after the change that I commented on. I also saw the fatal message, but I couldn't see compliance crashing after the sensor was gone. I guess we would need to check this on the original code from before this PR.

I have reverted to the original code and also implemented a similar mechanic for the NodeScan loop.
I tested this by killing the sensor pod while compliance is running and was able to verify that both loops connect to (the newly created) sensor again.

I don't think there's a need to wrap error another time (see the calling code)

time.Hour * 4 is very specific for node rescan interval, therefore this function cannot be presented as generic "util". I think you should simply define it as GetNodeRescanInterval in pkg/env/node_scan.go. Then, instead of doing this

rescanInterval := complianceUtils.VerifyAndUpdateDuration(env.NodeRescanInterval.DurationSetting())

you can simply do this

rescanInterval := env.GetNodeRescanInterval()

I don't think we need to worry about this because DurationSetting() already handles this:

// DurationSetting returns the Duration object represented by the environment variable
func (d *DurationSetting) DurationSetting() time.Duration {
	val := os.Getenv(d.envVar)
	if val != "" {
		dur, err := time.ParseDuration(val)
		if err == nil {
			return dur
		}
		log.Warnf("%s is not a valid environment variable for %s, using default value: %s", val, d.envVar, d.defaultDuration.String())
	}
	return d.defaultDuration
}

That being said, in another PR, we should probably add a panic in the registerDurationSetting if the default is not valid

#3558

ah actually time.ParseDuration does allow for negative durations, so DurationSetting() does not handle this. Perhaps we should include that in the PR above ^

Thanks for linking the PR! We'll keep an eye on it and update our code as soon as your PR is merged.
For now, we want to get this PR merged rather quick, though.

@vikin91 @Maddosaurus
Unless I read the code incorrectly, the original manageStream function does not function the way it supposed to.
I think, its idea should be to check whether the connection got closed (ctx.Done()) and in this case shut down the process (sig.Signal()), or (if the connection is still alive) receive a message from Sensor and respond back (runRecv()). The problem is that runRecv() is a blocking function that runs an infinite loop of receive-reply logic. Once the code gets to it, it will be running forever until an error happens or unsupported message arrives.
When the code is in runRecv(), the shutdown logic of manageStream won't be executed normally.
Am I right or am I missing something?

Maybe we should check other places like Sensor or Central for how shutdown-receive-send logic is organized there and copy that logic here. To me it seems that we're having difficulties plugging in scanner logic just because the existing approach in Compliance is not prepared for extensibility.

I definitely want to look at sensor and see how it is done there.

For compliance, I see that client.Recv() is being called on a client that was created with the context client, err := cli.Communicate(ctx), so I hope that canceling the context would shutdown the client and release the lock.
(I used word "hope" as I have not verified this and do not know the internals of grpc stream).

I will do some more reading on this.

So sensor utilizes two clever objects to manage the stream to central: a pointer *grpcUtil.LazyClientConn holding a lazy connection, and centralclient.CentralConnectionFactory that sets, monitors the connection and restarts it when failed. All communication attempts are blocking and waiting when the connection is down (no message is lost). The operations resume when the connection is back up.

We could apply this to compliance-sensor as well, but this may deserve a separate refactoring PR. Would it be okay to continue in the current shape and improve this in a separate PR @msugakov ?

The code has changed now a lot. Issues that were before are now gone but new things still don't look 100% self-explanatory to me. I think we (I?) need to confidently understand well what we do. I would not classify "understanding" as "improvement" or "optimization". "Understanding" is our core skill.

Yes, sorry for adding so many changes after you started the review. I think they were needed to address some obvious drawbacks regarding sending without being sure that the connection is up.

While there are still some bits about the GRPC stream that I would like to understand better, I feel pretty confident that the code handling sending and receiving in this component follows the same principles as in other parts of our code.

This could be an else instead of the continue

Since we only care about RHCOS at this time, I'd leave this empty/nil

Not very important, but maybe use 8 instead of 7 due to RHCOS

rhel:8 here as well

let's ignore executables as well since this is for Active Vuln mgmt, and I don't think we plan on having that in the first iteration

I don't think we need to worry about this because DurationSetting() already handles this:

// DurationSetting returns the Duration object represented by the environment variable
func (d *DurationSetting) DurationSetting() time.Duration {
	val := os.Getenv(d.envVar)
	if val != "" {
		dur, err := time.ParseDuration(val)
		if err == nil {
			return dur
		}
		log.Warnf("%s is not a valid environment variable for %s, using default value: %s", val, d.envVar, d.defaultDuration.String())
	}
	return d.defaultDuration
}

That being said, in another PR, we should probably add a panic in the registerDurationSetting if the default is not valid

#3558

I wonder if we should throw this inside of manageStream so we can just reuse the same stream and use the same stop signals

We tried this and this is not trivial, because the runRecv blocks. I have a variation on this here #3548 but it also uses two loops - one for sending and one for receiving.

In a35000a, we applied the idea from #3548 and now, we have separate loops for sending and receiving

I read up about gRPC yesterday and got a bit more dangerous ;-) although still not very educated.

And I'm definitely here to spoil the party that started (even though I haven't used the opportunity to do that earlier). It seems this code will open a stream each time there's something to send. I think gRPC maintains a pool of connections on the client, plus streams get multiplexed over HTTP/2 (but only when that's used), so opening a stream doesn't mean establishing new connection, but I think it still has some overhead.

What is worse, I believe that we reuse the same rpc Communicate as the ordinary Compliance does. An effect of that might be that Sensor will push its messages over that stream too, but on Compliance side there's nothing receiving and acting on them for this stream. In case Sensor pushes a message to perform Compliance checks (message TriggerRun) through this stream, it will get lost.

From looking at Sensor code, it seems to me that more recent connection (actually stream) replaces previous in the map keyed by a node name. Therefore, I strongly suspect that what I described actually happens if the feature flag for node scans is turned on, i.e. compliance checks stop working properly.

I don't think it's an issue for now as the feature is gated but I think the logic should be reworked before we are ready to release.

WDYT?

That is a very good find. We should definitely improve this before releasing - it deserves a ticket.

Applying the connection factory with shared pointer to the stream (as seen in Sesor-Central) should fix the problem that you described, but I am not saying that this is the only viable solution to it. I will open a ticket.

A thought about this:

In case Sensor pushes a message to perform Compliance checks (message TriggerRun) through this stream, it will get lost.

we may close the stream immediately after sending the scan results to notify sensor that this connection should not be used anymore.

we may close the stream immediately after sending the scan results to notify sensor that this connection should not be used anymore.

I'm afraid the older active connection established by regular compliance logic may not get restored in Sensor's connection map. It deserves some investigation.

I think the problem is that currently Sensor first calls Compliance to do something (after the initial handshake) whereas in our new flow Compliance will first call Sensor. I don't think gRPC bidirectional streams allow receiving and sending at the same time (i.e. full duplex). As a peer in the stream you can either receive or send (or do nothing) but not receive and send concurrently. Therefore the existing stream can only be reused if Sensor initiates scans (going back to one of your proposals Piotr). We will need a different rpc if otherwise we want Compliance to initiate sending scan results (and Sensor ack or nack it).
This may be confusing to read, probably more efficient would be to explain with showing what I mean in code.

I think, your idea about Sensors initiating scans deserves to be reconsidered.

As a peer in the stream you can either receive or send (or do nothing) but not receive and send concurrently.

This would be a limitation. I got the impression from Sensor<->Central that the sending and receiving works concurrently, but I didn't confirm this experimentally.

Having a separate rpc in the proto file would be also an option (maybe even a much cleaner one), but I do not think that this is necessary. The way we initiate the stream clients gives us possibility to send without depending on whether something was received (it is not in the final PR, but we had this variant shortly in the branch).

I am thinking how to design a manual test to confirm or deny our theories 🤔

Created https://issues.redhat.com/browse/ROX-13383 to not forget about the issue we are discussing here.

It turns I was wrong about no duplex support. Looking at Sensor code, one stream is used for both sending and receiving concurrently.

In case Sensor pushes a message to perform Compliance checks (message TriggerRun) through this stream, it will get lost. (@msugakov)

A single manual test is nothing to go by, but I was able to successfully run a compliance check with the FF turned on, so it might be the message isn't lost necessarily.
But we should definitely investigate this further! Thanks for creating the ticket @vikin91