From 0d84ec6dab92b2a937ab5c7a2f75da6dbfd9ce8c Mon Sep 17 00:00:00 2001 From: silverwind <2021+silverwind@noreply.gitea.com> Date: Thu, 15 Oct 2026 14:15:41 +0000 Subject: [PATCH] fix: remove the job network even when container removal overruns (#1318) Job teardown ran every step on one shared minute. When removing the job container took longer, the network removal ran on an expired context and the network stayed behind until the orphan reclaim, which only takes networks older than `workdir_cleanup_age`, 24h by default. All that time it held a subnet of the daemon's address pool. The job and service containers are now removed in parallel, and the docker proxy sweep, the network and the volumes get a fresh budget after them. The idle reclaim takes unused networks and volumes once they are 10 minutes old, and network removal treats a network that is already gone as removed. Reviewed-on: https://gitea.com/gitea/runner/pulls/1318 Reviewed-by: bircni Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> --- README.md | 2 +- act/container/docker_network.go | 57 ++++++++++---------- act/container/docker_network_test.go | 25 +++++++-- act/runner/run_context.go | 41 +++++++++++--- act/runner/run_context_test.go | 42 ++++++++++----- internal/app/run/runner.go | 6 ++- internal/app/run/runner_idle_cleanup_test.go | 8 +-- internal/pkg/config/config.example.yaml | 6 +-- internal/pkg/config/config.go | 2 +- 9 files changed, 127 insertions(+), 62 deletions(-) diff --git a/README.md b/README.md index d1399d42..f854198d 100644 --- a/README.md +++ b/README.md @@ -368,7 +368,7 @@ While the runner is idle it cleans up after earlier jobs: - when `container.bind_workdir` is enabled, stale task workspace directories older than `runner.workdir_cleanup_age` are removed (default: `24h`; set `0` to disable) - only purely numeric subdirectories under `container.workdir_parent` are treated as task workspaces and may be removed - cleanup assumes `container.workdir_parent` is not shared across multiple runners -- on runners that use docker, per-job networks left behind by jobs the runner did not live to tear down are removed, identified by the `com.gitea.runner.uuid` label carrying this runner's uuid +- on runners that use docker, per-job networks and volumes left behind by jobs the runner did not live to tear down are removed once unused and 10 minutes old, identified by the `com.gitea.runner.uuid` label carrying this runner's uuid - on runners with `kubernetes` labels, job pods with that label older than `runner.workdir_cleanup_age` are removed the same way, together with their Secrets - cleanup runs every `runner.idle_cleanup_interval` (default: `10m`; set `0` to disable), and setting either knob to `0` disables all of the above diff --git a/act/container/docker_network.go b/act/container/docker_network.go index e872a713..ee072529 100644 --- a/act/container/docker_network.go +++ b/act/container/docker_network.go @@ -53,7 +53,9 @@ func removeOrphanNetworks(ctx context.Context, cli client.APIClient, runnerUUID, networks: for _, n := range networks.Items { result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{}) - if err != nil { + if cerrdefs.IsNotFound(err) { + continue + } else if err != nil { errs = append(errs, fmt.Errorf("failed to inspect network %s: %w", n.Name, err)) continue } @@ -71,7 +73,9 @@ networks: continue networks } } - if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil { + if _, err := cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); cerrdefs.IsNotFound(err) { + continue + } else if err != nil { errs = append(errs, fmt.Errorf("failed to remove network %s: %w", n.Name, err)) continue } @@ -203,36 +207,35 @@ func NewDockerNetworkRemoveExecutor(name string) common.Executor { return err } defer cli.Close() + return removeNetwork(ctx, cli, name) + } +} - // Make sure that all network of the specified name are removed - // cli.NetworkRemove refuses to remove a network if there are duplicates - networks, err := cli.NetworkList(ctx, client.NetworkListOptions{}) - if err != nil { +func removeNetwork(ctx context.Context, cli client.APIClient, name string) error { + networks, err := cli.NetworkList(ctx, client.NetworkListOptions{}) // NetworkRemove by name refuses duplicates, so remove each by ID + if err != nil { + return err + } + var errs []error + for _, n := range networks.Items { + if n.Name != name { + continue + } + result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{}) + if cerrdefs.IsNotFound(err) { + continue + } else if err != nil { return err } - // For Gitea, reduce log noise - // common.Logger(ctx).Debugf("%v", networks) - var errs []error - for _, n := range networks.Items { - if n.Name == name { - result, err := cli.NetworkInspect(ctx, n.ID, client.NetworkInspectOptions{}) - if err != nil { - return err - } - - // it holds a subnet out of the daemon's pool until something reclaims it - if len(result.Network.Containers) != 0 { - common.Logger(ctx).Warnf("Refusing to remove network %s because it still has active endpoints, the idle cleanup reclaims it once they are gone", name) - continue - } - if _, err = cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil { - errs = append(errs, fmt.Errorf("failed to remove network %s: %w", name, err)) - } - } + if len(result.Network.Containers) != 0 { + common.Logger(ctx).Warnf("Refusing to remove network %s because it still has active endpoints, the orphan cleanup reclaims it once they are gone", name) + continue + } + if _, err = cli.NetworkRemove(ctx, n.ID, client.NetworkRemoveOptions{}); err != nil && !cerrdefs.IsNotFound(err) { + errs = append(errs, fmt.Errorf("failed to remove network %s: %w", name, err)) } - - return errors.Join(errs...) } + return errors.Join(errs...) } // IsolatedCacheContainer returns the container holding addr, when jobs on jobNetwork cannot reach it there. diff --git a/act/container/docker_network_test.go b/act/container/docker_network_test.go index 5f8a67df..6c6853c3 100644 --- a/act/container/docker_network_test.go +++ b/act/container/docker_network_test.go @@ -116,7 +116,7 @@ func TestConnectAndDisconnectNetwork(t *testing.T) { // Of this runner's networks, only the ones nothing but its own cache container is attached to // and old enough to predate any job now starting are the runner's to reclaim. An unexpected // NetworkRemove fails the test on its own, since testify has no expectation to match it against. -func TestRemoveOrphanNetworks(t *testing.T) { +func TestRemoveNetworks(t *testing.T) { ctx := t.Context() cutoff := time.Date(2026, time.April, 29, 20, 0, 0, 0, time.UTC) client := &mockDockerClient{} @@ -128,11 +128,13 @@ func TestRemoveOrphanNetworks(t *testing.T) { {ID: "starting"}, {ID: "cached"}, {ID: "shared"}, + {ID: "gone"}, + {ID: "raced"}, }}, nil) client.On("NetworkInspect", ctx, "orphan", mobyclient.NetworkInspectOptions{}). - Return(mobyclient.NetworkInspectResult{}, nil) + Return(mobyclient.NetworkInspectResult{}, nil).Twice() client.On("NetworkInspect", ctx, "busy", mobyclient.NetworkInspectOptions{}). - Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"c": {}}}}, nil) + Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"c": {}}}}, nil).Twice() client.On("NetworkInspect", ctx, "starting", mobyclient.NetworkInspectOptions{}). Return(mobyclient.NetworkInspectResult{Network: network.Inspect{ Network: network.Network{Created: cutoff.Add(time.Second)}, @@ -142,13 +144,28 @@ func TestRemoveOrphanNetworks(t *testing.T) { Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}}}}, nil) client.On("NetworkInspect", ctx, "shared", mobyclient.NetworkInspectOptions{}). Return(mobyclient.NetworkInspectResult{Network: network.Inspect{Containers: map[string]network.EndpointResource{"0123456789abcdef": {}, "job": {}}}}, nil) + client.On("NetworkInspect", ctx, "gone", mobyclient.NetworkInspectOptions{}). + Return(mobyclient.NetworkInspectResult{}, cerrdefs.ErrNotFound).Twice() + client.On("NetworkInspect", ctx, "raced", mobyclient.NetworkInspectOptions{}). + Return(mobyclient.NetworkInspectResult{}, nil).Twice() + client.On("NetworkRemove", ctx, "raced", mobyclient.NetworkRemoveOptions{}). + Return(mobyclient.NetworkRemoveResult{}, cerrdefs.ErrNotFound).Twice() client.On("NetworkRemove", ctx, "orphan", mobyclient.NetworkRemoveOptions{}). - Return(mobyclient.NetworkRemoveResult{}, nil) + Return(mobyclient.NetworkRemoveResult{}, nil).Twice() client.On("NetworkDisconnect", ctx, "cached", mobyclient.NetworkDisconnectOptions{Container: "0123456789abcdef", Force: true}). Return(nil).Once() client.On("NetworkRemove", ctx, "cached", mobyclient.NetworkRemoveOptions{}). Return(mobyclient.NetworkRemoveResult{}, nil).Once() require.NoError(t, removeOrphanNetworks(ctx, client, "runner-1", "0123456789ab", cutoff)) + + client.On("NetworkList", ctx, mobyclient.NetworkListOptions{}).Return(mobyclient.NetworkListResult{Items: []network.Summary{ + {ID: "orphan", Name: "job"}, + {ID: "busy", Name: "job"}, + {ID: "gone", Name: "job"}, + {ID: "raced", Name: "job"}, + {ID: "shared", Name: "other"}, + }}, nil) + require.NoError(t, removeNetwork(ctx, client, "job")) client.AssertExpectations(t) } diff --git a/act/runner/run_context.go b/act/runner/run_context.go index e0afdbd8..f43224b6 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -22,6 +22,7 @@ import ( "path/filepath" "regexp" "runtime" + "runtime/debug" "slices" "strconv" "strings" @@ -719,31 +720,55 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet logger := common.Logger(ctx) rc.printServiceLogs(ctx) errs := []error{rc.closeDockerProxy(ctx)} + var containers []common.Executor if rc.JobContainer != nil { - errs = append(errs, rc.JobContainer.Remove()(ctx)) + containers = append(containers, rc.JobContainer.Remove()) } if len(rc.serviceContainers) > 0 { logger.Infof("Cleaning up services for job %s", rc.JobName) - errs = append(errs, rc.stopServiceContainers()(ctx)) + containers = append(containers, rc.stopServiceContainers()) } + errs = append(errs, runAll(ctx, containers)) + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), jobCleanupTimeout) + defer cancel() if !common.Dryrun(ctx) && !rc.Config.NoSweep && (preclean || rc.hadDockerProxy) { - errs = append(errs, container.RemoveDockerJobResources(ctx, rc.jobContainerName())) + errs = append(errs, container.RemoveDockerJobResources(ctx, rc.jobContainerName())) // after the containers, whose endpoints would block removing the job networks } + var rest []common.Executor if preclean || rc.deferVolumeCleanup == nil { - errs = append(errs, rc.cleanupJobVolumes(ctx)) + rest = append(rest, rc.cleanupJobVolumes) } else { rc.deferVolumeCleanup(rc.cleanupJobVolumes) } if createAndDeleteNetwork { logger.Infof("Cleaning up network for job %s, and network name is: %s", rc.JobName, networkName) - errs = append(errs, container.NewDockerNetworkDisconnectExecutor(networkName, rc.Config.CacheContainer). - IfBool(rc.Config.CacheContainer != "")(ctx)) - errs = append(errs, container.NewDockerNetworkRemoveExecutor(networkName)(ctx)) + rest = append(rest, func(ctx context.Context) error { + return errors.Join( + container.NewDockerNetworkDisconnectExecutor(networkName, rc.Config.CacheContainer).IfBool(rc.Config.CacheContainer != "")(ctx), + container.NewDockerNetworkRemoveExecutor(networkName)(ctx)) + }) } - return errors.Join(errs...) + return errors.Join(append(errs, runAll(ctx, rest))...) } } +func runAll(ctx context.Context, executors []common.Executor) error { + errs := make([]error, len(executors)) + var wg sync.WaitGroup + for i, executor := range executors { + wg.Go(func() { + defer func() { + if recovered := recover(); recovered != nil { + errs[i] = fmt.Errorf("panic: %v\n%s", recovered, debug.Stack()) + } + }() + errs[i] = executor(ctx) + }) + } + wg.Wait() + return errors.Join(errs...) +} + func (rc *RunContext) createJobVolumes(ctx context.Context, mounts map[string]string) error { runnerUUID := rc.Config.ContainerNetworkCreateOptions.RunnerUUID if runnerUUID == "" || common.Dryrun(ctx) { diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index a9ecabff..685f1251 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -20,6 +20,7 @@ import ( "runtime" "slices" "strings" + "sync" "testing" "testing/synctest" "time" @@ -775,10 +776,10 @@ func TestRemoveDockerActionImagesByTag(t *testing.T) { assert.Equal(t, []string{"act-action-dockeraction:latest-job"}, removed) } -func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { +func TestCleanupJobResources(t *testing.T) { t.Setenv("TMPDIR", "/tmp") proxyDir := t.TempDir() - for _, name := range []string{"synchronous", "proxy", "closed proxy", "kept proxy", "deferred", "preclean deferred", "canceled"} { + for _, name := range []string{"synchronous", "proxy", "closed proxy", "kept proxy", "deferred", "preclean deferred", "canceled proxy"} { t.Run(name, func(t *testing.T) { proxy, preclean, deferred := strings.HasSuffix(name, "proxy"), strings.HasPrefix(name, "preclean"), strings.HasSuffix(name, "deferred") removesLabelled := (proxy || preclean) && name != "kept proxy" @@ -787,8 +788,20 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { } jobError, removeError, closeError := errors.New("remove job"), errors.New("remove service"), errors.New("close service") job, service := &containerMock{}, &containerMock{} - job.On("Remove").Return(func(context.Context) error { return jobError }).Once() - service.On("Remove").Return(func(context.Context) error { return removeError }).Once() + removeAlongside := func(started, other chan struct{}, failure error) func(context.Context) error { + return func(context.Context) error { + close(started) + select { + case <-other: + case <-time.After(10 * time.Second): + t.Error("the containers are removed one after the other instead of alongside each other") + } + return failure + } + } + jobRemoving, serviceRemoving := make(chan struct{}), make(chan struct{}) + job.On("Remove").Return(removeAlongside(jobRemoving, serviceRemoving, jobError)).Once() + service.On("Remove").Return(removeAlongside(serviceRemoving, jobRemoving, removeError)).Once() service.On("Close").Return(func(context.Context) error { return closeError }).Once() rc := &RunContext{ Config: &Config{CacheContainer: "cache-container", NoSweep: name == "kept proxy"}, @@ -825,7 +838,7 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { } ctx, cancel := context.WithCancel(t.Context()) defer cancel() - if name == "canceled" { + if name == "canceled proxy" { cancel() } err := rc.cleanupJobResources("job-network", true, preclean)(ctx) @@ -839,14 +852,10 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { for _, failure := range []error{jobError, removeError, closeError} { require.ErrorIs(t, err, failure) } - if name == "canceled" { - require.ErrorIs(t, err, context.Canceled) - } else { - for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "POST /networks/job-network/disconnect", "GET /networks"} { - require.ErrorContains(t, err, operation) - } - assert.Equal(t, 2, volumeRemovals) + for _, operation := range []string{"DELETE /volumes/" + rc.jobContainerName(), "DELETE /volumes/" + rc.jobContainerName() + "-env", "POST /networks/job-network/disconnect", "GET /networks"} { + require.ErrorContains(t, err, operation) } + assert.Equal(t, 2, volumeRemovals) assert.Equal(t, removesLabelled, strings.Contains(err.Error(), "labelled GET /containers/json")) if removesLabelled { require.ErrorContains(t, err, "labelled GET /networks") @@ -1360,8 +1369,11 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { } { t.Run(testcase.name, func(t *testing.T) { var operations []string + var operationsMu sync.Mutex record := func(operation string, failure error) func(context.Context) error { return func(context.Context) error { + operationsMu.Lock() + defer operationsMu.Unlock() operations = append(operations, operation) return failure } @@ -1403,7 +1415,7 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { ctx := common.WithLogger(common.WithDryrun(t.Context(), true), logger) rc.ExprEval = rc.NewExpressionEvaluator(ctx) err = rc.startContainer().Then(rc.stopContainer()).Finally(rc.closeContainer())(ctx) - want := "job.Remove service.Remove service.Close service.Pull job.Pull" + want := "job.Remove service.Close service.Remove service.Pull job.Pull" switch { case testcase.pullError != nil: require.ErrorIs(t, err, testcase.pullError) @@ -1418,7 +1430,9 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { require.NoError(t, err) } } - assert.Equal(t, strings.Fields(want+" job.Remove service.Remove service.Close job.Close"), operations) + slices.Sort(operations[:3]) + slices.Sort(operations[len(operations)-4 : len(operations)-1]) + assert.Equal(t, strings.Fields(want+" job.Remove service.Close service.Remove job.Close"), operations) assert.Empty(t, rc.Masks) assert.Equal(t, testcase.pullError == nil && testcase.health != container.HealthUnhealthy, slices.ContainsFunc(hook.AllRecords(), func(entry logging.Record) bool { return strings.HasPrefix(entry.Message, "::group::Print service container logs") diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 85a0a697..30531e3d 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -157,6 +157,8 @@ var removeOrphanJobVolumes = container.RemoveOrphanJobVolumes var removeOrphanKubernetesResources = container.RemoveOrphanKubernetesResources +const orphanDockerResourceAge = 10 * time.Minute // spares the network and volumes of a job still starting its first container + // OnIdle performs lightweight maintenance during polling idle windows. // It runs synchronously on the poller goroutine; shouldRunIdleCleanup // throttles invocations to runner.idle_cleanup_interval so the impact on @@ -189,10 +191,12 @@ func (r *Runner) OnIdle(ctx context.Context) { } func (r *Runner) cleanupOrphanDockerResources(ctx context.Context) { + ctx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() if r.uuid == "" || (!r.requiresDocker() && !dockerReachable(ctx)) { return } - cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge) + cutoff := r.now().Add(-orphanDockerResourceAge) if err := removeOrphanNetworks(ctx, r.uuid, r.isolatedCacheContainer(), cutoff); err != nil { logging.Warnf("failed to clean up networks left behind by earlier jobs: %v", err) } diff --git a/internal/app/run/runner_idle_cleanup_test.go b/internal/app/run/runner_idle_cleanup_test.go index d92c774d..0de30b0f 100644 --- a/internal/app/run/runner_idle_cleanup_test.go +++ b/internal/app/run/runner_idle_cleanup_test.go @@ -317,7 +317,9 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { var sweptVolumes []string var sweptCutoff time.Time origRemoveOrphanNetworks := removeOrphanNetworks - removeOrphanNetworks = func(_ context.Context, runnerUUID, detach string, createdBefore time.Time) error { + removeOrphanNetworks = func(ctx context.Context, runnerUUID, detach string, createdBefore time.Time) error { + _, hasDeadline := ctx.Deadline() + assert.True(t, hasDeadline) swept = append(swept, runnerUUID) detached = append(detached, detach) sweptCutoff = createdBefore @@ -327,7 +329,7 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { origRemoveOrphanJobVolumes := removeOrphanJobVolumes removeOrphanJobVolumes = func(_ context.Context, runnerUUID string, createdBefore time.Time) error { sweptVolumes = append(sweptVolumes, runnerUUID) - assert.Equal(t, now.Add(-24*time.Hour), createdBefore) + assert.Equal(t, now.Add(-orphanDockerResourceAge), createdBefore) return nil } t.Cleanup(func() { removeOrphanJobVolumes = origRemoveOrphanJobVolumes }) @@ -350,7 +352,7 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { assert.Equal(t, []string{"runner-1"}, swept) assert.Equal(t, swept, sweptVolumes) // a network of a job starting during the pass is younger than this and so out of scope - assert.Equal(t, now.Add(-24*time.Hour), sweptCutoff) + assert.Equal(t, now.Add(-orphanDockerResourceAge), sweptCutoff) // a host-only runner has no daemon to sweep origDockerReachable := dockerReachable diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 1b74c098..63ff95b8 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -61,9 +61,9 @@ runner: #workdir_cleanup_age: 24h # Cadence for the idle cleanup pass. Besides the directories above, on runners that use # docker it removes the per-job networks and volumes of jobs this runner did not live to - # tear down, which would otherwise hold a subnet of the daemon address pool and the job's - # workspace contents until the host is rebuilt. On runners with kubernetes labels it - # removes such jobs' pods and Secrets. + # tear down once unused and 10 minutes old, which would otherwise hold a subnet of the + # daemon address pool and the job's workspace contents until the host is rebuilt. On + # runners with kubernetes labels it removes such jobs' pods and Secrets. #idle_cleanup_interval: 10m # The base interval for periodic log flush to the Gitea instance. # Logs may be sent earlier if the buffer reaches log_report_batch_size diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 8bbc67e1..c8063bb2 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -71,7 +71,7 @@ type Runner struct { FetchTimeout time.Duration `yaml:"fetch_timeout"` // FetchTimeout specifies the timeout duration for fetching resources. FetchInterval time.Duration `yaml:"fetch_interval"` // FetchInterval specifies the interval duration for fetching resources. FetchIntervalMax time.Duration `yaml:"fetch_interval_max"` // FetchIntervalMax specifies the maximum backoff interval when idle. - WorkdirCleanupAge time.Duration `yaml:"workdir_cleanup_age"` // WorkdirCleanupAge removes stale bind-workdir task directories, orphaned host-mode scratch dirs and orphaned docker and Kubernetes job resources older than this duration during idle cleanup. + WorkdirCleanupAge time.Duration `yaml:"workdir_cleanup_age"` // WorkdirCleanupAge removes stale bind-workdir task directories, orphaned host-mode scratch dirs and orphaned Kubernetes job resources older than this duration during idle cleanup. IdleCleanupInterval time.Duration `yaml:"idle_cleanup_interval"` // IdleCleanupInterval runs the idle cleanup (stale directories, orphaned docker networks and volumes, orphaned job pods) periodically while the runner is idle. Set to 0 to disable cleanup cadence. LogReportInterval time.Duration `yaml:"log_report_interval"` // LogReportInterval specifies the base interval for periodic log flush. LogReportMaxLatency time.Duration `yaml:"log_report_max_latency"` // LogReportMaxLatency specifies the max time a log row can wait before being sent.