Optimize pull request - avoid unnecessary go routines and move constant definition

Signed-off-by: Klaus Wagner <Klaus.Wagner@erstegroup.com>

Author: Klaus Wagner

plugins/container/go-worker/pkg/container/cri.go

@@ -450,7 +450,6 @@ func (c *criEngine) List(ctx context.Context) ([]event.Event, error) {
 func (c *criEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan event.Event, error) {
 	containerEventsCh := make(chan *v1.ContainerEventResponse)
 	containerEventsErrorCh := make(chan error)
-	const containerEventsErrorTimeout = 10 * time.Millisecond
 	wg.Add(1)
 	go func() {
 		defer close(containerEventsCh)
@@ -464,14 +463,7 @@ func (c *criEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan even
 	case err := <-containerEventsErrorCh:
 		return nil, err
 	case <-time.After(containerEventsErrorTimeout):
-		// continue reading of error channel to avoid blocking initial go-routine
-		go func() {
-			for {
-				if _, ok := <-containerEventsErrorCh; !ok {
-					break
-				}
-			}
-		}()
+		break
 	}
 
 	outCh := make(chan event.Event)
@@ -483,6 +475,11 @@ func (c *criEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan even
 			select {
 			case <-ctx.Done():
 				return
+			case _, ok := <- containerEventsErrorCh:
+				if !ok {
+					// containerEventsErrorCh has been closed - block further reads from channel
+					containerEventsErrorCh = nil
+				}
 			case evt, ok := <-containerEventsCh:
 				if !ok {
 					// containerEventsCh has been closed - block further reads from channel

plugins/container/go-worker/pkg/container/engine.go

@@ -28,6 +28,8 @@ const (
 	typeCri        engineType = "cri"
 	typeCrio       engineType = "cri-o"
 	typeContainerd engineType = "containerd"
+
+	containerEventsErrorTimeout = 10 * time.Millisecond
 )
 
 type engineType string

plugins/container/go-worker/pkg/container/fetcher.go

@@ -67,8 +67,8 @@ func (f *fetcher) List(_ context.Context) ([]event.Event, error) {
 // a retry is set up every containerFetchRetryInterval until containerFetchRetryTimeout is reached.
 // On success, publish event on output channel.
 func (f *fetcher) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan event.Event, error) {
-	const containerFetchRetryInterval = 10 * time.Millisecond
-	const containerFetchRetryTimeout = time.Second
+	const containerFetchRetryInterval = 30 * time.Millisecond
+	const containerFetchRetryTimeout = 150 * time.Millisecond
 	outCh := make(chan event.Event)
 	wg.Add(1)
 	go func() {

plugins/container/go-worker/pkg/container/podman.go

@@ -272,11 +272,11 @@ func (pc *podmanEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan
 
 	evChn := make(chan types.Event)
 	evErrorChn := make(chan error)
-	const eventsErrorTimeout = 10 * time.Millisecond
 	cancelChan := make(chan bool)
 	wg.Add(1)
 	// producers
 	go func(ch chan types.Event) {
+		defer close(evErrorChn)
 		defer wg.Done()
 		evErrorChn <- system.Events(pc.pCtx, ch, cancelChan, &system.EventsOptions{
 			Filters: filters,
@@ -288,15 +288,8 @@ func (pc *podmanEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan
 	select {
 	case err := <-evErrorChn:
 		return nil, err
-	case <-time.After(eventsErrorTimeout):
-		// continue reading of error channel to avoid blocking initial go-routine
-		go func() {
-			for {
-				if _, ok := <-evErrorChn; !ok {
-					break
-				}
-			}
-		}()
+	case <-time.After(containerEventsErrorTimeout):
+		break
 	}
 
 	outCh := make(chan event.Event)
@@ -312,6 +305,11 @@ func (pc *podmanEngine) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan
 			select {
 			case <-ctx.Done():
 				return
+			case _, ok := <-evErrorChn:
+				if !ok {
+					// evErrorChn has been closed - block further reads from channel
+					evErrorChn = nil
+				}
 			case ev, ok := <-evChn:
 				var (
 					ctr *define.InspectContainerData

plugins/container/go-worker/worker_api.go

@@ -132,5 +132,7 @@ func AskForContainerInfo(pCtx unsafe.Pointer, containerId *C.cchar_t) bool {
 			return false
 		}
 	}
+	// In case the fetch channel is nil a retry from falco
+	// does not make sense, report the containerId as handled
 	return true
 }

plugins/container/src/plugin.cpp

@@ -386,7 +386,9 @@ void my_plugin::on_new_process(const falcosecurity::table_entry& thread_entry,
                 if(AskForContainerInfo(m_async_ctx, container_id.c_str()))
                 {
                     m_asked_containers.insert(container_id);
-                } else {
+                }
+                else
+                {
                     m_logger.log(
                             fmt::format("failed to ask the plugin to fetch "
                                         "info for "