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 <bircni@icloud.com>
Co-authored-by: silverwind <me@silverwind.io>
This commit is contained in:
silverwind
2026-09-10 19:00:06 +00:00
committed by bircni
parent 498282caaa
commit 54978255f5
11 changed files with 591 additions and 254 deletions
+21 -85
View File
@@ -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.
+34 -23
View File
@@ -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))
}
+8
View File
@@ -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
}
+55 -25
View File
@@ -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
}
}
+52 -21
View File
@@ -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)
}
}
+251 -86
View File
@@ -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) {
+31 -9
View File
@@ -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,
@@ -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)
}
+115
View File
@@ -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{}
+4 -3
View File
@@ -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
+2 -2
View File
@@ -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.