From 54978255f52855269e2509ec4a58b00df7dc3ea6 Mon Sep 17 00:00:00 2001 From: silverwind Date: Thu, 10 Sep 2026 19:00:06 +0000 Subject: [PATCH] perf: cut redundant work out of job setup and teardown (#1218) Implement speedups to job start and shutdown. - Create the job container while services are still becoming healthy, and poll their health at a flat one second instead of a 2s to 32s doubling backoff - Pull each service image once instead of twice, and fetch a warm action cache once instead of twice - Report the job result before reclaiming its volumes, and reap volumes stranded by a runner that died mid-job | Step | Scenario | Before | After | | --- | --- | --- | --- | | Complete job | Large workspace volume | 4.2s | 0.4s | | Set up job | One service, 2s health interval | 7.08s | 3.26s | | Set up job | Two cached actions from github.com | 1.81s | 1.34s | | Set up job | Two cached actions from gitea.com | 2.42s | 1.95s | | Set up job | Two actions, cold action cache | 6.62s | unchanged | | Set up job | Minimal job, no services or actions | 0.62s | unchanged | Assisted-by: Claude Code:Opus 5 Reviewed-on: https://gitea.com/gitea/runner/pulls/1218 Reviewed-by: bircni Co-authored-by: silverwind --- act/common/git/git.go | 106 ++---- act/common/git/git_test.go | 57 ++-- act/container/docker_stub.go | 8 + act/container/docker_volume.go | 80 +++-- act/runner/run_context.go | 73 ++-- act/runner/run_context_test.go | 337 ++++++++++++++----- internal/app/run/runner.go | 40 ++- internal/app/run/runner_idle_cleanup_test.go | 18 + internal/app/run/runner_test.go | 115 +++++++ internal/pkg/config/config.example.yaml | 7 +- internal/pkg/config/config.go | 4 +- 11 files changed, 591 insertions(+), 254 deletions(-) diff --git a/act/common/git/git.go b/act/common/git/git.go index febb5106..1f1885d9 100644 --- a/act/common/git/git.go +++ b/act/common/git/git.go @@ -325,33 +325,6 @@ func CloneIfRequired(ctx context.Context, refName plumbing.ReferenceName, input return r, false, nil } -func gitOptions(token string) (fetchOptions git.FetchOptions, pullOptions git.PullOptions) { - fetchOptions.RefSpecs = []config.RefSpec{"refs/*:refs/*", "HEAD:refs/heads/HEAD"} - fetchOptions.Force = true - pullOptions.Force = true - - if token != "" { - auth := &http.BasicAuth{ - Username: "token", - Password: token, - } - fetchOptions.Auth = auth - pullOptions.Auth = auth - } - - return fetchOptions, pullOptions -} - -// staleRefreshErr reports why a failed refresh must abort: the resolve and -// checkout that follow are local and succeed on a cancelled context, which -// would hand back the cached revision as if it were fresh. -func staleRefreshErr(ctx context.Context, err error) error { - if err == nil || errors.Is(err, git.NoErrAlreadyUpToDate) { - return nil - } - return ctx.Err() -} - // NewGitCloneExecutor creates an executor to clone git repos func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { return func(ctx context.Context) error { @@ -373,18 +346,21 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { isOfflineMode := input.OfflineMode - // fetch latest changes - fetchOptions, pullOptions := gitOptions(input.Token) - - if input.InsecureSkipTLS { // For Gitea - fetchOptions.InsecureSkipTLS = true - pullOptions.InsecureSkipTLS = true + fetchOptions := git.FetchOptions{ + RefSpecs: []config.RefSpec{"refs/*:refs/*", "HEAD:refs/heads/HEAD", "refs/heads/*:refs/remotes/origin/*"}, + Force: true, + InsecureSkipTLS: input.InsecureSkipTLS, + } + if input.Token != "" { + fetchOptions.Auth = &http.BasicAuth{ + Username: "token", + Password: input.Token, + } } // Action clones only ever need the tip commit, so keep a shallow cache cheap on update at depth 1 regardless of its original depth // Turning action_shallow_clone off does not convert an existing shallow cache; evict it for a full clone. - shallow := isShallow(r) - if shallow { + if isShallow(r) { fetchOptions.Depth = 1 if spec, ok := shallowFetchRefSpec(r, input.Ref); ok { fetchOptions.RefSpecs = []config.RefSpec{spec} @@ -392,7 +368,9 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { } // A just-cloned ref is as current as a fetch would make it, and a commit hash never moves. - _, _, present := refRevision(r, input.Ref) + // TODO: revalidate a mutable ref with a conditional archive request instead, once every + // supported Gitea sends an ETag for them: https://github.com/go-gitea/gitea/pull/39289 + _, present := refRevision(r, input.Ref) refresh := !isOfflineMode && (!present || (reused && !plumbing.IsHash(input.Ref))) if refresh { @@ -415,7 +393,7 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { } } - rev, refType, _ := refRevision(r, input.Ref) + rev, _ = refRevision(r, input.Ref) if hash, err = r.ResolveRevision(rev); err != nil { logger.Errorf("Unable to resolve %s: %v", input.Ref, err) @@ -427,47 +405,13 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { return err } - // If the hash resolved doesn't match the ref provided in a workflow then we're - // using a branch or tag ref, not a sha - // - // Repos on disk point to commit hashes, and need to checkout input.Ref before - // we try and pull down any changes - if hash.String() != input.Ref && refType == "branch" { - logger.Debugf("Provided ref is not a sha. Checking out branch before pulling changes") - sourceRef := plumbing.ReferenceName(path.Join("refs", "remotes", "origin", input.Ref)) - if err = w.Checkout(&git.CheckoutOptions{ - Branch: sourceRef, - Force: true, - }); err != nil { - logger.Errorf("Unable to checkout %s: %v", sourceRef, err) - return err - } - } - reusedMsg := "" - - switch { - case refresh && !shallow: - // In shallow mode the depth-limited fetch above already advanced the ref. - if err = w.PullContext(ctx, &pullOptions); err != nil && !errors.Is(err, git.NoErrAlreadyUpToDate) { - logger.Debugf("Unable to pull %s: %v", refName, err) - } - if err := staleRefreshErr(ctx, err); err != nil { - return err - } - case isOfflineMode && reused: + if isOfflineMode && reused { reusedMsg = " (reused in offline mode)" } logger.Debugf("Cloned %s to %s%s", input.URL, input.Dir, reusedMsg) - if hash.String() != input.Ref && refType == "branch" { - logger.Debugf("Provided ref is not a sha. Updating branch ref after pull") - if hash, err = r.ResolveRevision(rev); err != nil { - logger.Errorf("Unable to resolve %s: %v", input.Ref, err) - return err - } - } if err = w.Checkout(&git.CheckoutOptions{ Hash: *hash, Force: true, @@ -476,14 +420,6 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) common.Executor { return err } - if err = w.Reset(&git.ResetOptions{ - Mode: git.HardReset, - Commit: *hash, - }); err != nil { - logger.Errorf("Unable to reset to %s: %v", hash.String(), err) - return err - } - logger.Debugf("Checked out %s", input.Ref) return nil } @@ -562,22 +498,22 @@ func pinnedRefSpec(sha string) config.RefSpec { } // refRevision picks the revision to check out and reports whether it resolves locally. -func refRevision(r *git.Repository, ref string) (plumbing.Revision, string, bool) { +func refRevision(r *git.Repository, ref string) (plumbing.Revision, bool) { if plumbing.IsHash(ref) { // git ignores a ref named as 40 hex digits, so a full hash always denotes the commit itself. _, err := r.CommitObject(plumbing.NewHash(ref)) - return plumbing.Revision(ref), "sha", err == nil + return plumbing.Revision(ref), err == nil } if _, err := r.Tag(ref); err == nil { - return plumbing.Revision(path.Join("refs", "tags", ref)), "tag", true + return plumbing.Revision(path.Join("refs", "tags", ref)), true } remoteRef := plumbing.ReferenceName(path.Join("refs", "remotes", "origin", ref)) if _, err := r.Reference(remoteRef, false); err == nil { - return plumbing.Revision(remoteRef), "branch", true + return plumbing.Revision(remoteRef), true } rev := plumbing.Revision(ref) _, err := r.ResolveRevision(rev) - return rev, "sha", err == nil + return rev, err == nil } // isShallow reports whether the local repository was cloned with a limited depth. diff --git a/act/common/git/git_test.go b/act/common/git/git_test.go index 8cfbee5e..17cc7bb6 100644 --- a/act/common/git/git_test.go +++ b/act/common/git/git_test.go @@ -6,7 +6,6 @@ package git import ( "context" - "errors" "fmt" "net/http" "net/http/httptest" @@ -524,30 +523,51 @@ func TestGitCloneExecutorShallow(t *testing.T) { }) } -func TestGitCloneExecutorColdCloneSkipsRefresh(t *testing.T) { - remoteDir := t.TempDir() - require.NoError(t, gitCmd("init", "--bare", "--initial-branch=main", remoteDir)) +func TestGitCloneExecutorTransportSessions(t *testing.T) { workDir := t.TempDir() - require.NoError(t, gitCmd("clone", remoteDir, workDir)) - require.NoError(t, gitCmd("-C", workDir, "checkout", "-b", "main")) + require.NoError(t, gitCmd("init", "--initial-branch=main", workDir)) require.NoError(t, gitCmd("-C", workDir, "commit", "--allow-empty", "-m", "c1")) require.NoError(t, gitCmd("-C", workDir, "tag", "v1")) - require.NoError(t, gitCmd("-C", workDir, "push", "-u", "origin", "main")) - require.NoError(t, gitCmd("-C", workDir, "push", "origin", "v1")) + require.NoError(t, gitCmd("-C", workDir, "tag", "-a", "v2", "-m", "v2")) for name, tt := range map[string]struct { Ref string Depth int }{ - "shallow branch": {"main", 1}, - "full clone tag": {"v1", 0}, + "shallow branch": {"main", 1}, + "full clone branch": {"main", 0}, + "full lightweight tag": {"v1", 0}, + "full annotated tag": {"v2", 0}, } { t.Run(name, func(t *testing.T) { counter := installCountingTransport(t) - require.NoError(t, NewGitCloneExecutor(NewGitCloneExecutorInput{ - URL: remoteDir, Ref: tt.Ref, Dir: t.TempDir(), Depth: tt.Depth, - })(t.Context())) - assert.Equal(t, int64(1), counter.sessions.Load()) + dir := t.TempDir() + clone := NewGitCloneExecutor(NewGitCloneExecutorInput{ + URL: workDir, Ref: tt.Ref, Dir: dir, Depth: tt.Depth, + }) + require.NoError(t, clone(t.Context())) + assert.Equal(t, int64(1), counter.sessions.Swap(0), "cold clone") + assert.Equal(t, gitRevParse(t, workDir, tt.Ref+"^{commit}"), gitRevParse(t, dir, "HEAD")) + + require.NoError(t, clone(t.Context())) + assert.Equal(t, int64(1), counter.sessions.Swap(0), "unchanged warm cache") + + require.NoError(t, os.WriteFile(filepath.Join(workDir, "action.yml"), []byte(name), 0o644)) + require.NoError(t, gitCmd("-C", workDir, "add", "action.yml")) + require.NoError(t, gitCmd("-C", workDir, "commit", "-m", name)) + require.NoError(t, gitCmd("-C", workDir, "tag", "--force", "v1")) + require.NoError(t, gitCmd("-C", workDir, "tag", "--force", "-a", "v2", "-m", "v2")) + + require.NoError(t, os.WriteFile(filepath.Join(dir, "action.yml"), []byte("staged"), 0o644)) + require.NoError(t, gitCmd("-C", dir, "add", "action.yml")) + require.NoError(t, os.WriteFile(filepath.Join(dir, "action.yml"), []byte("unstaged"), 0o644)) + + require.NoError(t, clone(t.Context())) + assert.Equal(t, int64(1), counter.sessions.Load(), "updated warm cache") + assert.Equal(t, gitRevParse(t, workDir, "HEAD"), gitRevParse(t, dir, "HEAD")) + status, err := exec.Command("git", "-C", dir, "status", "--porcelain").Output() + require.NoError(t, err) + assert.Empty(t, string(status)) }) } } @@ -736,12 +756,3 @@ func TestNewGitCloneExecutorFetchHonoursContext(t *testing.T) { t.Fatal("fetch ignored context cancellation") } } - -func TestStaleRefreshErr(t *testing.T) { - ctx, cancel := context.WithCancel(t.Context()) - require.NoError(t, staleRefreshErr(ctx, errors.New("remote hung up"))) - - cancel() - require.ErrorIs(t, staleRefreshErr(ctx, errors.New("remote hung up")), context.Canceled) - require.NoError(t, staleRefreshErr(ctx, gogit.NoErrAlreadyUpToDate)) -} diff --git a/act/container/docker_stub.go b/act/container/docker_stub.go index afeb4649..552b43b2 100644 --- a/act/container/docker_stub.go +++ b/act/container/docker_stub.go @@ -89,3 +89,11 @@ func NewDockerNetworkRemoveExecutor(name string) common.Executor { func RemoveOrphanNetworks(ctx context.Context, runnerUUID string, createdBefore time.Time) error { return nil } + +func CreateJobVolumes(ctx context.Context, runnerUUID string, volumeNames []string) error { + return nil +} + +func RemoveOrphanJobVolumes(ctx context.Context, runnerUUID string, createdBefore time.Time) error { + return nil +} diff --git a/act/container/docker_volume.go b/act/container/docker_volume.go index 30046ea7..8e97e318 100644 --- a/act/container/docker_volume.go +++ b/act/container/docker_volume.go @@ -8,40 +8,67 @@ package container import ( "context" + "errors" + "fmt" + "time" "gitea.com/gitea/runner/act/common" + cerrdefs "github.com/containerd/errdefs" "github.com/moby/moby/client" ) -func NewDockerVolumeRemoveExecutor(volumeName string, force bool) common.Executor { - return func(ctx context.Context) error { - cli, err := GetDockerClient(ctx) - if err != nil { - return err - } - defer cli.Close() - - list, err := cli.VolumeList(ctx, client.VolumeListOptions{}) - if err != nil { - return err - } - - for _, vol := range list.Items { - if vol.Name == volumeName { - return removeExecutor(volumeName, force)(ctx) - } - } - - // Volume not found - do nothing - return nil +func CreateJobVolumes(ctx context.Context, runnerUUID string, volumeNames []string) error { + cli, err := GetDockerClient(ctx) + if err != nil { + return err } + defer cli.Close() + + for _, volumeName := range volumeNames { + if _, err := cli.VolumeCreate(ctx, client.VolumeCreateOptions{ + Name: volumeName, + Labels: runnerLabels(runnerUUID), + }); err != nil { + return err + } + } + return nil } -func removeExecutor(volume string, force bool) common.Executor { +func RemoveOrphanJobVolumes(ctx context.Context, runnerUUID string, createdBefore time.Time) error { + cli, err := GetDockerClient(ctx) + if err != nil { + return fmt.Errorf("failed to connect to the docker daemon: %w", err) + } + defer cli.Close() + + volumes, err := cli.VolumeList(ctx, client.VolumeListOptions{ + Filters: make(client.Filters).Add("label", runnerUUIDLabel+"="+runnerUUID).Add("dangling", "true"), + }) + if err != nil { + return err + } + + var errs []error + for _, item := range volumes.Items { + created, err := time.Parse(time.RFC3339, item.CreatedAt) + // an unreadable or recent timestamp may belong to a job that is still starting up + if err != nil || created.After(createdBefore) || item.Labels[runnerUUIDLabel] != runnerUUID { + continue + } + if _, err := cli.VolumeRemove(ctx, item.Name, client.VolumeRemoveOptions{}); err != nil && !cerrdefs.IsNotFound(err) { + errs = append(errs, fmt.Errorf("failed to remove volume %s: %w", item.Name, err)) + continue + } + common.Logger(ctx).Infof("removed docker volume %s left behind by an earlier job", item.Name) + } + return errors.Join(errs...) +} + +func NewDockerVolumeRemoveExecutor(volumeName string, force bool) common.Executor { return func(ctx context.Context) error { - logger := common.Logger(ctx) - logger.Debugf("docker volume rm %s", volume) + common.Logger(ctx).Debugf("docker volume rm %s", volumeName) if common.Dryrun(ctx) { return nil @@ -53,7 +80,10 @@ func removeExecutor(volume string, force bool) common.Executor { } defer cli.Close() - _, err = cli.VolumeRemove(ctx, volume, client.VolumeRemoveOptions{Force: force}) + _, err = cli.VolumeRemove(ctx, volumeName, client.VolumeRemoveOptions{Force: force}) + if cerrdefs.IsNotFound(err) { // already gone is the outcome we wanted + return nil + } return err } } diff --git a/act/runner/run_context.go b/act/runner/run_context.go index f8c51451..b059d81d 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -68,6 +68,7 @@ type RunContext struct { actionInputs map[string]any // inputs of the composite action this runs, nil for a job Masks []string cleanUpJobContainer common.Executor + deferVolumeCleanup func(common.Executor) caller *caller // job calling this RunContext (reusable workflows) workflowCallInputs map[string]any // the caller's with:, resolved once by resolveWorkflowCall workflowCallSecrets map[string]string // the caller's secrets:, resolved once by resolveWorkflowCall @@ -98,6 +99,7 @@ type RunContext struct { hasBash *bool // memoized implicit-shell probe, only set on the top-level RunContext jobNetworkName string dockerProxy *container.DockerProxy + hadDockerProxy bool // stepEnv is a copy of the running step's environment, so that workflow commands parsed out // of the container's output can be judged against it. Written by runStepExecutor and read on // the log-writer goroutine, hence unsecureCommandMu, which also guards unsecureCommandErr. @@ -459,8 +461,15 @@ func printStartJobContainerGroup(ctx context.Context, image, name, network strin // newContainer is a variable so tests can substitute a container that needs no Docker daemon. var newContainer = container.NewContainer +type jobVolumeCleanupKey struct{} + +func WithJobVolumeCleanup(ctx context.Context, deferCleanup func(common.Executor)) context.Context { + return context.WithValue(ctx, jobVolumeCleanupKey{}, deferCleanup) +} + func (rc *RunContext) startJobContainer() common.Executor { return func(ctx context.Context) error { + rc.deferVolumeCleanup, _ = ctx.Value(jobVolumeCleanupKey{}).(func(common.Executor)) logger := common.Logger(ctx) image := rc.platformImage logWriter := rc.commandLogWriter(ctx) @@ -484,7 +493,7 @@ func (rc *RunContext) startJobContainer() common.Executor { // if using service containers, will create a new network for the containers. // and it will be removed after at last. networkName, createAndDeleteNetwork := rc.networkNameForGitea() - rc.cleanUpJobContainer = rc.cleanupJobResources(networkName, createAndDeleteNetwork) + rc.cleanUpJobContainer = rc.cleanupJobResources(networkName, createAndDeleteNetwork, false) // add service containers for serviceID, spec := range rc.Run.Job().Services { @@ -610,10 +619,10 @@ func (rc *RunContext) startJobContainer() common.Executor { } defer printStartJobContainerGroup(ctx, image, name, networkName)() if err := common.NewPipelineExecutor( - rc.stopJobContainer(), + rc.cleanupJobResources(networkName, createAndDeleteNetwork, true), rc.pullServicesImages(rc.Config.ForcePull), rc.JobContainer.Pull(rc.Config.ForcePull), - ).Finally(rc.closeContainer())(ctx); err != nil { + )(ctx); err != nil { return err } rc.startDockerProxy(ctx) @@ -629,7 +638,7 @@ func (rc *RunContext) startJobContainer() common.Executor { IfBool(createAndDeleteNetwork), rc.startServiceContainers(), rc.reportUnstartedServices(), - rc.waitForServiceContainers(), + func(ctx context.Context) error { return rc.createJobVolumes(ctx, containerInput.Mounts) }, rc.JobContainer.Create(rc.Config.ContainerCapAdd, rc.Config.ContainerCapDrop), rc.JobContainer.Start(false), rc.captureJobContainerInfo(), @@ -642,6 +651,7 @@ func (rc *RunContext) startJobContainer() common.Executor { Mode: 0o666, Body: "", }), + rc.waitForServiceContainers(), )(ctx) } } @@ -654,7 +664,7 @@ func (rc *RunContext) commandLogWriter(ctx context.Context) io.Writer { }) } -func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNetwork bool) common.Executor { +func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNetwork, preclean bool) common.Executor { return func(ctx context.Context) error { logger := common.Logger(ctx) errs := []error{rc.closeDockerProxy(ctx)} @@ -665,14 +675,13 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet logger.Infof("Cleaning up services for job %s", rc.JobName) errs = append(errs, rc.stopServiceContainers()(ctx)) } - if !common.Dryrun(ctx) { + if !common.Dryrun(ctx) && (preclean || rc.hadDockerProxy) { errs = append(errs, container.RemoveDockerJobResources(ctx, rc.jobContainerName())) } - if rc.JobContainer != nil { - name := rc.jobContainerName() - errs = append(errs, - container.NewDockerVolumeRemoveExecutor(name, false)(ctx), - container.NewDockerVolumeRemoveExecutor(name+"-env", false)(ctx)) + if preclean || rc.deferVolumeCleanup == nil { + errs = append(errs, rc.cleanupJobVolumes(ctx)) + } else { + rc.deferVolumeCleanup(rc.cleanupJobVolumes) } if createAndDeleteNetwork { logger.Infof("Cleaning up network for job %s, and network name is: %s", rc.JobName, networkName) @@ -682,10 +691,34 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet } } +func (rc *RunContext) createJobVolumes(ctx context.Context, mounts map[string]string) error { + runnerUUID := rc.Config.ContainerNetworkCreateOptions.RunnerUUID + if runnerUUID == "" || common.Dryrun(ctx) { + return nil + } + name := rc.jobContainerName() + volumeNames := []string{name, name + "-env"} + if _, ok := mounts[name]; !ok { // the workspace is a bind mount, the env volume is always ours + volumeNames = volumeNames[1:] + } + return container.CreateJobVolumes(ctx, runnerUUID, volumeNames) +} + +func (rc *RunContext) cleanupJobVolumes(ctx context.Context) error { + if rc.JobContainer == nil { + return nil + } + name := rc.jobContainerName() + return errors.Join( + container.NewDockerVolumeRemoveExecutor(name, false)(ctx), + container.NewDockerVolumeRemoveExecutor(name+"-env", false)(ctx)) +} + func (rc *RunContext) closeDockerProxy(ctx context.Context) error { if rc.dockerProxy == nil { return nil } + rc.hadDockerProxy = true err := rc.dockerProxy.Close(ctx) rc.dockerProxy = nil if err != nil { @@ -777,7 +810,6 @@ func (rc *RunContext) startServiceContainers() common.Executor { execs := []common.Executor{} for _, svc := range rc.serviceContainers { execs = append(execs, common.NewPipelineExecutor( - svc.container.Pull(false), svc.container.Create(rc.Config.ContainerCapAdd, rc.Config.ContainerCapDrop), svc.container.Start(false), )) @@ -803,12 +835,9 @@ func (rc *RunContext) stopServiceContainers() common.Executor { } } -const ( - defaultServiceReadyTimeout = 5 * time.Minute - serviceReadyPollMax = 32 * time.Second -) +const defaultServiceReadyTimeout = 5 * time.Minute -var serviceReadyPollInterval = 2 * time.Second // a variable so tests need not wait +const serviceReadyPollInterval = time.Second // reportUnstartedServices logs a service that did not start. The steps that need it // report it better than the runner can, so the job carries on. @@ -868,7 +897,7 @@ func (rc *RunContext) waitForServiceContainers() common.Executor { // ready at once and one that exited is left to the steps that need it. func (svc *serviceContainer) waitUntilHealthy(ctx context.Context, timeout time.Duration) error { rawLogger := common.Logger(ctx).WithField(rawOutputField, true) - interval := serviceReadyPollInterval + loggedStarting := false ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() @@ -898,12 +927,14 @@ func (svc *serviceContainer) waitUntilHealthy(ctx context.Context, timeout time. return nil } - rawLogger.Infof("%s service is starting, waiting %d seconds before checking again.", svc.name, int(interval.Seconds())) + if !loggedStarting { + rawLogger.Infof("%s service is starting.", svc.name) + loggedStarting = true + } select { case <-ctx.Done(): // reported at the top of the loop - case <-time.After(interval): + case <-time.After(serviceReadyPollInterval): } - interval = min(interval*2, serviceReadyPollMax) } } diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index 823543ac..7d559b89 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -7,15 +7,21 @@ package runner import ( "bytes" "context" + "encoding/json/v2" "errors" "fmt" "io" + "maps" + "net" "net/http" "net/http/httptest" "os" + "path/filepath" "runtime" + "slices" "strings" "testing" + "testing/synctest" "time" "gitea.com/gitea/runner/act/common" @@ -24,6 +30,7 @@ import ( "gitea.dev/actionslib/pkg/exprparser" "gitea.dev/actionslib/pkg/model" "github.com/docker/cli/cli/compose/loader" + "github.com/moby/moby/api/types/volume" log "github.com/sirupsen/logrus" assert "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -240,8 +247,20 @@ func (fakeContainer) Inspect(context.Context) (*container.Info, error) { func (fakeContainer) DumpLogs(context.Context) error { return nil } -// startJobContainerInputs runs startJobContainer against fakeContainer and returns the -// inputs it built, one per container. +func fakeDockerDaemon(t *testing.T, handler http.HandlerFunc) { + t.Helper() + daemon := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "application/json") + if strings.HasSuffix(request.URL.Path, "/_ping") { + writer.Header().Set("API-Version", "1.47") + } else { + handler(writer, request) + } + })) + t.Cleanup(daemon.Close) + t.Setenv("DOCKER_HOST", daemon.URL) +} + func startJobContainerInputs(t *testing.T, workflowYAML string, cfg *Config) []*container.NewContainerInput { t.Helper() workflow, err := model.ReadWorkflow(strings.NewReader(workflowYAML)) @@ -282,6 +301,8 @@ func startJobContainerInputs(t *testing.T, workflowYAML string, cfg *Config) []* _, _ = io.WriteString(w, "[]") case strings.HasSuffix(r.URL.Path, "/volumes"): _, _ = io.WriteString(w, `{"Volumes":[]}`) + case r.Method == http.MethodDelete && strings.Contains(r.URL.Path, "/volumes/"): + w.WriteHeader(http.StatusNoContent) case strings.HasSuffix(r.URL.Path, "/info"): _, _ = io.WriteString(w, `{"Architecture":"amd64","OSType":"linux"}`) default: @@ -647,39 +668,142 @@ func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) { serviceContainers: []*serviceContainer{{name: "svc", container: service}}, } - err := rc.cleanupJobResources("external-network", false)(common.WithDryrun(t.Context(), true)) + err := rc.cleanupJobResources("external-network", false, true)(common.WithDryrun(t.Context(), true)) require.NoError(t, err) service.AssertExpectations(t) } -// cleanup used to bail out on a previous step's error and on a cancelled context -func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { - t.Setenv("DOCKER_HOST", "unix:///nonexistent.sock") - - removeError, closeError := errors.New("remove service"), errors.New("close service") - jobContainer := &containerMock{} - jobContainer.On("Remove").Return(func(context.Context) error { return errors.New("removal failed") }).Once() - service := &containerMock{} - service.On("Remove").Return(func(context.Context) error { return removeError }).Once() - service.On("Close").Return(func(context.Context) error { return closeError }).Once() - +func TestCleanupJobVolumesReapsAbandonedDeferredCleanup(t *testing.T) { + now := time.Date(2026, time.September, 10, 12, 0, 0, 0, time.UTC) rc := &RunContext{ - Name: "job", - Config: &Config{}, - Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}, JobID: "job"}, - JobContainer: jobContainer, - serviceContainers: []*serviceContainer{{name: "svc", container: service}}, + Config: &Config{ContainerNetworkCreateOptions: container.NewDockerNetworkCreateExecutorInput{RunnerUUID: "runner-1"}}, + Run: &model.Run{Workflow: &model.Workflow{Name: "workflow"}, JobID: "job"}, + JobContainer: fakeContainer{}, } + volumes := map[string]volume.Volume{} + fakeDockerDaemon(t, func(writer http.ResponseWriter, request *http.Request) { + path := strings.TrimPrefix(request.URL.Path, "/v1.47") + switch { + case path == "/volumes/create": + options := volume.Volume{CreatedAt: now.Format(time.RFC3339)} + assert.NoError(t, json.UnmarshalRead(request.Body, &options)) + volumes[options.Name] = options + assert.NoError(t, json.MarshalWrite(writer, volumes[options.Name])) + case path == "/volumes": + assert.JSONEq(t, `{"label":{"com.gitea.runner.uuid=runner-1":true},"dangling":{"true":true}}`, request.URL.Query().Get("filters")) + assert.NoError(t, json.MarshalWrite(writer, map[string]any{"Volumes": slices.Collect(maps.Values(volumes))})) + case request.Method == http.MethodDelete: + name := strings.TrimPrefix(path, "/volumes/") + assert.NotContains(t, []string{"1", "true"}, request.URL.Query().Get("force")) + if name == "became-active" || name == "remove-failed" { + writer.WriteHeader(http.StatusConflict) + _, _ = fmt.Fprintf(writer, `{"message":%q}`, name) + return + } + delete(volumes, name) + writer.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected Docker request: %s %s", request.Method, request.URL) + } + }) + name := rc.jobContainerName() + require.NoError(t, rc.createJobVolumes(t.Context(), map[string]string{name: "/workspace", name + "-env": "/var/run/act", "shared-cache": "/cache"})) + assert.ElementsMatch(t, []string{name, name + "-env"}, slices.Collect(maps.Keys(volumes))) + rc.deferVolumeCleanup = func(common.Executor) {} + require.NoError(t, rc.cleanupJobResources("", false, false)(t.Context())) + require.Len(t, volumes, 2) + for _, name := range []string{"became-active", "remove-failed"} { + volumes[name] = volume.Volume{Name: name, Labels: volumes[rc.jobContainerName()].Labels, CreatedAt: now.Format(time.RFC3339)} + } + volumes["foreign"] = volume.Volume{Name: "foreign", Labels: map[string]string{"com.gitea.runner.uuid": "runner-2"}, CreatedAt: now.Format(time.RFC3339)} + volumes["fresh"] = volume.Volume{Name: "fresh", Labels: volumes[name].Labels, CreatedAt: now.Add(48 * time.Hour).Format(time.RFC3339)} + volumes["unknown-age"] = volume.Volume{Name: "unknown-age", Labels: volumes[name].Labels} + err := container.RemoveOrphanJobVolumes(t.Context(), "runner-1", now.Add(24*time.Hour)) + require.ErrorContains(t, err, "became-active") + require.ErrorContains(t, err, "remove-failed") + assert.ElementsMatch(t, []string{"became-active", "remove-failed", "foreign", "fresh", "unknown-age"}, slices.Collect(maps.Keys(volumes))) +} - ctx, cancel := context.WithCancel(context.Background()) - cancel() - err := rc.cleanupJobResources("job-network", true)(ctx) - require.ErrorContains(t, err, "removal failed") - require.ErrorIs(t, err, removeError) - require.ErrorIs(t, err, closeError) - require.ErrorIs(t, err, context.Canceled) - jobContainer.AssertExpectations(t) - service.AssertExpectations(t) +func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { + t.Setenv("TMPDIR", "/tmp") + proxyDir := t.TempDir() + for _, name := range []string{"synchronous", "proxy", "closed proxy", "deferred", "preclean deferred", "canceled"} { + t.Run(name, func(t *testing.T) { + proxy, preclean, deferred := strings.HasSuffix(name, "proxy"), strings.HasPrefix(name, "preclean"), strings.HasSuffix(name, "deferred") + if proxy && runtime.GOOS == "windows" { + t.Skip("Unix socket ownership is unavailable on Windows") + } + 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() + service.On("Close").Return(func(context.Context) error { return closeError }).Once() + rc := &RunContext{ + Config: &Config{}, + Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}, JobID: "job"}, + JobContainer: job, + serviceContainers: []*serviceContainer{{name: "svc", container: service}}, + } + var volumeCleanup []common.Executor + if deferred { + rc.deferVolumeCleanup = func(cleanup common.Executor) { volumeCleanup = append(volumeCleanup, cleanup) } + } + volumeRemovals := 0 + fakeDockerDaemon(t, func(writer http.ResponseWriter, request *http.Request) { + if request.Method == http.MethodDelete { + volumeRemovals++ + } + operation := request.Method + " " + strings.TrimPrefix(request.URL.Path, "/v1.47") + if request.URL.Query().Has("filters") { + operation = "labelled " + operation + } + writer.WriteHeader(http.StatusInternalServerError) + _, _ = fmt.Fprintf(writer, `{"message":%q}`, operation) + }) + if proxy { + listener, err := net.Listen("unix", filepath.Join(proxyDir, "d.sock")) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, listener.Close()) }) + rc.dockerProxy, err = container.StartDockerProxy(listener.Addr().String(), proxyDir, rc.jobContainerName()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, rc.closeDockerProxy(context.Background())) }) + if name == "closed proxy" { + require.NoError(t, rc.closeDockerProxy(t.Context())) + } + } + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + if name == "canceled" { + cancel() + } + err := rc.cleanupJobResources("job-network", true, preclean)(ctx) + if deferred && !preclean { + require.Len(t, volumeCleanup, 1) + assert.Zero(t, volumeRemovals) + err = errors.Join(err, volumeCleanup[0](ctx)) + } else { + assert.Empty(t, volumeCleanup) + } + 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", "GET /networks"} { + require.ErrorContains(t, err, operation) + } + assert.Equal(t, 2, volumeRemovals) + } + assert.Equal(t, proxy || preclean, strings.Contains(err.Error(), "labelled GET /containers/json")) + if proxy || preclean { + require.ErrorContains(t, err, "labelled GET /networks") + require.ErrorContains(t, err, "labelled GET /volumes") + } + assert.Nil(t, rc.dockerProxy) + assert.Equal(t, proxy, rc.hadDockerProxy) + }) + } } // TestInterpolateOutputsIsPerMatrixCombo guards the matrix-output fix: combinations share one @@ -1089,65 +1213,96 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { type ctxKey string const sentinel = ctxKey("sentinel") - // the fresh context is cancelled via defer on return, so capture state inside the stub - type capture struct { - calls int - err error - sentinel any - cancel context.CancelFunc - } - newRC := func(c *capture) *RunContext { - return &RunContext{ - JobName: "job", - cleanUpJobContainer: func(ctx context.Context) error { - c.calls++ - if c.cancel != nil { - c.cancel() - } + for name, canceled := range map[string]bool{"cancellation during cleanup": false, "already canceled": true} { + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.WithValue(t.Context(), sentinel, "v")) + defer cancel() + if canceled { + cancel() + } + calls := 0 + (&RunContext{cleanUpJobContainer: func(ctx context.Context) error { + calls++ + cancel() deadline, ok := ctx.Deadline() require.True(t, ok) assert.WithinDuration(t, time.Now().Add(time.Minute), deadline, time.Second) - c.err = ctx.Err() - c.sentinel = ctx.Value(sentinel) + require.NoError(t, ctx.Err()) + assert.Equal(t, "v", ctx.Value(sentinel)) return nil - }, - } + }}).cleanupFailedStart(ctx) + assert.Equal(t, 1, calls) + }) } - t.Run("detaches teardown from cancellation during cleanup", func(t *testing.T) { - ctx, cancel := context.WithCancel(context.WithValue(context.Background(), sentinel, "v")) - defer cancel() - c := capture{cancel: cancel} + for _, testcase := range []struct { + name, health string + pullError error + }{ + {name: "healthy", health: container.HealthHealthy}, + {name: "unhealthy", health: container.HealthUnhealthy}, + {name: "pull failure", pullError: errors.New("pull failed")}, + } { + t.Run(testcase.name, func(t *testing.T) { + var operations []string + record := func(operation string, failure error) func(context.Context) error { + return func(context.Context) error { + operations = append(operations, operation) + return failure + } + } + job, service := &containerMock{}, &containerMock{} + for name, instance := range map[string]*containerMock{"job": job, "service": service} { + instance.On("Pull", true).Return(record(name+".Pull", map[string]error{"job": testcase.pullError}[name])) + instance.On("Create", mock.Anything, mock.Anything).Return(record(name+".Create", nil)) + instance.On("Start", false).Return(record(name+".Start", nil)) + instance.On("Remove").Return(record(name+".Remove", nil)) + instance.On("Close").Return(record(name+".Close", nil)) + } + job.On("Copy", mock.Anything, mock.Anything).Return(record("job.Copy", nil)) + job.On("Inspect", mock.Anything).Return(&container.Info{ID: "job-id"}, nil). + Run(func(mock.Arguments) { operations = append(operations, "job.Inspect") }) + for _, health := range []string{container.HealthStarting, testcase.health} { + service.On("Inspect", mock.Anything).Return(&container.Info{State: "running", Health: health}, nil). + Run(func(mock.Arguments) { operations = append(operations, "service.Inspect") }).Once() + } + service.On("DumpLogs", mock.Anything).Return(nil). + Run(func(mock.Arguments) { operations = append(operations, "service.DumpLogs") }) + origNewContainer := newContainer + newContainer = func(input *container.NewContainerInput) container.ExecutionsEnvironment { + return map[string]*containerMock{"postgres:latest": service, "node:20": job}[input.Image] + } + t.Cleanup(func() { newContainer = origNewContainer }) + workflow, err := model.ReadWorkflow(strings.NewReader("jobs: {job: {services: {postgres: {image: postgres:latest}}}}")) + require.NoError(t, err) + rc := &RunContext{ + Config: &Config{ForcePull: true, ContainerNetworkMode: "host", Workdir: "/workspace"}, + Run: &model.Run{JobID: "job", Workflow: workflow}, + platformImage: "node:20", + } + ctx := common.WithDryrun(t.Context(), true) + 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" + if testcase.pullError != nil { + require.ErrorIs(t, err, testcase.pullError) + } else { + want += " service.Create service.Start service.Inspect job.Create job.Start job.Inspect job.Copy service.Inspect" + if testcase.health == container.HealthUnhealthy { + require.ErrorContains(t, err, "the service 'postgres' is unhealthy") + want += " service.DumpLogs" + } else { + require.NoError(t, err) + } + } + assert.Equal(t, strings.Fields(want+" job.Remove service.Remove service.Close job.Close"), operations) + }) + } - newRC(&c).cleanupFailedStart(ctx) - - assert.Equal(t, 1, c.calls) - require.NoError(t, c.err) - assert.Equal(t, "v", c.sentinel) - }) - - t.Run("falls back to a fresh context when the input is done", func(t *testing.T) { - var c capture - ctx, cancel := context.WithCancel(context.WithValue(context.Background(), sentinel, "v")) - cancel() - - newRC(&c).cleanupFailedStart(ctx) - - assert.Equal(t, 1, c.calls) - require.NoError(t, c.err) - assert.Equal(t, "v", c.sentinel) - }) - - t.Run("no-op when there is nothing to clean up", func(t *testing.T) { - assert.NotPanics(t, func() { (&RunContext{}).cleanupFailedStart(context.Background()) }) - }) + (&RunContext{}).cleanupFailedStart(t.Context()) } func TestWaitForServiceContainers(t *testing.T) { - origInterval := serviceReadyPollInterval - serviceReadyPollInterval = time.Millisecond - defer func() { serviceReadyPollInterval = origInterval }() - newRunContext := func(timeout time.Duration, services ...*serviceContainer) *RunContext { return &RunContext{ Config: &Config{ServiceReadyTimeout: timeout}, @@ -1165,16 +1320,26 @@ func TestWaitForServiceContainers(t *testing.T) { service.AssertExpectations(t) }) - t.Run("waits while a service is still starting", func(t *testing.T) { - service := &containerMock{} - service.On("Inspect", mock.Anything). - Return(&container.Info{ID: "id", State: "running", Health: container.HealthStarting}, nil).Twice() - service.On("Inspect", mock.Anything). - Return(&container.Info{ID: "id", State: "running", Health: container.HealthHealthy}, nil).Once() + t.Run("polls at a fixed interval and logs starting only once", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var pollTimes []time.Duration + started := time.Now() + service := &containerMock{} + service.On("Inspect", mock.Anything). + Run(func(_ mock.Arguments) { pollTimes = append(pollTimes, time.Since(started)) }). + Return(&container.Info{ID: "id", State: "running", Health: container.HealthStarting}, nil).Twice() + service.On("Inspect", mock.Anything). + Run(func(_ mock.Arguments) { pollTimes = append(pollTimes, time.Since(started)) }). + Return(&container.Info{ID: "id", State: "running", Health: container.HealthHealthy}, nil).Once() - rc := newRunContext(0, &serviceContainer{name: "postgres", container: service}) - require.NoError(t, rc.waitForServiceContainers()(context.Background())) - service.AssertExpectations(t) + var output bytes.Buffer + logger := log.New() + logger.SetOutput(&output) + require.NoError(t, newRunContext(0, &serviceContainer{name: "postgres", container: service}).waitForServiceContainers()(common.WithLogger(t.Context(), logger.WithFields(nil)))) + assert.Equal(t, []time.Duration{0, time.Second, 2 * time.Second}, pollTimes) + assert.Equal(t, 1, strings.Count(output.String(), "postgres service is starting.")) + assert.Contains(t, output.String(), "postgres service is healthy.") + }) }) t.Run("fails with the probe output when a service is unhealthy", func(t *testing.T) { diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 2fc86715..30930ff7 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -146,6 +146,8 @@ func (r *Runner) Close() error { // removeOrphanNetworks is a variable so tests can substitute one that needs no Docker daemon. var removeOrphanNetworks = container.RemoveOrphanNetworks +var removeOrphanJobVolumes = container.RemoveOrphanJobVolumes + // 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 @@ -167,21 +169,20 @@ func (r *Runner) OnIdle(ctx context.Context) { if hostRoot := filepath.FromSlash(r.cfg.Host.WorkdirParent); hostRoot != "" { r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir) } - r.cleanupOrphanNetworks(ctx) + r.cleanupOrphanDockerResources(ctx) } -// cleanupOrphanNetworks reclaims the per-job networks of jobs this runner did not live to -// tear down. A labelled network with no containers on it is finished with, and as for the -// directories above, a task beginning during the pass is safe because the cutoff keeps a -// network it has created but not yet attached a container to out of scope. -func (r *Runner) cleanupOrphanNetworks(ctx context.Context) { - if r.uuid == "" || !r.requiresDocker() { +func (r *Runner) cleanupOrphanDockerResources(ctx context.Context) { + if r.uuid == "" || (!r.requiresDocker() && !dockerReachable(ctx)) { return } cutoff := r.now().Add(-r.cfg.Runner.WorkdirCleanupAge) if err := removeOrphanNetworks(ctx, r.uuid, cutoff); err != nil { log.Warnf("failed to clean up networks left behind by earlier jobs: %v", err) } + if err := removeOrphanJobVolumes(ctx, r.uuid, cutoff); err != nil { + log.Warnf("failed to clean up volumes left behind by earlier jobs: %v", err) + } } func (r *Runner) shouldRunIdleCleanup() bool { @@ -287,6 +288,7 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error { defer r.runningTasks.Delete(task.Id) r.runningCount.Add(1) + defer r.runningCount.Add(-1) start := time.Now() @@ -294,15 +296,25 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error { defer cancel() // A proxy URL may carry credentials, and every job is given it; keep them out of the log. reporter := report.NewReporter(ctx, cancel, r.client, task, r.cfg, proxyPasswords()...) + var volumeCleanup []common.Executor + var volumeCleanupMu sync.Mutex + if r.cfg.Runner.PostTaskScript == "" { + ctx = runner.WithJobVolumeCleanup(ctx, func(cleanup common.Executor) { + volumeCleanupMu.Lock() + defer volumeCleanupMu.Unlock() + volumeCleanup = append(volumeCleanup, cleanup) + }) + } var runErr error defer func() { - r.runningCount.Add(-1) - lastWords := "" if runErr != nil { lastWords = runErr.Error() } _ = reporter.Close(lastWords) + if err := cleanupJobVolumes(ctx, volumeCleanup); err != nil { + log.Warnf("task %d volume cleanup after reporting: %v", task.Id, err) + } metrics.JobDuration.Observe(time.Since(start).Seconds()) metrics.JobsTotal.WithLabelValues(metrics.ResultToStatusLabel(reporter.Result())).Inc() @@ -313,6 +325,16 @@ func (r *Runner) Run(ctx context.Context, task *runnerv1.Task) error { return nil } +func cleanupJobVolumes(ctx context.Context, cleanups []common.Executor) error { + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Minute) + defer cancel() + var errs []error + for _, cleanup := range cleanups { + errs = append(errs, cleanup(ctx)) + } + return errors.Join(errs...) +} + func (r *Runner) cloneEnvs() map[string]string { // Reserve space for the per-task keys injected by run(): // ACTIONS_ID_TOKEN_REQUEST_URL, ACTIONS_ID_TOKEN_REQUEST_TOKEN, ACTIONS_RUNTIME_TOKEN, diff --git a/internal/app/run/runner_idle_cleanup_test.go b/internal/app/run/runner_idle_cleanup_test.go index c19fb796..e5451f04 100644 --- a/internal/app/run/runner_idle_cleanup_test.go +++ b/internal/app/run/runner_idle_cleanup_test.go @@ -311,6 +311,7 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { } var swept []string + var sweptVolumes []string var sweptCutoff time.Time origRemoveOrphanNetworks := removeOrphanNetworks removeOrphanNetworks = func(_ context.Context, runnerUUID string, createdBefore time.Time) error { @@ -319,15 +320,32 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { return nil } t.Cleanup(func() { removeOrphanNetworks = origRemoveOrphanNetworks }) + 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) + return nil + } + t.Cleanup(func() { removeOrphanJobVolumes = origRemoveOrphanJobVolumes }) r := &Runner{uuid: "runner-1", cfg: cfg, now: func() time.Time { return now }} r.OnIdle(context.Background()) 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) // a host-only runner has no daemon to sweep + origDockerReachable := dockerReachable + dockerReachable = func(context.Context) bool { return false } + t.Cleanup(func() { dockerReachable = origDockerReachable }) hostOnly := &Runner{uuid: "runner-2", cfg: &config.Config{Runner: cfg.Runner}, now: func() time.Time { return now }} hostOnly.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1"}, swept) + assert.Equal(t, swept, sweptVolumes) + + dockerReachable = func(context.Context) bool { return true } + now = now.Add(time.Minute) + hostOnly.OnIdle(context.Background()) + assert.Equal(t, []string{"runner-1", "runner-2"}, sweptVolumes) } diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go index 4eebbf87..1c40f342 100644 --- a/internal/app/run/runner_test.go +++ b/internal/app/run/runner_test.go @@ -5,10 +5,18 @@ package run import ( "context" + "io" "net/http" + "net/http/httptest" + "os" + "path/filepath" + "runtime" "strings" + "sync/atomic" "testing" + "time" + "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/runner" clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" @@ -73,6 +81,113 @@ func TestRunnerRunningCountAndNullLogger(t *testing.T) { require.NotNil(t, logger.Out) } +func TestRunnerReclaimsVolumesAfterReporting(t *testing.T) { + for _, mode := range []string{"deferred", "post-task script"} { + t.Run(mode, func(t *testing.T) { + cfg := &config.Config{ + Cache: config.Cache{Enabled: new(false)}, + Runner: config.Runner{Timeout: time.Minute, LogReportInterval: time.Minute, StateReportInterval: time.Minute}, + Container: config.Container{Network: "host", DockerHost: "-"}, + } + if mode == "post-task script" { + if runtime.GOOS == "windows" { + t.Skip("uses a POSIX script") + } + cfg.Runner.PostTaskScript = filepath.Join(t.TempDir(), "post-task.sh") + require.NoError(t, os.WriteFile(cfg.Runner.PostTaskScript, []byte("#!/bin/sh\n: > \"$0.done\"\n"), 0o700)) + } + cli := clientmocks.NewClient(t) + cli.AddressValue = "https://gitea.example/" + r := NewRunner(cfg, &config.Registration{UUID: "runner-1", Labels: []string{"ubuntu:docker://node:20"}}, cli) + var reported atomic.Bool + var removed atomic.Int64 + cli.On("UpdateLog", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect.Request[runnerv1.UpdateLogRequest]) (*connect.Response[runnerv1.UpdateLogResponse], error) { + assert.False(t, reported.Load()) + return connect.NewResponse(&runnerv1.UpdateLogResponse{AckIndex: req.Msg.Index + int64(len(req.Msg.Rows))}), nil + }) + cli.On("UpdateTask", mock.Anything, mock.Anything).Return(func(_ context.Context, req *connect.Request[runnerv1.UpdateTaskRequest]) (*connect.Response[runnerv1.UpdateTaskResponse], error) { + if req.Msg.State.Result != runnerv1.Result_RESULT_UNSPECIFIED { + assert.Equal(t, runnerv1.Result_RESULT_SUCCESS, req.Msg.State.Result) + assert.Equal(t, int64(1), r.RunningCount()) + if mode == "post-task script" { + assert.FileExists(t, cfg.Runner.PostTaskScript+".done") + } + reported.Store(true) + } + return connect.NewResponse(&runnerv1.UpdateTaskResponse{State: req.Msg.State}), nil + }) + var volumesCreated atomic.Bool + daemon := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { + writer.Header().Set("Content-Type", "application/json") + writer.Header().Set("API-Version", "1.47") + path := strings.TrimPrefix(request.URL.Path, "/v1.47") + if response, ok := map[string]string{ + "/_ping": "OK", + "/info": `{"Architecture":"amd64","OSType":"linux"}`, + "/containers/json": "[]", + "/networks": "[]", + "/volumes": `{"Volumes":[]}`, + "/containers/create": `{"Id":"job-id"}`, + "/containers/job-id/json": `{"Id":"job-id","Config":{},"State":{"Status":"running"}}`, + }[path]; ok { + _, _ = io.WriteString(writer, response) + return + } + switch { + case strings.HasPrefix(path, "/images/"): + _, _ = io.WriteString(writer, `{"Id":"image-id","Config":{},"Os":"linux","Architecture":"amd64"}`) + case path == "/volumes/create": + volumesCreated.Store(true) + _, _ = io.WriteString(writer, "{}") + case strings.HasSuffix(path, "/exec"): + _, _ = io.WriteString(writer, `{"Id":"exec-id"}`) + case strings.HasPrefix(path, "/exec/"): + _, _ = io.WriteString(writer, `{"Running":false,"ExitCode":0}`) + case request.Method == http.MethodDelete || strings.HasSuffix(path, "/start") || strings.HasSuffix(path, "/kill") || strings.HasSuffix(path, "/archive"): + if request.Method == http.MethodDelete && strings.HasPrefix(path, "/volumes/") && volumesCreated.Load() { + assert.Equal(t, mode != "post-task script", reported.Load()) + if mode == "post-task script" { + assert.NoFileExists(t, cfg.Runner.PostTaskScript+".done") + } + assert.Equal(t, int64(1), r.RunningCount()) + removed.Add(1) + } + writer.WriteHeader(http.StatusNoContent) + default: + t.Errorf("unexpected Docker request: %s %s", request.Method, request.URL) + http.NotFound(writer, request) + } + })) + t.Cleanup(daemon.Close) + t.Setenv("DOCKER_HOST", daemon.URL) + require.NoError(t, r.Run(t.Context(), &runnerv1.Task{ + Context: &structpb.Struct{}, + WorkflowPayload: []byte("jobs:\n job:\n runs-on: ubuntu\n steps:\n - run: exit 0\n if: false\n"), + })) + assert.True(t, reported.Load()) + assert.Equal(t, int64(2), removed.Load()) + assert.Zero(t, r.RunningCount()) + }) + } +} + +func TestCleanupJobVolumesJoinsErrorsAfterCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + err := cleanupJobVolumes(ctx, []common.Executor{ + func(ctx context.Context) error { + require.NoError(t, ctx.Err()) + deadline, ok := ctx.Deadline() + assert.True(t, ok) + assert.InDelta(t, time.Minute.Seconds(), time.Until(deadline).Seconds(), 1) + return io.EOF + }, + func(context.Context) error { return io.ErrClosedPipe }, + }) + require.ErrorIs(t, err, io.EOF) + require.ErrorIs(t, err, io.ErrClosedPipe) +} + func TestNewRunnerInitializesLabelsAndEnvironment(t *testing.T) { cacheEnabled := false cfg := &config.Config{} diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 4cd8c0fa..03130012 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -53,11 +53,12 @@ runner: # scratch directories (left behind when a host cleanup delete stalls) older than # this duration. Setting either workdir_cleanup_age or idle_cleanup_interval to 0 # (or any non-positive value) disables stale-directory cleanup entirely, along with - # the docker network cleanup below. + # the docker network and volume cleanup below. #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 of jobs this runner did not live to tear down, - # which would otherwise hold a subnet of the daemon address pool until the host is rebuilt. + # 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. #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 62c859c5..94c5439a 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -63,8 +63,8 @@ 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 and orphaned host-mode scratch dirs older than this duration during idle cleanup. - IdleCleanupInterval time.Duration `yaml:"idle_cleanup_interval"` // IdleCleanupInterval runs the idle cleanup (stale directories and orphaned docker networks) periodically while the runner is idle. Set to 0 to disable cleanup cadence. + WorkdirCleanupAge time.Duration `yaml:"workdir_cleanup_age"` // WorkdirCleanupAge removes stale bind-workdir task directories, orphaned host-mode scratch dirs and orphaned docker job resources older than this duration during idle cleanup. + IdleCleanupInterval time.Duration `yaml:"idle_cleanup_interval"` // IdleCleanupInterval runs the idle cleanup (stale directories and orphaned docker networks and volumes) 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. LogReportBatchSize int `yaml:"log_report_batch_size"` // LogReportBatchSize triggers immediate log flush when buffer reaches this size.