From c6567296a03fc6ca26f29b7a7735e86b0f9ac2a7 Mon Sep 17 00:00:00 2001 From: silverwind Date: Thu, 29 Oct 2026 15:53:32 +0000 Subject: [PATCH] fix: stop the docker proxy probe from leaking its container (#1357) When the probe's deadline expired during `ContainerCreate`, dockerd still created the container but the probe never got its ID, leaving it in `Created` with its image's anonymous volumes. A slow daemon leaked one per job. The probe now runs detached with its own timeout, so the create is never cancelled and the container is always removed. Callers still fall back after their own deadline, and the runner waits for pending probes before exiting. Fixes: https://gitea.com/gitea/runner/issues/1353 Reviewed-on: https://gitea.com/gitea/runner/pulls/1357 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> Co-authored-by: silverwind --- act/container/docker_proxy.go | 108 +++++++++++------ act/container/docker_proxy_test.go | 178 ++++++++++++++++++++--------- act/container/docker_run_test.go | 2 +- 3 files changed, 195 insertions(+), 93 deletions(-) diff --git a/act/container/docker_proxy.go b/act/container/docker_proxy.go index a105abd9..17bf48f2 100644 --- a/act/container/docker_proxy.go +++ b/act/container/docker_proxy.go @@ -40,7 +40,9 @@ const ( jobLabel = "com.gitea.runner.job." maxCreateBody = 8 << 20 - dockerProxyProbeTimeout = 5 * time.Second + dockerProxyProbeTimeout = 5 * time.Second + dockerProxyProbeWorkerTimeout = time.Minute + dockerProxyProbeTarget = "/gitea-runner-probe" ) var ( @@ -48,7 +50,7 @@ var ( rawStreamPath = regexp.MustCompile(`^(/v[0-9.]+)?/(containers/[^/]+/attach|exec/[^/]+/start)$`) ) -var dockerProxyDirectories, dockerProxyProbes sync.Map +var dockerProxyProbes sync.Map type dockerProxyDirectory struct { dir, daemonDir string @@ -68,7 +70,7 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy { } probeCtx, cancel := context.WithTimeout(ctx, dockerProxyProbeTimeout) defer cancel() - cli, err := GetDockerClient(probeCtx) + cli, err := client.New(client.FromEnv) // negotiates on first request, so a cached probe costs no round trip if err != nil { return nil } @@ -83,65 +85,69 @@ func NewDockerProxy(ctx context.Context, job string) *DockerProxy { } workdir, _ := os.Getwd() key := [3]string{cli.DaemonHost(), workdir, os.TempDir()} - cached, _ := dockerProxyDirectories.Load(key) - directory, seen := cached.(dockerProxyDirectory) - seen = seen && os.SameFile(directory.socket, socket) // daemon replacement invalidates cached paths - if !seen { - directory, seen, err = probeDockerProxyDirectory(probeCtx, cli, key, socket) - } + probe, err := probeDockerProxyDirectory(probeCtx, cli, key, socket) if err != nil { common.Logger(ctx).Infof("docker proxy probe failed, jobs get the daemon socket directly: %v", err) return nil } - if !seen { + if !probe.seen { common.Logger(ctx).Infof("the docker daemon cannot reach the runner's temporary or working directory, jobs get the daemon socket directly") return nil } if ctx.Err() != nil { return nil } - proxy, err := StartDockerProxy(daemonSocket, directory.dir, job) + proxy, err := StartDockerProxy(daemonSocket, probe.directory.dir, job) if err != nil { - dockerProxyDirectories.Delete(key) + dockerProxyProbes.CompareAndDelete(key, probe) common.Logger(ctx).Warnf("docker proxy not started, the job gets the daemon socket directly: %v", err) return nil } - dockerProxyDirectories.Store(key, directory) - proxy.Socket = directory.daemonDir + strings.TrimPrefix(proxy.Socket, directory.dir) + proxy.Socket = probe.directory.daemonDir + strings.TrimPrefix(proxy.Socket, probe.directory.dir) return proxy } -func probeDockerProxyDirectory(ctx context.Context, cli client.APIClient, key [3]string, socket os.FileInfo) (dockerProxyDirectory, bool, error) { +func probeDockerProxyDirectory(ctx context.Context, cli client.APIClient, key [3]string, socket os.FileInfo) (*dockerProxyProbe, error) { for { probe := &dockerProxyProbe{directory: dockerProxyDirectory{socket: socket}, done: make(chan struct{})} actual, loaded := dockerProxyProbes.LoadOrStore(key, probe) if loaded { probe, _ = actual.(*dockerProxyProbe) - if !os.SameFile(probe.directory.socket, socket) { + if !os.SameFile(probe.directory.socket, socket) { // daemon replacement invalidates cached paths dockerProxyProbes.CompareAndDelete(key, probe) continue } - select { - case <-ctx.Done(): - return dockerProxyDirectory{}, false, ctx.Err() - case <-probe.done: - } - if ctx.Err() == nil && (errors.Is(probe.err, context.Canceled) || errors.Is(probe.err, context.DeadlineExceeded)) { - continue - } } else { - probe.directory.dir, probe.directory.daemonDir = runnerContainerWorkdir(ctx, cli) - probe.seen = probe.directory.daemonDir != "" - if !probe.seen { - if probe.directory.dir, probe.err = filepath.Abs(key[2]); probe.err == nil { - probe.directory.daemonDir = probe.directory.dir - probe.seen, probe.err = daemonSeesDir(ctx, cli, probe.directory.dir, probe.directory.daemonDir) + go func() { // outlives callers' deadlines so a slow daemon still gets its result cached + defer cli.Close() + ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeWorkerTimeout) + defer cancel() + probe.directory.dir, probe.directory.daemonDir = runnerContainerWorkdir(ctx, cli) + probe.seen = probe.directory.daemonDir != "" + if !probe.seen { + if probe.directory.dir, probe.err = filepath.Abs(key[2]); probe.err == nil { + probe.directory.daemonDir = probe.directory.dir + probe.seen, probe.err = daemonSeesDir(ctx, cli, probe.directory.dir, probe.directory.daemonDir) + } } - } - dockerProxyProbes.CompareAndDelete(key, probe) - close(probe.done) + if probe.seen { + dockerProxyProbes.LoadOrStore(key, probe) + } else { + dockerProxyProbes.CompareAndDelete(key, probe) + } + close(probe.done) + removeStaleProbeContainers(cli) + }() } - return probe.directory, probe.seen, probe.err + select { + case <-ctx.Done(): + return &dockerProxyProbe{}, ctx.Err() + case <-probe.done: + } + if ctx.Err() == nil && errors.Is(probe.err, context.DeadlineExceeded) { + continue + } + return probe, probe.err } } @@ -200,7 +206,7 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str created, err = cli.ContainerCreate(ctx, client.ContainerCreateOptions{ Config: &container.Config{Image: image.ID, Cmd: []string{"true"}}, HostConfig: &container.HostConfig{Mounts: []mount.Mount{ - {Type: mount.TypeBind, Source: daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it + {Type: mount.TypeBind, Source: daemonDir, Target: dockerProxyProbeTarget, ReadOnly: true}, // not the marker itself, podman creates a missing bind source where docker rejects it }}, }) if !cerrdefs.IsNotFound(err) { @@ -213,7 +219,7 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str if err != nil { return false, err } - seen, err := containerSeesMarker(ctx, cli, dir, created.ID, "/gitea-runner-probe") + seen, err := containerSeesMarker(ctx, cli, dir, created.ID, dockerProxyProbeTarget) cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), dockerProxyProbeTimeout) defer cancel() if _, removeErr := cli.ContainerRemove(cleanupCtx, created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil { @@ -222,6 +228,36 @@ func daemonSeesDir(ctx context.Context, cli client.APIClient, dir, daemonDir str return seen, err } +// removeStaleProbeContainers removes probe containers left by abandoned creates once any probe in flight when listed has ended. +func removeStaleProbeContainers(cli client.APIClient) { + ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeTimeout) + defer cancel() + listed, err := cli.ContainerList(ctx, client.ContainerListOptions{ + All: true, + Filters: make(client.Filters).Add("status", string(container.StateCreated)).Add("volume", dockerProxyProbeTarget), + }) + if err != nil { + common.Logger(ctx).Warnf("listing stale docker proxy probe containers failed: %v", err) + return + } + listed.Items = slices.DeleteFunc(listed.Items, func(stale container.Summary) bool { + return stale.State != container.StateCreated || !slices.ContainsFunc(stale.Mounts, func(mountPoint container.MountPoint) bool { + return mountPoint.Type == mount.TypeBind && !mountPoint.RW && mountPoint.Destination == dockerProxyProbeTarget + }) + }) + if len(listed.Items) == 0 { + return + } + time.Sleep(dockerProxyProbeWorkerTimeout + dockerProxyProbeTimeout) + for _, stale := range listed.Items { + ctx, cancel := context.WithTimeout(context.Background(), dockerProxyProbeTimeout) + if err := (&containerReference{cli: cli, id: stale.ID}).remove()(ctx); err != nil { + common.Logger(ctx).Warnf("removing stale docker proxy probe container failed: %v", err) + } + cancel() + } +} + // StartDockerProxy serves a job's docker socket in dir, labelling what the job creates through it. func StartDockerProxy(daemonSocket, dir, job string) (*DockerProxy, error) { info, err := os.Stat(daemonSocket) diff --git a/act/container/docker_proxy_test.go b/act/container/docker_proxy_test.go index 0cce800e..7c18d7c0 100644 --- a/act/container/docker_proxy_test.go +++ b/act/container/docker_proxy_test.go @@ -94,6 +94,8 @@ func TestDockerProxy(t *testing.T) { return } _, _ = w.Write([]byte(`[{"Id":"probe-image"}]`)) + case strings.HasSuffix(r.URL.Path, "/containers/json"): + _, _ = w.Write([]byte(`[]`)) case strings.HasSuffix(r.URL.Path, "/json"): http.NotFound(w, r) case strings.HasSuffix(r.URL.Path, "/archive"): @@ -235,7 +237,8 @@ func TestDockerProxy(t *testing.T) { t.Skip("automatic proxy discovery requires Linux") } t.Setenv("DOCKER_HOST", "unix://"+daemonSocket) - t.Setenv("TMPDIR", shortTempDir(t)) + tempDir := shortTempDir(t) + t.Setenv("TMPDIR", tempDir) require.Nil(t, NewDockerProxy(t.Context(), "failed")) for _, job := range []string{"first", "second"} { proxy := NewDockerProxy(t.Context(), job) @@ -243,12 +246,10 @@ func TestDockerProxy(t *testing.T) { require.NoError(t, proxy.Close(context.Background())) } assert.EqualValues(t, 2, probes.Load()) - dockerProxyDirectories.Range(func(key, value any) bool { - directory, _ := value.(dockerProxyDirectory) - directory.socket, _ = os.Stat(t.TempDir()) - dockerProxyDirectories.Store(key, directory) - return true - }) + workdir, err := os.Getwd() + require.NoError(t, err) + cached, _ := dockerProxyProbes.Load([3]string{"unix://" + daemonSocket, workdir, tempDir}) + cached.(*dockerProxyProbe).directory.socket, _ = os.Stat(t.TempDir()) proxy := NewDockerProxy(t.Context(), "replaced daemon") require.NotNil(t, proxy) require.NoError(t, proxy.Close(context.Background())) @@ -476,10 +477,23 @@ func BenchmarkDockerProxy(b *testing.B) { type probeClient struct { mobyclient.APIClient - create func(mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) - remove func(context.Context, string, mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) - stat func(string) error - mounts []container.MountPoint + create func(context.Context, mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) + remove func(context.Context, string, mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) + stat func(string) error + mounts []container.MountPoint + containers []container.Summary +} + +func (c *probeClient) Close() error { + return nil +} + +func (c *probeClient) ContainerList(context.Context, mobyclient.ContainerListOptions) (mobyclient.ContainerListResult, error) { + return mobyclient.ContainerListResult{Items: c.containers}, nil +} + +func (c *probeClient) ContainerKill(context.Context, string, mobyclient.ContainerKillOptions) (mobyclient.ContainerKillResult, error) { + return mobyclient.ContainerKillResult{}, cerrdefs.ErrConflict } func (c *probeClient) ContainerInspect(context.Context, string, mobyclient.ContainerInspectOptions) (mobyclient.ContainerInspectResult, error) { @@ -490,8 +504,8 @@ func (c *probeClient) ImageList(context.Context, mobyclient.ImageListOptions) (m return mobyclient.ImageListResult{Items: []image.Summary{{ID: "removed-image"}, {ID: "probe-image"}}}, nil } -func (c *probeClient) ContainerCreate(_ context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { - return c.create(opts) +func (c *probeClient) ContainerCreate(ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + return c.create(ctx, opts) } func (c *probeClient) ContainerRemove(ctx context.Context, id string, opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { @@ -518,7 +532,7 @@ func TestDockerProxyDirectoryProbes(t *testing.T) { var calls, removals atomic.Int32 visible := false cli := &probeClient{ - create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + create: func(_ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { calls.Add(1) <-release return mobyclient.ContainerCreateResult{ID: "probe"}, createErr @@ -538,23 +552,18 @@ func TestDockerProxyDirectoryProbes(t *testing.T) { results := make(chan result, 8) for range cap(results) { go func() { - _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) - results <- result{seen, err} + probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + results <- result{probe.seen, err} }() } synctest.Wait() assert.EqualValues(t, 1, calls.Load()) - ctx, cancel := context.WithCancel(t.Context()) - cancel() - _, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket) - require.ErrorIs(t, err, context.Canceled) - assert.False(t, seen) replacement, err := os.Stat(t.TempDir()) require.NoError(t, err) replaced := make(chan result, 1) go func() { - _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, replacement) - replaced <- result{seen, err} + probe, err := probeDockerProxyDirectory(t.Context(), cli, key, replacement) + replaced <- result{probe.seen, err} }() synctest.Wait() assert.EqualValues(t, 2, calls.Load()) @@ -574,11 +583,11 @@ func TestDockerProxyDirectoryProbes(t *testing.T) { assert.Zero(t, removals.Load()) } createErr, visible = nil, true - directory, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) require.NoError(t, err) - assert.True(t, seen) - assert.Equal(t, dir, directory.dir) - assert.Equal(t, dir, directory.daemonDir) + assert.True(t, probe.seen) + assert.Equal(t, dir, probe.directory.dir) + assert.Equal(t, dir, probe.directory.daemonDir) assert.EqualValues(t, 3, calls.Load()) entries, err := os.ReadDir(dir) require.NoError(t, err) @@ -586,7 +595,7 @@ func TestDockerProxyDirectoryProbes(t *testing.T) { }) }) } - t.Run("cancelled probe does not cancel another caller", func(t *testing.T) { + t.Run("cancelled caller leaves the probe to finish for the others", func(t *testing.T) { synctest.Test(t, func(t *testing.T) { dir := t.TempDir() socket, err := os.Stat(t.TempDir()) @@ -595,49 +604,105 @@ func TestDockerProxyDirectoryProbes(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() release := make(chan struct{}) - calls := 0 + var calls, removals atomic.Int32 cli := &probeClient{ - create: func(_opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { - calls++ - if calls == 1 { - <-release + create: func(ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + calls.Add(1) + <-release + return mobyclient.ContainerCreateResult{ID: "probe"}, ctx.Err() + }, + stat: func(_path string) error { return nil }, + remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { + removals.Add(1) + return mobyclient.ContainerRemoveResult{}, nil + }, + } + leader, follower, late := make(chan result, 1), make(chan result, 1), make(chan result, 1) + go func() { + probe, err := probeDockerProxyDirectory(ctx, cli, key, socket) + leader <- result{probe.seen, err} + }() + synctest.Wait() + go func() { + probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + follower <- result{probe.seen, err} + }() + synctest.Wait() + cancel() + leaderResult := <-leader + require.ErrorIs(t, leaderResult.err, context.Canceled) + assert.False(t, leaderResult.seen) + go func() { + probe, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) + late <- result{probe.seen, err} + }() + synctest.Wait() + close(release) + for _, results := range []chan result{follower, late} { + result := <-results + require.NoError(t, result.err) + assert.True(t, result.seen) + } + assert.EqualValues(t, 1, calls.Load()) + assert.EqualValues(t, 1, removals.Load()) + _, cached := dockerProxyProbes.Load(key) + assert.True(t, cached) + }) + }) + t.Run("caller joining a probe that ran out of time retries", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + dir := t.TempDir() + socket, err := os.Stat(t.TempDir()) + require.NoError(t, err) + var calls atomic.Int32 + cli := &probeClient{ + create: func(ctx context.Context, _opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + if calls.Add(1) == 1 { + <-ctx.Done() return mobyclient.ContainerCreateResult{}, ctx.Err() - } else { - return mobyclient.ContainerCreateResult{ID: "probe"}, nil } + return mobyclient.ContainerCreateResult{ID: "probe"}, nil }, stat: func(_path string) error { return nil }, remove: func(_ctx context.Context, _id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { return mobyclient.ContainerRemoveResult{}, nil }, } - leader, follower := make(chan result, 1), make(chan result, 1) - go func() { - _, seen, err := probeDockerProxyDirectory(ctx, cli, key, socket) - leader <- result{seen, err} - }() - synctest.Wait() - go func() { - _, seen, err := probeDockerProxyDirectory(t.Context(), cli, key, socket) - follower <- result{seen, err} - }() - synctest.Wait() - assert.Equal(t, 1, calls) - cancel() - close(release) - leaderResult, followerResult := <-leader, <-follower - require.ErrorIs(t, leaderResult.err, context.Canceled) - assert.False(t, leaderResult.seen) - require.NoError(t, followerResult.err) - assert.True(t, followerResult.seen) - assert.Equal(t, 2, calls) + probe, err := probeDockerProxyDirectory(t.Context(), cli, [3]string{"daemon", dir, dir}, socket) + require.NoError(t, err) + assert.True(t, probe.seen) + assert.EqualValues(t, 2, calls.Load()) }) }) } +func TestRemoveStaleProbeContainers(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + probeMount := container.MountPoint{Type: mount.TypeBind, Destination: "/gitea-runner-probe"} + writableMount := container.MountPoint{Type: mount.TypeBind, Destination: "/gitea-runner-probe", RW: true} + var removed []string + cli := &probeClient{ + containers: []container.Summary{ + {ID: "abandoned", State: container.StateCreated, Mounts: []container.MountPoint{probeMount}}, + {ID: "foreign", State: container.StateCreated, Mounts: []container.MountPoint{writableMount}}, + {ID: "started", State: container.StateExited, Mounts: []container.MountPoint{probeMount}}, + }, + remove: func(_ctx context.Context, id string, _opts mobyclient.ContainerRemoveOptions) (mobyclient.ContainerRemoveResult, error) { + removed = append(removed, id) + return mobyclient.ContainerRemoveResult{}, nil + }, + } + start := time.Now() + removeStaleProbeContainers(cli) + assert.Equal(t, []string{"abandoned"}, removed) + assert.GreaterOrEqual(t, time.Since(start), dockerProxyProbeWorkerTimeout+dockerProxyProbeTimeout) + }) +} + func TestDaemonSeesDir(t *testing.T) { dir := t.TempDir() markers := make(map[string]bool) + for _, testCase := range []struct { name string daemonDir string @@ -656,7 +721,7 @@ func TestDaemonSeesDir(t *testing.T) { removed := false var attemptedImages []string cli := &probeClient{ - create: func(opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + create: func(_ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { assert.Equal(t, []mount.Mount{{Type: mount.TypeBind, Source: testCase.daemonDir, Target: "/gitea-runner-probe", ReadOnly: true}}, opts.HostConfig.Mounts) attemptedImages = append(attemptedImages, opts.Config.Image) if opts.Config.Image == "removed-image" { @@ -692,6 +757,7 @@ func TestDaemonSeesDir(t *testing.T) { assert.Equal(t, []string{"removed-image", "probe-image"}, attemptedImages) assert.Equal(t, testCase.seen, seen) assert.Equal(t, testCase.removed, removed) + entries, err := os.ReadDir(dir) require.NoError(t, err) assert.Empty(t, entries) diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index fc1a4081..4633ecc4 100644 --- a/act/container/docker_run_test.go +++ b/act/container/docker_run_test.go @@ -980,7 +980,7 @@ func TestDockerCreateKeepsInputEntrypoint(t *testing.T) { RunnerOptions: "--entrypoint=runner-entrypoint", WorkflowOptions: "--entrypoint=" + optionEntrypoint + " --init", }, - cli: &probeClient{create: func(opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { + cli: &probeClient{create: func(_ctx context.Context, opts mobyclient.ContainerCreateOptions) (mobyclient.ContainerCreateResult, error) { if len(entrypoint) > 0 { assert.Equal(t, entrypoint, opts.Config.Entrypoint) } else {