fix(engine): feedback deadlock issues

pkg/analyze/analyze.go

@@ -50,14 +50,10 @@ func Analyze(cfg Config) {
 	sigNamesToIds := sigs.CreateEventsFromSignatures(events.StartSignatureID, signatures)
 
 	engineConfig := engine.Config{
+		Mode:                engine.ModeAnalyze,
 		Signatures:          signatures,
 		SignatureBufferSize: 1000,
-		Enabled:             true, // simulate tracee single binary mode
 		SigNameToEventID:    sigNamesToIds,
-		ShouldDispatchEvent: func(eventIdInt32 int32) bool {
-			// in analyze mode we don't need to filter by policy
-			return true
-		},
 	}
 
 	// two seperate contexts.
@@ -68,6 +64,7 @@ func Analyze(cfg Config) {
 
 	engineOutput := make(chan *detect.Finding)
 	engineInput := make(chan protocol.Event)
+	fromFile := make(chan protocol.Event)
 
 	source := engine.EventSources{Tracee: engineInput}
 	sigEngine, err := engine.NewEngine(engineConfig, source, engineOutput)
@@ -91,13 +88,18 @@ func Analyze(cfg Config) {
 	}
 
 	// producer
-	go produce(fileReadCtx, stop, cfg.Source, engineInput)
+	go produce(fileReadCtx, stop, cfg.Source, fromFile)
 
 	cfg.Printer.Preamble()
 	defer cfg.Printer.Close()
 	// consumer
 	for {
 		select {
+		case event, ok := <-fromFile:
+			if !ok {
+				return
+			}
+			engineInput <- event
 		case finding, ok := <-engineOutput:
 			if !ok {
 				return
@@ -116,6 +118,11 @@ drain:
 	defer close(engineInput)
 	for {
 		select {
+		case event, ok := <-fromFile:
+			if !ok {
+				return
+			}
+			engineInput <- event
 		case finding, ok := <-engineOutput:
 			if !ok {
 				return

pkg/cmd/cobra/cobra.go

@@ -314,7 +314,7 @@ func GetTraceeRunner(c *cobra.Command, version string) (cmd.Runner, error) {
 	runner.InstallPath = traceeInstallPath
 
 	runner.TraceeConfig.EngineConfig = engine.Config{
-		Enabled:          true,
+		Mode:             engine.ModeSingleBinary,
 		SigNameToEventID: sigNameToEventId,
 		Signatures:       signatures,
 		// This used to be a flag, we have removed the flag from this binary to test

pkg/ebpf/events_pipeline.go

@@ -14,6 +14,7 @@ import (
 	"github.com/aquasecurity/tracee/pkg/errfmt"
 	"github.com/aquasecurity/tracee/pkg/events"
 	"github.com/aquasecurity/tracee/pkg/logger"
+	"github.com/aquasecurity/tracee/pkg/signatures/engine"
 	traceetime "github.com/aquasecurity/tracee/pkg/time"
 	"github.com/aquasecurity/tracee/pkg/utils"
 	"github.com/aquasecurity/tracee/types/trace"
@@ -72,7 +73,7 @@ func (t *Tracee) handleEvents(ctx context.Context, initialized chan<- struct{})
 
 	// Engine events stage: events go through the signatures engine for detection.
 
-	if t.config.EngineConfig.Enabled {
+	if t.config.EngineConfig.Mode == engine.ModeSingleBinary {
 		eventsChan, errc = t.engineEvents(ctx, eventsChan)
 		errcList = append(errcList, errc)
 	}
@@ -615,7 +616,7 @@ func (t *Tracee) sinkEvents(ctx context.Context, in <-chan *trace.Event) <-chan
 			event.MatchedPolicies = t.policyManager.MatchedNames(event.MatchedPoliciesUser)
 
 			// Parse args here if the rule engine is NOT enabled (parsed there if it is).
-			if t.config.Output.ParseArguments && !t.config.EngineConfig.Enabled {
+			if t.config.Output.ParseArguments && t.config.EngineConfig.Mode != engine.ModeSingleBinary {
 				err := t.parseArguments(event)
 				if err != nil {
 					t.handleError(err)

pkg/signatures/engine/engine.go

@@ -18,10 +18,18 @@ const EVENT_CONTAINER_ORIGIN = "container"
 const EVENT_HOST_ORIGIN = "host"
 const ALL_EVENT_TYPES = "*"
 
+type Mode uint8
+
+const (
+	ModeRules Mode = iota
+	ModeAnalyze
+	ModeSingleBinary
+)
+
 // Config defines the engine's configurable values
 type Config struct {
 	// Engine-in-Pipeline related configuration
-	Enabled          bool             // Enables the signatures engine to run in the events pipeline
+	Mode             Mode
 	SigNameToEventID map[string]int32 // Cache of loaded signature event names to event ids, used to filter in dispatching
 
 	// Callback from tracee to determine if event should be dispatched to signature.
@@ -148,9 +156,15 @@ func (engine *Engine) unloadAllSignatures() {
 func (engine *Engine) matchHandler(res *detect.Finding) {
 	_ = engine.stats.Detections.Increment()
 	engine.output <- res
-	if !engine.config.Enabled {
+	// TODO: the feedback here is enabled only in analyze, as it was causing a deadlock in the pipeline
+	//       when the engine was blocked on sending a new event to the feedbacking signature.
+	//       This is because the engine would eventually block on trying to to send
+	//       a new event to the feedbacking signature. This would cause a deadlock
+	//       there - propagating back to the engine and pipeline in general.
+	// TODO2: Once we integrate the pipeline into analyze mode, we can remove this logic.
+	if !(engine.config.Mode == ModeAnalyze) {
 		return
-		// next section is relevant only for engine-in-pipeline and analyze
+		// next section is relevant only for  analyze
 	}
 	e, err := findings.FindingToEvent(res)
 	if err != nil {
@@ -272,8 +286,8 @@ drain:
 }
 
 func (engine *Engine) dispatchEvent(s detect.Signature, event protocol.Event) {
-	if engine.config.Enabled {
-		// Do this test only if engine runs as part of the event pipeline
+	if engine.config.Mode == ModeSingleBinary {
+		// Filter only if engine runs as part of the event pipeline (single binary mode)
 		if ok := engine.filterDispatchInPipeline(s, event); !ok {
 			return
 		}