feat(ebpf): synchronize ready callback to pipeline

Make sure the event pipeline starts before the ready callback is called.
This is so the ready callback can reliably use tracee methods which may
only work post pipeline initialization.

pkg/ebpf/events_pipeline.go

@@ -25,7 +25,7 @@ const noSyscall int32 = -1
 // handleEvents is the main pipeline of tracee. It receives events from the perf buffer
 // and passes them through a series of stages, each stage is a goroutine that performs a
 // specific task on the event. The pipeline is started in a separate goroutine.
-func (t *Tracee) handleEvents(ctx context.Context) {
+func (t *Tracee) handleEvents(ctx context.Context, initialized chan<- struct{}) {
 	logger.Debugw("Starting handleEvents goroutine")
 	defer logger.Debugw("Stopped handleEvents goroutine")
 
@@ -79,6 +79,8 @@ func (t *Tracee) handleEvents(ctx context.Context) {
 	errc = t.sinkEvents(ctx, eventsChan)
 	errcList = append(errcList, errc)
 
+	initialized <- struct{}{}
+
 	// Pipeline started. Waiting for pipeline to complete
 
 	if err := t.WaitForPipeline(errcList...); err != nil {

pkg/ebpf/tracee.go

@@ -1419,8 +1419,9 @@ func (t *Tracee) Run(ctx gocontext.Context) error {
 
 	t.eventsPerfMap.Poll(pollTimeout)
 
+	pipelineReady := make(chan struct{}, 1)
 	go t.processLostEvents() // termination signaled by closing t.done
-	go t.handleEvents(ctx)
+	go t.handleEvents(ctx, pipelineReady)
 
 	// Parallel perf buffer with file writes events
 
@@ -1443,6 +1444,7 @@ func (t *Tracee) Run(ctx gocontext.Context) error {
 
 	// Management
 
+	<-pipelineReady
 	t.running.Store(true) // set running state after writing pid file
 	t.ready(ctx)          // executes ready callback, non blocking
 	<-ctx.Done()          // block until ctx is cancelled elsewhere