chore(plugins/container): let async_ctx own the fetcher channel.

Signed-off-by: Federico Di Pierro <nierro92@gmail.com>

Author: FedeDP

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

@@ -7,35 +7,31 @@ import (
 )
 
 /*
-Fetcher is a fake engine that listens on fetcherChan for published containerIDs.
+Fetcher is a fake engine that listens on a channel for published containerIDs.
 Everytime a containerID is published on the channel, the fetcher engine loops
 over all enabled engines and tries to get info about the container,
 until it succeeds and publish an event to the output channel.
 FetcherChan requests are published through a CGO exposed API: AskForContainerInfo(), in worker_api.
 */
 
-var fetcherChan chan string
-
-func GetFetcherChan() chan<- string {
-	return fetcherChan
-}
-
 type fetcher struct {
-	getters []getter
-	ctx     context.Context
+	getters     []getter
+	ctx         context.Context
+	fetcherChan <-chan string
 }
 
 // NewFetcherEngine returns a fetcher engine.
 // The fetcher engine is responsible to allow us to get() single container
 // trying all container engines enabled.
-func NewFetcherEngine(_ context.Context, containerEngines []Engine) Engine {
+func NewFetcherEngine(_ context.Context, fetcherChan <-chan string, containerEngines []Engine) Engine {
 	f := fetcher{
 		getters: make([]getter, len(containerEngines)),
 		// Since podman relies upon context to store
 		// connection-related info,
 		// we need a unique context for fetcher
 		// to avoid tampering with real podman engine context.
-		ctx: context.Background(),
+		ctx:         context.Background(),
+		fetcherChan: fetcherChan,
 	}
 	for i, engine := range containerEngines {
 		copyEngine, ok := engine.(copier)
@@ -67,19 +63,16 @@ func (f *fetcher) List(_ context.Context) ([]event.Event, error) {
 func (f *fetcher) Listen(ctx context.Context, wg *sync.WaitGroup) (<-chan event.Event, error) {
 	outCh := make(chan event.Event)
 	wg.Add(1)
-	fetcherChan = make(chan string)
 	go func() {
 		defer func() {
 			close(outCh)
-			close(fetcherChan)
-			fetcherChan = nil
 			wg.Done()
 		}()
 		for {
 			select {
 			case <-ctx.Done():
 				return
-			case containerId := <-fetcherChan:
+			case containerId := <-f.fetcherChan:
 				for _, e := range f.getters {
 					evt, _ := e.get(f.ctx, containerId)
 					if evt != nil {

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

@@ -32,7 +32,9 @@ func TestCRIFetcher(t *testing.T) {
 func testFetcher(t *testing.T, containerEngine Engine, containerId string, expectedEvent event.Event) {
 	// Create the fetcher engine with the docker engine as the only container engine
 	containerEngines := []Engine{containerEngine}
-	f := NewFetcherEngine(context.Background(), containerEngines)
+	fetchCh := make(chan string)
+	assert.NotNil(t, fetchCh)
+	f := NewFetcherEngine(context.Background(), fetchCh, containerEngines)
 	assert.NotNil(t, f)
 
 	// Check that fetcher is able to fetch the container
@@ -47,11 +49,9 @@ func testFetcher(t *testing.T, containerEngine Engine, containerId string, expec
 	assert.NoError(t, err)
 
 	// Send the container ID to the fetcher channel to request its info to be loaded
-	ch := GetFetcherChan()
-	assert.NotNil(t, ch)
 	go func() {
 		time.Sleep(1 * time.Second)
-		ch <- containerId
+		fetchCh <- containerId
 	}()
 
 	evt := waitOnChannelOrTimeout(t, listCh)

plugins/container/go-worker/worker_api.go

@@ -25,6 +25,7 @@ type PluginCtx struct {
 	ctxCancel    context.CancelFunc
 	stringBuffer ptr.StringBuffer
 	pinner       runtime.Pinner
+	fetchCh      chan string
 }
 
 //export StartWorker
@@ -79,9 +80,12 @@ func StartWorker(cb C.async_cb, initCfg *C.cchar_t, enabledSocks **C.cchar_t) un
 			}
 		}
 	}
+
+	pluginCtx.fetchCh = make(chan string)
+
 	// Always append the dummy engine that is required to
 	// be able to fetch container infos on the fly given other enabled engines.
-	containerEngines = append(containerEngines, container.NewFetcherEngine(ctx, containerEngines))
+	containerEngines = append(containerEngines, container.NewFetcherEngine(ctx, pluginCtx.fetchCh, containerEngines))
 
 	// Store json of attached sockets in `enabledSocks`
 	bytes, _ := json.Marshal(enabledEngines)
@@ -106,16 +110,20 @@ func StopWorker(pCtx unsafe.Pointer) {
 	pluginCtx.ctxCancel()
 	pluginCtx.wg.Wait()
 	pluginCtx.stringBuffer.Free()
+	close(pluginCtx.fetchCh)
+	pluginCtx.fetchCh = nil
 
 	pluginCtx.pinner.Unpin()
 	h.Delete()
 }
 
 //export AskForContainerInfo
-func AskForContainerInfo(containerId *C.cchar_t) {
+func AskForContainerInfo(pCtx unsafe.Pointer, containerId *C.cchar_t) {
+	h := (*cgo.Handle)(pCtx)
+	pluginCtx := h.Value().(*PluginCtx)
+
 	containerID := C.GoString(containerId)
-	ch := container.GetFetcherChan()
-	if ch != nil {
-		ch <- containerID
+	if pluginCtx.fetchCh != nil {
+		pluginCtx.fetchCh <- containerID
 	}
 }

plugins/container/src/plugin.cpp

@@ -372,12 +372,17 @@ void my_plugin::on_new_process(const falcosecurity::table_entry& thread_entry,
                          falcosecurity::_internal::SS_PLUGIN_LOG_SEV_DEBUG);
 #ifdef _HAS_ASYNC
             // Check if already asked
-            if(m_asked_containers.find(container_id) ==
-               m_asked_containers.end())
+            if(m_async_ctx != nullptr &&
+               m_asked_containers.find(container_id) ==
+                       m_asked_containers.end())
             {
+                m_logger.log(fmt::format("asking the plugin to fetch info for "
+                                         "container {}",
+                                         container_id),
+                             falcosecurity::_internal::SS_PLUGIN_LOG_SEV_DEBUG);
                 m_asked_containers.insert(container_id);
                 // Implemented by GO worker.go
-                AskForContainerInfo(container_id.c_str());
+                AskForContainerInfo(m_async_ctx, container_id.c_str());
             }
 #endif
         }