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.