add_docker_metadata: fix data races in lazy cgroup cache init (#51688)

add_docker_metadata can run as a beat-level (global) processor, in
which case a single instance is shared by every pipeline client and
its Run method is invoked concurrently. Two data races existed on
that path, both reachable by default because match_pids defaults to
["process.pid", "process.parent.pid"].

lazyCgroupCacheInit guarded the cache with a plain
"if d.cgroups == nil" check-then-set. Concurrent Run calls raced on
the pointer, and two callers could each build a cache and start its
own janitor goroutine, leaking a goroutine and orphaning a cache.
Replace the field with an atomic.Pointer[common.Cache] initialized
exactly once through sync.Once, and load it everywhere it is read.

common.Cache.Get held only the read lock, but get updates the
element's last access time on access-expiry caches (the cgroup cache
is one), so concurrent Get calls mutated the same element under a
shared lock. Take the exclusive lock in Get.

Co-authored-by: Lee E. Hinman <lee.e.hinman@elastic.co>

Author: orestisfl

changelog/fragments/1783070200-fix-add-docker-metadata-cgroup-cache-races.yaml

@@ -0,0 +1,3 @@
+kind: bug-fix
+summary: fix data races in the add_docker_metadata cache initialization
+component: all

libbeat/common/cache.go

@@ -178,8 +178,9 @@ func (c *Cache) ReplaceWithTimeout(k Key, v Value, timeout time.Duration) Value
 // Get the current value associated with a key or nil if the key is not
 // present. The last access time of the element is updated.
 func (c *Cache) Get(k Key) Value {
-	c.RLock()
-	defer c.RUnlock()
+	// Exclusive lock because get() updates the element's last access time on access-expiry caches.
+	c.Lock()
+	defer c.Unlock()
 	v, _ := c.get(k)
 	return v
 }

libbeat/processors/add_docker_metadata/add_docker_metadata.go

@@ -68,14 +68,16 @@ type addDockerMetadata struct {
 	fields          []string
 	sourceProcessor beat.Processor
 
-	pidFields       []string      // Field names that contain PIDs.
-	cgroups         *common.Cache // Cache of PID (int) to container ids (string).
-	dedot           bool          // If set to true, replace dots in labels with `_`.
-	dockerAvailable atomic.Bool   // If Docker exists in env, then it is set to true
-	closeRetry      chan struct{} // Channel to signal the connection retry goroutine to stop
+	pidFields       []string                     // Field names that contain PIDs.
+	cgroups         atomic.Pointer[common.Cache] // Cache of PID (int) to container ids (string).
+	cgroupsOnce     sync.Once                    // Guards the lazy initialization of cgroups.
+	dedot           bool                         // If set to true, replace dots in labels with `_`.
+	dockerAvailable atomic.Bool                  // If Docker exists in env, then it is set to true
+	closeRetry      chan struct{}                // Channel to signal the connection retry goroutine to stop
 	waitRetry       sync.WaitGroup
 	closeOnce       sync.Once
 	closeErr        error
+	closed          atomic.Bool // Set by Close so a late cgroupCache skips starting the janitor.
 	cgreader        processors.CGReader
 	retryPeriod     time.Duration // Period to wait when reconnecting to Docker
 	retryTimeout    time.Duration // Maximum time to wait when connecting to Docker, 0 means wait forever.
@@ -239,15 +241,22 @@ func (d *addDockerMetadata) retryConnectToDocker(connectToDocker func() error, r
 	}
 }
 
-func lazyCgroupCacheInit(d *addDockerMetadata) {
-	if d.cgroups == nil {
+// cgroupCache returns the PID-to-container-ID cache, creating it and starting
+// its janitor on first use. It is safe to call from concurrent Run goroutines.
+func (d *addDockerMetadata) cgroupCache() *common.Cache {
+	d.cgroupsOnce.Do(func() {
 		d.log.Debug("Initializing cgroup cache")
 		evictionListener := func(k common.Key, v common.Value) {
 			d.log.Debugf("Evicted cached cgroups for PID=%v", k)
 		}
-		d.cgroups = common.NewCacheWithRemovalListener(cgroupCacheExpiration, 100, evictionListener)
-		d.cgroups.StartJanitor(5 * time.Second)
-	}
+		cache := common.NewCacheWithRemovalListener(cgroupCacheExpiration, 100, evictionListener)
+		d.cgroups.Store(cache)
+		// Avoid a race and only start the janitor only if Close() has not be called yet.
+		if !d.closed.Load() {
+			cache.StartJanitor(5 * time.Second)
+		}
+	})
+	return d.cgroups.Load()
 }
 
 func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
@@ -334,8 +343,9 @@ func (d *addDockerMetadata) Run(event *beat.Event) (*beat.Event, error) {
 
 func (d *addDockerMetadata) Close() error {
 	d.closeOnce.Do(func() {
-		if d.cgroups != nil {
-			d.cgroups.StopJanitor()
+		d.closed.Store(true) // Prevent the janitor from starting.
+		if cgroups := d.cgroups.Load(); cgroups != nil {
+			cgroups.StopJanitor()
 		}
 
 		// Stop the retry goroutine, this is safe to call even if the goroutine is not running.
@@ -377,8 +387,8 @@ func (d *addDockerMetadata) lookupContainerIDByPID(event *beat.Event) (string, e
 			continue
 		}
 
-		if d.cgroups != nil {
-			if cid := d.cgroups.Get(pid); cid != nil {
+		if cgroups := d.cgroups.Load(); cgroups != nil {
+			if cid := cgroups.Get(pid); cid != nil {
 				d.log.Debugf("Using cached cgroups for pid=%v", pid)
 				cidStr, ok := cid.(string)
 				if !ok {
@@ -401,11 +411,9 @@ func (d *addDockerMetadata) lookupContainerIDByPID(event *beat.Event) (string, e
 			d.log.Debugf("failed to get cgroups for pid=%v: %v", pid, err)
 		}
 
-		// Initialize at time of first use.
-		lazyCgroupCacheInit(d)
-
 		cid, err := getContainerIDFromCgroups(cgroups)
-		d.cgroups.Put(pid, cid)
+		// Cache the result, creating the cache on first use.
+		d.cgroupCache().Put(pid, cid)
 
 		return cid, err
 	}

libbeat/processors/add_docker_metadata/add_docker_metadata_test.go

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"os"
 	"runtime"
+	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -33,6 +34,7 @@ import (
 
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/processors"
+	"github.com/elastic/beats/v7/libbeat/tests/resources"
 	"github.com/elastic/elastic-agent-autodiscover/bus"
 	"github.com/elastic/elastic-agent-autodiscover/docker"
 	"github.com/elastic/elastic-agent-libs/config"
@@ -447,6 +449,84 @@ func TestMatchPIDs(t *testing.T) {
 	})
 }
 
+func TestMatchPIDsConcurrent(t *testing.T) {
+	containerID := "8c147fdfab5a2608fe513d10294bf77cb502a231da9725093a155bd25cd1f14b"
+	p, err := buildDockerMetadataProcessor(logp.NewNopLogger(), config.NewConfig(), MockWatcherFactory(
+		map[string]*docker.Container{
+			containerID: {
+				ID:    containerID,
+				Image: "image",
+				Name:  "name",
+			},
+		},
+		nil,
+	))
+	require.NoError(t, err, "initializing add_docker_metadata processor")
+	t.Cleanup(func() {
+		assert.NoError(t, processors.Close(p), "closing add_docker_metadata processor")
+	})
+
+	// Concurrent Run calls on a shared processor must not race on the lazily
+	// initialized cgroup cache.
+	start := make(chan struct{})
+	var wg sync.WaitGroup
+	for range 10 {
+		wg.Go(func() {
+			<-start
+
+			fields := mapstr.M{}
+			fields.Put("process.pid", 1000)
+
+			result, err := p.Run(&beat.Event{Fields: fields})
+			if !assert.NoError(t, err, "processing an event") {
+				return
+			}
+			cid, err := result.Fields.GetValue("container.id")
+			assert.NoError(t, err, "getting container.id")
+			assert.Equal(t, containerID, cid)
+		})
+	}
+	close(start)
+	wg.Wait()
+}
+
+// TestCloseBeforeCgroupCacheNoJanitorLeak covers Close running before the lazy cgroup cache is
+// initialized. A later cgroupCache call (as a late Run would trigger) must not start a janitor
+// goroutine that Close can no longer stop.
+func TestCloseBeforeCgroupCacheNoJanitorLeak(t *testing.T) {
+	containerID := "8c147fdfab5a2608fe513d10294bf77cb502a231da9725093a155bd25cd1f14b"
+	p, err := buildDockerMetadataProcessor(logp.NewNopLogger(), config.NewConfig(), MockWatcherFactory(
+		map[string]*docker.Container{
+			containerID: {
+				ID:    containerID,
+				Image: "image",
+				Name:  "name",
+			},
+		},
+		nil,
+	))
+	require.NoError(t, err, "initializing add_docker_metadata processor")
+	d := p.(*addDockerMetadata)
+
+	assert.NoError(t, processors.Close(p), "closing processor")
+	assert.Nil(t, d.cgroups.Load(), "cgroups should be nil, the cache was never initialized before Close")
+
+	// Stop any janitor a regression would start so it does not affect other tests.
+	t.Cleanup(func() {
+		if c := d.cgroups.Load(); c != nil {
+			c.StopJanitor()
+		}
+	})
+
+	// Baseline after Close, once all processor goroutines have stopped.
+	goroutinesChecker := resources.NewGoroutinesChecker()
+	goroutinesChecker.FinalizationTimeout = 2 * time.Second
+
+	d.cgroupCache()
+
+	goroutinesChecker.Check(t)
+}
+
 func TestWatcherError(t *testing.T) {
 	logger, observedLogs := logptest.NewTestingLoggerWithObserver(t, "")
 	testConfig, err := config.NewConfigFrom(map[string]interface{}{