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 {