fix: add running flag to engine

Since signature Init is now called only on engine.Start(), a
synchronization primitive is needed to signal when signature data is
available to correctly read (since an Init call is required to guarantee
correctness).

Use this flag in the engine tests where needed.

pkg/signatures/engine/engine.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"github.com/aquasecurity/tracee/pkg/logger"
 	"github.com/aquasecurity/tracee/pkg/signatures/metrics"
@@ -37,6 +38,7 @@ type Engine struct {
 	stats            metrics.Stats
 	dataSources      map[string]map[string]detect.DataSource
 	dataSourcesMutex sync.RWMutex
+	running          atomic.Bool
 }
 
 // EventSources is a bundle of input sources used to configure the Engine
@@ -48,6 +50,10 @@ func (engine *Engine) Stats() *metrics.Stats {
 	return &engine.stats
 }
 
+func (engine *Engine) Running() bool {
+	return engine.running.Load()
+}
+
 // NewEngine creates a new signatures-engine with the given arguments
 // inputs and outputs are given as channels created by the consumer
 func NewEngine(config Config, sources EventSources, output chan detect.Finding) (*Engine, error) {
@@ -60,6 +66,8 @@ func NewEngine(config Config, sources EventSources, output chan detect.Finding)
 	engine.inputs = sources
 	engine.output = output
 	engine.config = config
+	engine.running = atomic.Bool{}
+	engine.running.Store(false)
 
 	engine.signaturesMutex.Lock()
 	engine.signatures = make(map[detect.Signature]chan protocol.Event)
@@ -109,6 +117,7 @@ func (engine *Engine) Start(ctx context.Context) {
 		go signatureStart(s, c, &engine.waitGroup)
 	}
 	engine.signaturesMutex.RUnlock()
+	engine.running.Store(true)
 	engine.consumeSources(ctx)
 }
 

pkg/signatures/engine/engine_test.go

@@ -361,6 +361,14 @@ func TestEngine_ConsumeSources(t *testing.T) {
 				e.Start(ctx)
 			}()
 
+			checkTicker := time.NewTicker(500 * time.Millisecond)
+			for range checkTicker.C {
+				if e.Running() {
+					checkTicker.Stop()
+					break
+				}
+			}
+
 			// send a test event
 			e.inputs.Tracee <- tc.inputEvent
 
@@ -419,6 +427,16 @@ func TestEngine_GetSelectedEvents(t *testing.T) {
 	config := Config{Signatures: sigs}
 	e, err := NewEngine(config, EventSources{Tracee: make(chan protocol.Event)}, make(chan detect.Finding))
 	require.NoError(t, err, "constructing engine")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go e.Start(ctx)
+	checkTicker := time.NewTicker(500 * time.Millisecond)
+	for range checkTicker.C {
+		if e.Running() {
+			checkTicker.Stop()
+			break
+		}
+	}
 	se := e.GetSelectedEvents()
 	expected := []detect.SignatureEventSelector{
 		{