diff --git a/internal/dockeragent/collect.go b/internal/dockeragent/collect.go index bd740324f..e3e440733 100644 --- a/internal/dockeragent/collect.go +++ b/internal/dockeragent/collect.go @@ -24,6 +24,16 @@ import ( // buildReport gathers all system and container metrics into a single report func (a *Agent) buildReport(ctx context.Context) (agentsdocker.Report, error) { + // Cycle-level containment: per-call deadlines below should bound every + // docker call, but one live exercise saw a call park for minutes with its + // deadline never firing, so the whole cycle gets its own ceiling plus an + // independent watchdog that dumps goroutines if even this deadline fails + // to abort the cycle. + ctx, cancel := context.WithTimeout(ctx, dockerCollectCycleTimeout) + defer cancel() + stopWatchdog := startCollectCycleWatchdog(a.logger, dockerCollectCycleTimeout+dockerCollectWatchdogGrace) + defer stopWatchdog() + info, err := dockerCallWithRetry(ctx, dockerInfoCallTimeout, func(callCtx context.Context) (systemtypes.Info, error) { return a.docker.Info(callCtx) }) diff --git a/internal/dockeragent/hardening.go b/internal/dockeragent/hardening.go index b532e0a75..82b15381d 100644 --- a/internal/dockeragent/hardening.go +++ b/internal/dockeragent/hardening.go @@ -5,9 +5,12 @@ import ( "errors" "fmt" "net" + "runtime" "strings" "syscall" "time" + + "github.com/rs/zerolog" ) const ( @@ -19,10 +22,48 @@ const ( dockerUpdateCallTimeout = 2 * time.Minute dockerUpdateOverallTimeout = 15 * time.Minute + // dockerCollectCycleTimeout bounds one whole collection cycle + // (buildReport). Every docker call inside the cycle already carries its + // own per-call timeout, so a healthy cycle finishes in seconds; this + // ceiling only trips when a call stalls without its deadline aborting it + // (observed once on 2026-07-17: a DiskUsage roundtrip parked 6+ minutes + // on a colima daemon with the 20s per-call deadline never firing). + dockerCollectCycleTimeout = 5 * time.Minute + + // dockerCollectWatchdogGrace is how long past dockerCollectCycleTimeout + // the watchdog waits before concluding that the cycle context deadline + // failed to abort the cycle and dumping goroutines for diagnosis. + dockerCollectWatchdogGrace = 30 * time.Second + dockerCallRetryAttempts = 2 dockerRetryBaseDelay = 200 * time.Millisecond ) +// startCollectCycleWatchdog arms an independent timer that fires only if a +// collection cycle is still running after budget. The cycle context deadline +// should abort the cycle well before then, so the watchdog firing means +// context cancellation was swallowed or a runtime timer was lost; it logs the +// full goroutine stack so a recurrence of the 2026-07-17 stall is diagnosable +// in the field. The returned stop function must be called when the cycle ends. +func startCollectCycleWatchdog(logger zerolog.Logger, budget time.Duration) (stop func()) { + done := make(chan struct{}) + go func() { + timer := time.NewTimer(budget) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + buf := make([]byte, 64*1024) + n := runtime.Stack(buf, true) + logger.Error(). + Dur("budget", budget). + Str("goroutines", string(buf[:n])). + Msg("Docker collect cycle still running past its watchdog budget; the cycle context deadline did not abort it") + } + }() + return func() { close(done) } +} + func dockerCallWithRetry[T any](ctx context.Context, timeout time.Duration, call func(context.Context) (T, error)) (T, error) { return dockerCallWithRetryAttempts(ctx, timeout, dockerCallRetryAttempts, call) } diff --git a/internal/dockeragent/hardening_test.go b/internal/dockeragent/hardening_test.go index 8b3cbda62..9e0d2d022 100644 --- a/internal/dockeragent/hardening_test.go +++ b/internal/dockeragent/hardening_test.go @@ -1,14 +1,37 @@ package dockeragent import ( + "bytes" "context" "errors" "strings" + "sync" "syscall" "testing" "time" + + "github.com/rs/zerolog" ) +// syncBuffer is a goroutine-safe bytes.Buffer for capturing watchdog log +// output written from another goroutine. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(p []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.Write(p) +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + return b.buf.String() +} + func TestDockerCallWithRetryAttempts_RetriesTransientErrors(t *testing.T) { t.Parallel() @@ -80,3 +103,40 @@ func TestAnnotateDockerConnectionError(t *testing.T) { t.Fatalf("expected timeout hint, got %q", timeoutErr.Error()) } } + +func TestCollectCycleWatchdogFiresWhenNotStopped(t *testing.T) { + var out syncBuffer + logger := zerolog.New(&out) + + stop := startCollectCycleWatchdog(logger, 20*time.Millisecond) + defer stop() + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if strings.Contains(out.String(), "watchdog budget") { + break + } + time.Sleep(10 * time.Millisecond) + } + + logged := out.String() + if !strings.Contains(logged, "watchdog budget") { + t.Fatalf("watchdog did not log after budget elapsed: %q", logged) + } + if !strings.Contains(logged, "goroutine") { + t.Fatalf("watchdog log does not include a goroutine dump: %q", logged) + } +} + +func TestCollectCycleWatchdogSilentWhenStopped(t *testing.T) { + var out syncBuffer + logger := zerolog.New(&out) + + stop := startCollectCycleWatchdog(logger, 30*time.Millisecond) + stop() + + time.Sleep(100 * time.Millisecond) + if logged := out.String(); logged != "" { + t.Fatalf("watchdog logged after being stopped: %q", logged) + } +} diff --git a/internal/dockeragent/hung_daemon_deadline_test.go b/internal/dockeragent/hung_daemon_deadline_test.go new file mode 100644 index 000000000..3cd0189b5 --- /dev/null +++ b/internal/dockeragent/hung_daemon_deadline_test.go @@ -0,0 +1,181 @@ +package dockeragent + +// Regression pins for the 2026-07-17 live-exercise stall: collectStorageUsage +// sat 6+ minutes inside moby DiskUsage even though dockerCallWithRetry wraps +// every call in context.WithTimeout. Investigation could not reproduce a +// deadline being swallowed anywhere in the moby client stack; these tests pin +// that a context deadline aborts the real moby client (request.go + otelhttp +// transport + API-version negotiation) against a hung unix-socket daemon, in +// both hang shapes, so a future moby/otelhttp upgrade cannot silently regress +// it: +// +// - hang before response headers (daemon computing /system/df forever) +// - hang mid-body (daemon streams partial JSON then stalls) + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "os" + "path/filepath" + "testing" + "time" + + "github.com/moby/moby/client" +) + +// hungSocketServer listens on a unix socket, accepts connections, discards +// whatever the client writes, and never responds. +func startHungSocketServer(t *testing.T) string { + t.Helper() + + dir, err := os.MkdirTemp("", "pulsedu") + if err != nil { + t.Fatalf("mkdtemp: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + + sockPath := filepath.Join(dir, "docker.sock") + ln, err := net.Listen("unix", sockPath) + if err != nil { + t.Fatalf("listen unix: %v", err) + } + t.Cleanup(func() { _ = ln.Close() }) + + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer c.Close() + buf := make([]byte, 4096) + for { + if _, err := c.Read(buf); err != nil { + return + } + // Read requests forever, never write a byte back. + } + }(conn) + } + }() + + return sockPath +} + +// startStallingHTTPServer answers /_ping so API-version negotiation succeeds, +// then for /system/df writes response headers plus a partial JSON body and +// stalls forever. +func startStallingHTTPServer(t *testing.T) string { + t.Helper() + + dir, err := os.MkdirTemp("", "pulsedu") + if err != nil { + t.Fatalf("mkdtemp: %v", err) + } + t.Cleanup(func() { _ = os.RemoveAll(dir) }) + + sockPath := filepath.Join(dir, "docker.sock") + ln, err := net.Listen("unix", sockPath) + if err != nil { + t.Fatalf("listen unix: %v", err) + } + + hang := make(chan struct{}) + mux := http.NewServeMux() + mux.HandleFunc("/_ping", func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Api-Version", "1.51") + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + // Any versioned endpoint (e.g. /v1.51/system/df): stream a partial + // body, then stall. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, _ = fmt.Fprint(w, `{"LayersSize": 1, "Images": [`) + if f, ok := w.(http.Flusher); ok { + f.Flush() + } + <-hang + }) + + srv := &http.Server{Handler: mux} + go func() { _ = srv.Serve(ln) }() + t.Cleanup(func() { + close(hang) + _ = srv.Close() + }) + + return sockPath +} + +func newTestMobyClient(t *testing.T, sockPath string) dockerClient { + t.Helper() + + cli, err := newMobyDockerClient( + client.WithHost("unix://"+sockPath), + client.WithAPIVersionNegotiation(), + ) + if err != nil { + t.Fatalf("new moby client: %v", err) + } + t.Cleanup(func() { _ = cli.Close() }) + return cli +} + +// diskUsageViaProductionPath mirrors collectStorageUsage's exact wrapper and +// options. +func diskUsageViaProductionPath(ctx context.Context, cli dockerClient, timeout time.Duration) error { + _, err := dockerCallWithRetryAttempts(ctx, timeout, 1, func(callCtx context.Context) (client.DiskUsageResult, error) { + return cli.DiskUsage(callCtx, dockerDiskUsageOptions{ + Containers: true, + Images: true, + BuildCache: true, + Volumes: true, + Verbose: true, + }) + }) + return err +} + +func TestDiskUsageDeadlineAbortsHungSocketBeforeHeaders(t *testing.T) { + sockPath := startHungSocketServer(t) + cli := newTestMobyClient(t, sockPath) + + const timeout = 2 * time.Second + start := time.Now() + err := diskUsageViaProductionPath(context.Background(), cli, timeout) + elapsed := time.Since(start) + + if err == nil { + t.Fatalf("expected error from hung daemon socket, got nil after %v", elapsed) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected context.DeadlineExceeded, got %v (after %v)", err, elapsed) + } + if elapsed > timeout+3*time.Second { + t.Fatalf("deadline did not abort the call promptly: elapsed %v for %v timeout", elapsed, timeout) + } + t.Logf("pre-header hang aborted after %v with: %v", elapsed, err) +} + +func TestDiskUsageDeadlineAbortsHungSocketMidBody(t *testing.T) { + sockPath := startStallingHTTPServer(t) + cli := newTestMobyClient(t, sockPath) + + const timeout = 2 * time.Second + start := time.Now() + err := diskUsageViaProductionPath(context.Background(), cli, timeout) + elapsed := time.Since(start) + + if err == nil { + t.Fatalf("expected error from mid-body stall, got nil after %v", elapsed) + } + if elapsed > timeout+3*time.Second { + t.Fatalf("deadline did not abort the mid-body stall promptly: elapsed %v for %v timeout (err=%v)", elapsed, timeout, err) + } + t.Logf("mid-body stall aborted after %v with: %v", elapsed, err) +}