workloadmeta/docker: recover panics in stream() event handlers (quickfix)

The docker workloadmeta collector's stream() goroutine runs forever and
handles two event streams (container + image) by calling
handleContainerEvent / handleImageEvent inline. Neither path had any
panic recovery, so a single panic inside buildCollectorEvent \u2014 for
example, one raised by moby/client's JSON decoder while processing a
malformed ContainerInspect response \u2014 would bubble up, terminate the
goroutine and crash the whole agent process.

This happened in production during an SMP regression-detector run on
the gpu_check_hopper_fake_nvml experiment (every replicate crashed the
agent after a few hundred seconds, exhausting the 1h10m SMP timeout);
see taskmds/05001 for the full captured stack and context.

Wrap both event handlers in a runWithRecovery helper that:

  * logs the panic + debug.Stack() at ERROR with enough context to
    identify the offending event (container ID + action for containers,
    action for images),
  * drops the offending event,
  * lets the stream loop continue processing subsequent events.

This is a quickfix, not a root-cause fix \u2014 the underlying panic is
still inside moby/client@v0.4.0's JSON decoder and needs to be fixed
there (or upstream in Go's encoding/json + reflect interaction with
container.InspectResponse). taskmds/05001 tracks that work. The
companion regression test for the agent-visible half of the bug is at
pkg/util/docker/inspect_panic_test.go (commit bc6a6f4).

Three unit tests cover the new helper:

  * TestRunWithRecovery_SwallowsPanic \u2014 a panicking function returns
    to its caller normally.
  * TestRunWithRecovery_NoPanicNoOp \u2014 the happy path is unchanged.
  * TestRunWithRecovery_SubsequentCallsAfterPanic \u2014 the reusability
    property the stream loop actually relies on: a panic in one call
    does not affect the next.

Author: scottopell

comp/core/workloadmeta/collectors/internal/docker/docker.go

@@ -12,6 +12,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime/debug"
 	"slices"
 	"strconv"
 	"strings"
@@ -167,16 +168,10 @@ func (c *collector) stream(ctx context.Context) {
 		case <-health.C:
 
 		case ev := <-c.containerEventsCh:
-			err := c.handleContainerEvent(ctx, ev)
-			if err != nil {
-				log.Warnf("%s", err.Error())
-			}
+			c.handleContainerEventSafely(ctx, ev)
 
 		case ev := <-c.imageEventsCh:
-			err := c.handleImageEvent(ctx, ev, nil)
-			if err != nil {
-				log.Warnf("%s", err.Error())
-			}
+			c.handleImageEventSafely(ctx, ev)
 
 		case <-ctx.Done():
 			var err error
@@ -198,6 +193,49 @@ func (c *collector) stream(ctx context.Context) {
 	}
 }
 
+// handleContainerEventSafely wraps handleContainerEvent in a defer/recover so
+// that a panic inside buildCollectorEvent (for example, a panic raised by
+// moby/client's JSON decoder while processing a malformed ContainerInspect
+// response — see taskmds/05001 for the full story) does not tear down the
+// agent. The event is dropped and the stream loop continues.
+func (c *collector) handleContainerEventSafely(ctx context.Context, ev *docker.ContainerEvent) {
+	runWithRecovery(
+		fmt.Sprintf("handling docker container event (containerID=%q action=%q)", ev.ContainerID, ev.Action),
+		func() {
+			if err := c.handleContainerEvent(ctx, ev); err != nil {
+				log.Warnf("%s", err.Error())
+			}
+		},
+	)
+}
+
+// handleImageEventSafely is the panic-recovery wrapper for image events.
+// Mirrors handleContainerEventSafely so a single bad image event cannot
+// take down the collector goroutine.
+func (c *collector) handleImageEventSafely(ctx context.Context, ev *docker.ImageEvent) {
+	runWithRecovery(
+		fmt.Sprintf("handling docker image event (action=%q)", ev.Action),
+		func() {
+			if err := c.handleImageEvent(ctx, ev, nil); err != nil {
+				log.Warnf("%s", err.Error())
+			}
+		},
+	)
+}
+
+// runWithRecovery invokes fn with a deferred panic recovery. If fn panics,
+// the panic value and a stack trace are logged at ERROR and runWithRecovery
+// returns normally. Extracted so the behaviour can be exercised from a unit
+// test without having to mock the full DockerUtil surface.
+func runWithRecovery(what string, fn func()) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Errorf("panic while %s, dropping event: %v\n%s", what, r, debug.Stack())
+		}
+	}()
+	fn()
+}
+
 func (c *collector) generateEventsFromContainerList(ctx context.Context, filter workloadfilter.FilterBundle) error {
 	if c.store == nil {
 		return errors.New("Start was not called")

comp/core/workloadmeta/collectors/internal/docker/docker_recovery_test.go

@@ -0,0 +1,78 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-present Datadog, Inc.
+
+//go:build docker
+
+package docker
+
+// Tests for the panic-recovery wrapper used inside the collector's stream
+// goroutine. See taskmds/05001 for the underlying bug that motivated adding
+// this wrapper.
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestRunWithRecovery_SwallowsPanic verifies that runWithRecovery swallows
+// a panic raised inside the passed function and returns normally. This is
+// the property the docker workloadmeta collector's stream goroutine relies
+// on — without it, a panic inside ContainerInspect (see taskmds/05001)
+// propagates up to the goroutine and kills the whole agent process.
+func TestRunWithRecovery_SwallowsPanic(t *testing.T) {
+	didReturn := false
+
+	runWithRecovery("unit test", func() {
+		panic("simulated moby/client json decode panic")
+	})
+
+	didReturn = true
+	require.Truef(t, didReturn,
+		"runWithRecovery must return to its caller even when the wrapped "+
+			"function panics; if this assertion fires it means the defer/recover "+
+			"pair was removed and any panic inside handleContainerEvent will "+
+			"tear down the collector goroutine (and in production, the agent "+
+			"process itself).")
+}
+
+// TestRunWithRecovery_NoPanicNoOp verifies the happy path: if the wrapped
+// function returns normally, runWithRecovery does not alter control flow.
+func TestRunWithRecovery_NoPanicNoOp(t *testing.T) {
+	callCount := 0
+	runWithRecovery("unit test", func() {
+		callCount++
+	})
+	require.Equal(t, 1, callCount,
+		"runWithRecovery must invoke the wrapped function exactly once "+
+			"when it returns normally")
+}
+
+// TestRunWithRecovery_SubsequentCallsAfterPanic is the behaviour the stream
+// loop cares about: after one call to runWithRecovery panics, subsequent
+// calls must still work. This is what makes a single bad container event
+// safe to drop without taking the collector with it.
+func TestRunWithRecovery_SubsequentCallsAfterPanic(t *testing.T) {
+	runs := 0
+
+	runWithRecovery("first", func() {
+		runs++
+		panic("first call panics")
+	})
+	runWithRecovery("second", func() {
+		runs++
+	})
+	runWithRecovery("third", func() {
+		runs++
+		panic("third call panics too")
+	})
+	runWithRecovery("fourth", func() {
+		runs++
+	})
+
+	require.Equal(t, 4, runs,
+		"runWithRecovery must be reusable across events — each call is "+
+			"independent; a panic in one call must not affect the next")
+}