From 68547886a58094325220c309220abaef016ff4a1 Mon Sep 17 00:00:00 2001 From: bircni Date: Wed, 5 Aug 2026 19:46:43 +0000 Subject: [PATCH] feat: wait for healthy services and fill the `job` context (#1107) Service containers were started and then left alone, so a job's first step could run while a database was still starting up. The runner now waits for every service whose image or `options` declare a healthcheck, as GitHub does. An unhealthy service fails the job with its container log, one that never becomes healthy fails it after `container.service_ready_timeout` (default `5m`, negative disables the wait), and one that exits without a healthcheck only gets its log and a warning. The started containers also fill the `job` context, whose fields existed but were never populated: `job.container.{id,network}` and `job.services..{id,network,ports}`. `ports` is keyed by the plain container port, so `job.services.postgres.ports['5432']` resolves to the host port Docker picked. --------- Co-authored-by: silverwind Reviewed-on: https://gitea.com/gitea/runner/pulls/1107 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: bircni --- README.md | 20 +++ act/container/container_types.go | 30 ++++ act/container/docker_run.go | 160 ++++++++++++------ act/container/docker_run_test.go | 65 +++++++- act/container/host_environment.go | 8 + act/model/job_context.go | 22 ++- act/runner/container_mock_test.go | 11 ++ act/runner/run_context.go | 213 ++++++++++++++++++++++-- act/runner/run_context_test.go | 205 ++++++++++++++++++++++- act/runner/runner.go | 1 + act/runner/testdata/services/push.yaml | 6 + go.mod | 2 +- internal/app/run/runner.go | 1 + internal/pkg/config/config.example.yaml | 3 + internal/pkg/config/config.go | 1 + 15 files changed, 670 insertions(+), 78 deletions(-) diff --git a/README.md b/README.md index 5bb08692..a42f9ae4 100644 --- a/README.md +++ b/README.md @@ -230,6 +230,26 @@ Whenever the resulting labels differ from the ones in the registration file, the > **Note:** A runner that only exposes `host` labels still needs access to a Docker daemon (e.g. a mounted `/var/run/docker.sock`) whenever a job uses a `docker://` action or a service container. `host` labels only change where the job's own steps run; container-based steps and actions are still executed with Docker. +#### Service containers + +A job's `services` are started before its steps run. When a service's image or its `options` declare a healthcheck, the runner waits for it to report healthy, so a workflow does not have to poll for its own services: + +```yaml +services: + postgres: + image: postgres:17 + options: >- + --health-cmd pg_isready + --health-interval 5s + --health-retries 10 +``` + +A service that reports unhealthy fails the job right away, with its container log. One that never becomes healthy fails it after `container.service_ready_timeout` (default `5m`, negative disables the wait). A service that exits without declaring a healthcheck only gets its log and a warning. + +A job in a container reaches a service by its id on the job network, on the port the service listens on, for example `psql -h postgres -p 5432`. The started containers also fill the `job` context: `job.container.{id,network}` and `job.services..{id,network,ports}`, where `ports` maps a container port to the host port Docker published it on, for the services that publish one. + +Unlike GitHub, a job whose steps run on the host (a `host` label without `container:`) starts no service containers, so `job.services` and `job.container` stay empty. Give such a job a `container:` when it needs services. + #### Proxy Set these variables in the runner's environment, with systemd `Environment=`, `docker run -e`, or Kubernetes `env:`: diff --git a/act/container/container_types.go b/act/container/container_types.go index 4e17b516..f4df9700 100644 --- a/act/container/container_types.go +++ b/act/container/container_types.go @@ -6,12 +6,14 @@ package container import ( "context" + "errors" "fmt" "io" "gitea.com/gitea/runner/act/common" "github.com/docker/go-connections/nat" + "github.com/moby/moby/api/types/container" ) // ExitCodeError reports a non-zero process exit code from a container command. @@ -57,6 +59,32 @@ type FileEntry struct { Body string } +// Container and healthcheck states, as plain strings so a caller of Info needs no docker +// SDK of its own. +const ( + StateRunning = string(container.StateRunning) + + HealthNone = string(container.NoHealthcheck) + HealthStarting = string(container.Starting) + HealthHealthy = string(container.Healthy) + HealthUnhealthy = string(container.Unhealthy) +) + +// ErrContainerNotFound reports a container the daemon no longer knows. Its text is a +// fragment, missingContainerError composes it into the message every operation shares. +var ErrContainerNotFound = errors.New("does not exist") + +// Info is a snapshot of a container, as of one inspect. +type Info struct { + ID string + State string // the docker container state: "created", "running", "exited", ... + ExitCode int + Health string // one of the Health* constants + // HealthOutput is the last healthcheck probe's output. + HealthOutput string + Ports map[string]string // container port ("5432") to the host port it is published on +} + // Container for managing docker run containers type Container interface { Create(capAdd, capDrop []string) common.Executor @@ -65,6 +93,8 @@ type Container interface { CopyTarStream(ctx context.Context, destPath string, tarStream io.Reader) error CopyDir(destPath, srcPath string, useGitIgnore bool) common.Executor GetContainerArchive(ctx context.Context, srcPath string) (io.ReadCloser, error) + Inspect(ctx context.Context) (*Info, error) + DumpLogs(ctx context.Context) error Pull(forcePull bool) common.Executor Start(attach bool) common.Executor Exec(command []string, env map[string]string, user, workdir string) common.Executor diff --git a/act/container/docker_run.go b/act/container/docker_run.go index bbd160e6..1db35b67 100644 --- a/act/container/docker_run.go +++ b/act/container/docker_run.go @@ -198,6 +198,109 @@ func (cr *containerReference) GetContainerArchive(ctx context.Context, srcPath s return result.Content, nil } +// Inspect resolves the container by name when its id is not known yet. One the daemon no +// longer knows is reported as ErrContainerNotFound. +func (cr *containerReference) Inspect(ctx context.Context) (*Info, error) { + if common.Dryrun(ctx) { + return &Info{Health: HealthNone, Ports: map[string]string{}}, nil + } + if err := cr.connect()(ctx); err != nil { + return nil, err + } + if cr.id == "" { // a known id is trusted, find() would spend a call validating it + if err := cr.find()(ctx); err != nil { + return nil, err + } + } + if cr.id == "" { + return nil, cr.missingContainerError("inspect it") + } + + result, err := cr.cli.ContainerInspect(ctx, cr.id, client.ContainerInspectOptions{}) + if cerrdefs.IsNotFound(err) { + return nil, cr.missingContainerError("inspect it") + } else if err != nil { + return nil, err + } + return containerInfoFromInspect(result.Container), nil +} + +// DumpLogs copies the container's log so far to its output writers. +func (cr *containerReference) DumpLogs(ctx context.Context) error { + if common.Dryrun(ctx) { + return nil + } + if err := cr.connect()(ctx); err != nil { + return err + } + if cr.id == "" { + return cr.missingContainerError("read its logs") + } + + logs, err := cr.cli.ContainerLogs(ctx, cr.id, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true}) + if err != nil { + return err + } + defer logs.Close() + return cr.copyOutput(logs) +} + +// copyOutput writes a container stream to the writers the container was created with, +// demultiplexing it unless the container has a TTY, which sends a single raw stream. +func (cr *containerReference) copyOutput(stream io.Reader) error { + outWriter := cr.input.Stdout + if outWriter == nil { + outWriter = os.Stdout + } + errWriter := cr.input.Stderr + if errWriter == nil { + errWriter = os.Stderr + } + + var err error + if !cr.input.AllocatePTY || os.Getenv("NORAW") != "" { + _, err = stdcopy.StdCopy(outWriter, errWriter, stream) + } else { + _, err = io.Copy(outWriter, stream) + } + // Flush any buffered, not-yet-newline-terminated trailing line so the final line of + // the output is not lost when it is not newline-terminated. + common.FlushWriter(outWriter) + common.FlushWriter(errWriter) + return err +} + +func containerInfoFromInspect(inspect container.InspectResponse) *Info { + info := &Info{ + ID: inspect.ID, + Health: HealthNone, + Ports: map[string]string{}, // an empty map, never null, in the expression context + } + + if state := inspect.State; state != nil { + info.State = string(state.Status) + info.ExitCode = state.ExitCode + if health := state.Health; health != nil { + info.Health = string(health.Status) + if len(health.Log) > 0 { + info.HealthOutput = strings.TrimSpace(health.Log[len(health.Log)-1].Output) + } + } + } + + if settings := inspect.NetworkSettings; settings != nil { + for port, bindings := range settings.Ports { + for _, binding := range bindings { // the last binding wins, a port maps to one host port + if binding.HostPort != "" { + info.Ports[port.Port()] = binding.HostPort + } + } + } + } + + return info +} + func (cr *containerReference) UpdateFromEnv(srcPath string, env *map[string]string) common.Executor { return parseEnvFile(cr, srcPath, env).IfNot(common.Dryrun) } @@ -343,10 +446,10 @@ func (cr *containerReference) Close() common.Executor { } } -// missingContainerError is the shared "container X does not exist" error -// used by ops that need a live cr.id. +// missingContainerError is the shared "container X does not exist" error used by ops that +// need a live cr.id, wrapping ErrContainerNotFound so a caller can tell it from a failing daemon. func (cr *containerReference) missingContainerError(format string, args ...any) error { - return fmt.Errorf("container %q does not exist; cannot "+format, append([]any{cr.input.Name}, args...)...) + return fmt.Errorf("container %q %w; cannot "+format, append([]any{cr.input.Name, ErrContainerNotFound}, args...)...) } func (cr *containerReference) find() common.Executor { @@ -737,7 +840,7 @@ func (cr *containerReference) exec(cmd []string, env map[string]string, user, wo } defer resp.Close() - err = cr.waitForCommand(ctx, isTerminal, resp.HijackedResponse, idResp, user, workdir) + err = cr.waitForCommand(ctx, resp.HijackedResponse, idResp, user, workdir) if err != nil { return err } @@ -795,7 +898,7 @@ func (cr *containerReference) tryReadGID() common.Executor { return cr.tryReadID("-g", func(id int) { cr.GID = id }) } -func (cr *containerReference) waitForCommand(ctx context.Context, isTerminal bool, resp client.HijackedResponse, _ client.ExecCreateResult, _, _ string) error { +func (cr *containerReference) waitForCommand(ctx context.Context, resp client.HijackedResponse, _ client.ExecCreateResult, _, _ string) error { logger := common.Logger(ctx) // Buffered so the copy goroutine never blocks on send if the grace-period @@ -803,28 +906,7 @@ func (cr *containerReference) waitForCommand(ctx context.Context, isTerminal boo cmdResponse := make(chan error, 1) go func() { - var outWriter io.Writer - outWriter = cr.input.Stdout - if outWriter == nil { - outWriter = os.Stdout - } - errWriter := cr.input.Stderr - if errWriter == nil { - errWriter = os.Stderr - } - - var err error - if !isTerminal || os.Getenv("NORAW") != "" { - _, err = stdcopy.StdCopy(outWriter, errWriter, resp.Reader) - } else { - _, err = io.Copy(outWriter, resp.Reader) - } - // Flush any buffered, not-yet-newline-terminated trailing line so the - // final line of a command's output is not lost (e.g. an error message - // printed without a trailing newline before the process exits). - common.FlushWriter(outWriter) - common.FlushWriter(errWriter) - cmdResponse <- err + cmdResponse <- cr.copyOutput(resp.Reader) }() select { @@ -1059,33 +1141,11 @@ func (cr *containerReference) attach() common.Executor { if err != nil { return fmt.Errorf("failed to attach to container: %w", err) } - isTerminal := cr.input.AllocatePTY - - var outWriter io.Writer - outWriter = cr.input.Stdout - if outWriter == nil { - outWriter = os.Stdout - } - errWriter := cr.input.Stderr - if errWriter == nil { - errWriter = os.Stderr - } done := make(chan struct{}) cr.attachDone = done go func() { defer close(done) - var copyErr error - if !isTerminal || os.Getenv("NORAW") != "" { - _, copyErr = stdcopy.StdCopy(outWriter, errWriter, out.Reader) - } else { - _, copyErr = io.Copy(outWriter, out.Reader) - } - // Flush any buffered, not-yet-newline-terminated trailing line once - // the stream reaches EOF, so the final line of the container's - // output is not lost when it is not newline-terminated. - common.FlushWriter(outWriter) - common.FlushWriter(errWriter) - if copyErr != nil { + if copyErr := cr.copyOutput(out.Reader); copyErr != nil { common.Logger(ctx).Error(copyErr) } }() diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index e3c5949f..d6a60d47 100644 --- a/act/container/docker_run_test.go +++ b/act/container/docker_run_test.go @@ -25,6 +25,7 @@ import ( "github.com/moby/moby/api/pkg/stdcopy" "github.com/moby/moby/api/types/container" "github.com/moby/moby/api/types/mount" + "github.com/moby/moby/api/types/network" mobyclient "github.com/moby/moby/client" "github.com/sirupsen/logrus/hooks/test" "github.com/stretchr/testify/assert" @@ -548,7 +549,7 @@ func TestRejectsMissingContainer(t *testing.T) { cr := &containerReference{cli: client, input: &NewContainerInput{Name: "job-1"}} check := func(op string, err error) { t.Helper() - require.Error(t, err, op) + require.ErrorIs(t, err, ErrContainerNotFound, op) assert.Contains(t, err.Error(), `container "job-1" does not exist`, op) } check("copyContent", cr.copyContent("/var/run/act", &FileEntry{Name: "x", Mode: 0o644})(ctx)) @@ -557,6 +558,15 @@ func TestRejectsMissingContainer(t *testing.T) { check("exec", cr.exec([]string{"echo"}, nil, "", "")(ctx)) _, err := cr.GetContainerArchive(ctx, "/var/run/act/x") check("GetContainerArchive", err) + _, err = cr.Inspect(ctx) + check("Inspect", err) + + // a known id the daemon has since dropped + client.On("ContainerInspect", ctx, "gone", mobyclient.ContainerInspectOptions{}). + Return(mobyclient.ContainerInspectResult{}, cerrdefs.ErrNotFound) + removed := &containerReference{id: "gone", cli: client, input: &NewContainerInput{Name: "job-1"}} + _, err = removed.Inspect(ctx) + check("Inspect after removal", err) } // End-to-end: a stale cr.id is cleared, repopulated from name lookup, @@ -825,6 +835,59 @@ func TestCheckVolumesRejectsEscapingHostPaths(t *testing.T) { assert.Empty(t, hostConf.Binds) } +func TestContainerInfoFromInspect(t *testing.T) { + t.Run("reports no healthcheck when the image declares none", func(t *testing.T) { + info := containerInfoFromInspect(container.InspectResponse{ + ID: "abc123", + State: &container.State{Status: "running", Running: true}, + }) + + assert.Equal(t, "abc123", info.ID) + assert.Equal(t, "running", info.State) + assert.Equal(t, HealthNone, info.Health) + assert.Empty(t, info.Ports) + }) + + t.Run("reports the health status and the last probe output", func(t *testing.T) { + info := containerInfoFromInspect(container.InspectResponse{ + State: &container.State{ + Status: "running", + Health: &container.Health{ + Status: container.Unhealthy, + Log: []*container.HealthcheckResult{ + {Output: "first\n"}, + {Output: "connection refused\n"}, + }, + }, + }, + }) + + assert.Equal(t, HealthUnhealthy, info.Health) + assert.Equal(t, "connection refused", info.HealthOutput) + }) + + t.Run("reports the published ports", func(t *testing.T) { + info := containerInfoFromInspect(container.InspectResponse{ + State: &container.State{Status: "running"}, + NetworkSettings: &container.NetworkSettings{ + Ports: network.PortMap{ + network.MustParsePort("5432/tcp"): []network.PortBinding{{HostPort: "49153"}}, + network.MustParsePort("6379/tcp"): nil, + }, + }, + }) + + assert.Equal(t, map[string]string{"5432": "49153"}, info.Ports) + }) + + t.Run("tolerates a container without state", func(t *testing.T) { + info := containerInfoFromInspect(container.InspectResponse{ID: "abc123"}) + + assert.Equal(t, "abc123", info.ID) + assert.Equal(t, HealthNone, info.Health) + }) +} + func TestMergeContainerConfigsVolumesReplaceRunnerMounts(t *testing.T) { logger, _ := test.NewNullLogger() ctx := common.WithLogger(context.Background(), logger) diff --git a/act/container/host_environment.go b/act/container/host_environment.go index 3ca2910c..f963cb70 100644 --- a/act/container/host_environment.go +++ b/act/container/host_environment.go @@ -154,6 +154,14 @@ func (e *HostEnvironment) CopyDir(destPath, srcPath string, useGitIgnore bool) c } } +func (e *HostEnvironment) DumpLogs(_ context.Context) error { + return nil +} + +func (e *HostEnvironment) Inspect(_ context.Context) (*Info, error) { + return &Info{Health: HealthNone, Ports: map[string]string{}}, nil +} + func (e *HostEnvironment) GetContainerArchive(ctx context.Context, srcPath string) (io.ReadCloser, error) { buf := &bytes.Buffer{} tw := tar.NewWriter(buf) diff --git a/act/model/job_context.go b/act/model/job_context.go index d5647803..ed748c0f 100644 --- a/act/model/job_context.go +++ b/act/model/job_context.go @@ -5,12 +5,18 @@ package model type JobContext struct { - Status string `json:"status"` - Container struct { - ID string `json:"id"` - Network string `json:"network"` - } `json:"container"` - Services map[string]struct { - ID string `json:"id"` - } `json:"services"` + Status string `json:"status"` + Container JobContainerContext `json:"container"` + Services map[string]JobService `json:"services"` +} + +type JobContainerContext struct { + ID string `json:"id"` + Network string `json:"network"` +} + +type JobService struct { + ID string `json:"id"` + Network string `json:"network"` + Ports map[string]string `json:"ports"` // container port to the published host port } diff --git a/act/runner/container_mock_test.go b/act/runner/container_mock_test.go index ccb00764..8a56b26f 100644 --- a/act/runner/container_mock_test.go +++ b/act/runner/container_mock_test.go @@ -78,3 +78,14 @@ func (cm *containerMock) GetContainerArchive(ctx context.Context, srcPath string } return args.Get(0).(io.ReadCloser), err } + +func (cm *containerMock) DumpLogs(ctx context.Context) error { + return cm.Called(ctx).Error(0) +} + +func (cm *containerMock) Inspect(ctx context.Context) (*container.Info, error) { + args := cm.Called(ctx) + info, _ := args.Get(0).(*container.Info) + err, _ := args.Get(1).(error) + return info, err +} diff --git a/act/runner/run_context.go b/act/runner/run_context.go index 1bda7c7f..1b0c1a35 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -35,6 +35,7 @@ import ( "github.com/docker/go-connections/nat" "github.com/moby/moby/api/types/mount" "github.com/opencontainers/selinux/go-selinux" + "golang.org/x/sync/errgroup" ) // RunContext contains info about current job @@ -56,7 +57,7 @@ type RunContext struct { IntraActionState map[string]map[string]string ExprEval ExpressionEvaluator JobContainer container.ExecutionsEnvironment - ServiceContainers []container.ExecutionsEnvironment + serviceContainers []*serviceContainer OutputMappings map[MappableOutput]MappableOutput JobName string ActionPath string @@ -84,6 +85,9 @@ type RunContext struct { // failures. Those failures must still make success() false and failure() true for later // main-step if evaluation. jobFailed bool + // empty for a host-mode job, which starts no container + jobContainerID string + jobNetworkName string // 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. @@ -92,6 +96,15 @@ type RunContext struct { unsecureCommandMu sync.Mutex } +// serviceContainer pairs a service container with the workflow id that keys job.services. +type serviceContainer struct { + name string + image string + container container.ExecutionsEnvironment + logsDumped bool + info *container.Info // last poll, the source of the `job.services` entry +} + // setCurrentStepEnv records the environment of the step about to run. func (rc *RunContext) setCurrentStepEnv(env map[string]string) { rc.unsecureCommandMu.Lock() @@ -508,7 +521,7 @@ func (rc *RunContext) startJobContainer() common.Executor { Privileged: rc.Config.Privileged, UsernsMode: rc.Config.UsernsMode, Platform: rc.Config.ContainerArchitecture, - AutoRemove: rc.Config.AutoRemove, + AutoRemove: false, // so a dead service's log survives, cleanupJobResources removes it Options: rc.ExprEval.Interpolate(ctx, spec.Options), NetworkMode: networkName, NetworkAliases: []string{serviceID}, @@ -516,7 +529,7 @@ func (rc *RunContext) startJobContainer() common.Executor { PortBindings: portBindings, AllocatePTY: rc.Config.AllocatePTY, }) - rc.ServiceContainers = append(rc.ServiceContainers, c) + rc.serviceContainers = append(rc.serviceContainers, &serviceContainer{name: serviceID, image: serviceImage, container: c}) } rc.cleanUpJobContainer = rc.cleanupJobResources(networkName, createAndDeleteNetwork) @@ -551,6 +564,8 @@ func (rc *RunContext) startJobContainer() common.Executor { return errors.New("Failed to create job container") } + rc.jobNetworkName = networkName + defer printStartJobContainerGroup(ctx, image, name, networkName)() return common.NewPipelineExecutor( rc.pullServicesImages(rc.Config.ForcePull), @@ -558,9 +573,12 @@ func (rc *RunContext) startJobContainer() common.Executor { rc.stopJobContainer(), container.NewDockerNetworkCreateExecutor(networkName, rc.Config.ContainerNetworkCreateOptions). IfBool(createAndDeleteNetwork), - rc.startServiceContainers(networkName), + rc.startServiceContainers(), + rc.reportUnstartedServices(), + rc.waitForServiceContainers(), rc.JobContainer.Create(rc.Config.ContainerCapAdd, rc.Config.ContainerCapDrop), rc.JobContainer.Start(false), + rc.captureJobContainerInfo(), rc.JobContainer.Copy(rc.JobContainer.GetActPath()+"/", &container.FileEntry{ Name: "workflow/event.json", Mode: 0o644, @@ -585,7 +603,7 @@ func (rc *RunContext) cleanupJobResources(networkName string, createAndDeleteNet if removeJobContainer { errs = append(errs, rc.JobContainer.Remove()(ctx)) } - if len(rc.ServiceContainers) > 0 { + if len(rc.serviceContainers) > 0 { logger.Infof("Cleaning up services for job %s", rc.JobName) if err := rc.stopServiceContainers()(ctx); err != nil { logger.Errorf("Error while cleaning services: %v", err) @@ -682,21 +700,21 @@ func (rc *RunContext) stopJobContainer() common.Executor { func (rc *RunContext) pullServicesImages(forcePull bool) common.Executor { return func(ctx context.Context) error { execs := []common.Executor{} - for _, c := range rc.ServiceContainers { - execs = append(execs, c.Pull(forcePull)) + for _, svc := range rc.serviceContainers { + execs = append(execs, svc.container.Pull(forcePull)) } return common.NewParallelExecutor(len(execs), execs...)(ctx) } } -func (rc *RunContext) startServiceContainers(_ string) common.Executor { +func (rc *RunContext) startServiceContainers() common.Executor { return func(ctx context.Context) error { execs := []common.Executor{} - for _, c := range rc.ServiceContainers { + for _, svc := range rc.serviceContainers { execs = append(execs, common.NewPipelineExecutor( - c.Pull(false), - c.Create(rc.Config.ContainerCapAdd, rc.Config.ContainerCapDrop), - c.Start(false), + svc.container.Pull(false), + svc.container.Create(rc.Config.ContainerCapAdd, rc.Config.ContainerCapDrop), + svc.container.Start(false), )) } return common.NewParallelExecutor(len(execs), execs...)(ctx) @@ -706,13 +724,159 @@ func (rc *RunContext) startServiceContainers(_ string) common.Executor { func (rc *RunContext) stopServiceContainers() common.Executor { return func(ctx context.Context) error { execs := []common.Executor{} - for _, c := range rc.ServiceContainers { - execs = append(execs, c.Remove().Finally(c.Close())) + for _, svc := range rc.serviceContainers { + execs = append(execs, svc.container.Remove().Finally(svc.container.Close())) } return common.NewParallelExecutor(len(execs), execs...)(ctx) } } +const ( + defaultServiceReadyTimeout = 5 * time.Minute + serviceReadyPollMax = 32 * time.Second +) + +var serviceReadyPollInterval = 2 * time.Second // a variable so tests need not wait + +// 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. +func (rc *RunContext) reportUnstartedServices() common.Executor { + return func(ctx context.Context) error { + logger := common.Logger(ctx) + for _, svc := range rc.serviceContainers { + info, err := svc.inspect(ctx) + if err != nil { + logger.Debugf("unable to inspect service '%s': %v", svc.name, err) + continue + } + if info.State == container.StateRunning { + continue + } + svc.dumpLogs(ctx) + logger.Warnf("Docker container %s is not in running state: %s (%d)", info.ID, info.State, info.ExitCode) + } + return nil + } +} + +// waitForServiceContainers blocks until every service that declares a healthcheck reports +// healthy, as GitHub does, so a first step cannot connect before the service listens. +func (rc *RunContext) waitForServiceContainers() common.Executor { + return func(ctx context.Context) error { + if len(rc.serviceContainers) == 0 { + return nil + } + + timeout := rc.Config.ServiceReadyTimeout + switch { + case timeout < 0: + // disabled, but still describe the containers for `job.services` + for _, svc := range rc.serviceContainers { + if _, err := svc.inspect(ctx); err != nil && !errors.Is(err, container.ErrContainerNotFound) { + return err + } + } + return nil + case timeout == 0: + timeout = defaultServiceReadyTimeout + } + + // the first error cancels the rest, so a failure does not wait out a sibling's timeout + group, groupCtx := errgroup.WithContext(ctx) + for _, svc := range rc.serviceContainers { + group.Go(func() error { + return svc.waitUntilHealthy(groupCtx, timeout) + }) + } + return group.Wait() + } +} + +// waitUntilHealthy waits on the healthcheck alone, so a container that declares none is +// 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 + + ctx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + + for { + info, err := svc.inspect(ctx) + if ctxErr := ctx.Err(); ctxErr != nil { // the wait ended, an inspect error only noticed it + if errors.Is(ctxErr, context.DeadlineExceeded) { + return fmt.Errorf("the service '%s' did not become healthy within %s%s", svc.name, timeout, svc.healthOutputSuffix()) + } + return ctxErr + } + switch { + case errors.Is(err, container.ErrContainerNotFound): + return nil // gone, so there is no health left to wait on + case err != nil: + return err + } + + switch { + case info.Health == container.HealthUnhealthy: + svc.dumpLogs(ctx) + common.Logger(ctx).Errorf("Failed to initialize container %s", svc.image) + return fmt.Errorf("the service '%s' is unhealthy%s", svc.name, svc.healthOutputSuffix()) + case info.Health != container.HealthStarting: + rawLogger.Infof("%s service is healthy.", svc.name) + return nil + } + + rawLogger.Infof("%s service is starting, waiting %d seconds before checking again.", svc.name, int(interval.Seconds())) + select { + case <-ctx.Done(): // reported at the top of the loop + case <-time.After(interval): + } + interval = min(interval*2, serviceReadyPollMax) + } +} + +// dumpLogs writes the container's log to the job log once, however often it is reported. +func (svc *serviceContainer) dumpLogs(ctx context.Context) { + if svc.logsDumped { + return + } + svc.logsDumped = true + if err := svc.container.DumpLogs(ctx); err != nil { + common.Logger(ctx).Debugf("unable to read the log of service '%s': %v", svc.name, err) + } +} + +// inspect also records the state for the `job.services` context. +func (svc *serviceContainer) inspect(ctx context.Context) (*container.Info, error) { + info, err := svc.container.Inspect(ctx) + if err != nil { + return nil, fmt.Errorf("failed to inspect service '%s': %w", svc.name, err) + } + svc.info = info + return info, nil +} + +func (svc *serviceContainer) healthOutputSuffix() string { + if svc.info == nil || svc.info.HealthOutput == "" { + return "" + } + return ": " + svc.info.HealthOutput +} + +// captureJobContainerInfo is a convenience: failing to describe the container must not +// fail the job. +func (rc *RunContext) captureJobContainerInfo() common.Executor { + return func(ctx context.Context) error { + info, err := rc.JobContainer.Inspect(ctx) + if err != nil { + common.Logger(ctx).Debugf("unable to inspect the job container: %v", err) + return nil + } + rc.jobContainerID = info.ID + return nil + } +} + // Prepare the mounts and binds for the worker // ActionCacheDir is for rc @@ -1053,9 +1217,26 @@ func (rc *RunContext) getJobContext() *model.JobContext { if rc.jobCancelled { jobStatus = "cancelled" } - return &model.JobContext{ - Status: jobStatus, + + jobContext := &model.JobContext{ + Status: jobStatus, + Services: map[string]model.JobService{}, // an empty map, never null } + if rc.jobContainerID != "" { + jobContext.Container.ID = rc.jobContainerID + jobContext.Container.Network = rc.jobNetworkName + } + for _, svc := range rc.serviceContainers { + if svc.info == nil { + continue + } + jobContext.Services[svc.name] = model.JobService{ + ID: svc.info.ID, + Network: rc.jobNetworkName, + Ports: svc.info.Ports, + } + } + return jobContext } func (rc *RunContext) getStepsContext() map[string]*model.StepResult { diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index 8179acfa..962c7704 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -13,6 +13,7 @@ import ( "runtime" "strings" "testing" + "time" "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/container" @@ -22,6 +23,7 @@ import ( "github.com/docker/cli/cli/compose/loader" log "github.com/sirupsen/logrus" assert "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" require "github.com/stretchr/testify/require" yaml "go.yaml.in/yaml/v4" ) @@ -225,6 +227,12 @@ func (fakeContainer) Copy(string, ...*container.FileEntry) common.Executor { return func(context.Context) error { return nil } } +func (fakeContainer) Inspect(context.Context) (*container.Info, error) { + return &container.Info{ID: "fake", State: "running", Health: container.HealthNone}, nil +} + +func (fakeContainer) DumpLogs(context.Context) error { return nil } + // Regression test: a service without a `credentials:` block resolves to empty // credentials, which used to overwrite the job container's own credentials. func TestStartJobContainerKeepsJobCredentialsWithServices(t *testing.T) { @@ -563,7 +571,7 @@ func TestCleanupJobResourcesCleansServicesWithoutJobContainer(t *testing.T) { rc := &RunContext{ Config: &Config{}, - ServiceContainers: []container.ExecutionsEnvironment{service}, + serviceContainers: []*serviceContainer{{name: "svc", container: service}}, } err := rc.cleanupJobResources("external-network", false)(context.Background()) @@ -586,7 +594,7 @@ func TestCleanupJobResourcesContinuesAfterFailure(t *testing.T) { Config: &Config{}, Run: &model.Run{Workflow: &model.Workflow{Name: "wf"}, JobID: "job"}, JobContainer: jobContainer, - ServiceContainers: []container.ExecutionsEnvironment{service}, + serviceContainers: []*serviceContainer{{name: "svc", container: service}}, } ctx, cancel := context.WithCancel(context.Background()) @@ -1045,6 +1053,199 @@ func TestRunContext_cleanupFailedStart(t *testing.T) { }) } +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}, + serviceContainers: services, + } + } + + t.Run("returns as soon as a service without a healthcheck runs", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything). + Return(&container.Info{ID: "id", State: "running", Health: container.HealthNone}, nil).Once() + + rc := newRunContext(0, &serviceContainer{name: "redis", container: service}) + require.NoError(t, rc.waitForServiceContainers()(context.Background())) + 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() + + rc := newRunContext(0, &serviceContainer{name: "postgres", container: service}) + require.NoError(t, rc.waitForServiceContainers()(context.Background())) + service.AssertExpectations(t) + }) + + t.Run("fails with the probe output when a service is unhealthy", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything).Return(&container.Info{ + State: "running", + Health: container.HealthUnhealthy, + HealthOutput: "connection refused", + }, nil).Once() + service.On("DumpLogs", mock.Anything).Return(nil).Once() + + rc := newRunContext(0, &serviceContainer{name: "postgres", container: service}) + err := rc.waitForServiceContainers()(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "the service 'postgres' is unhealthy: connection refused") + service.AssertExpectations(t) + }) + + t.Run("lets the steps run when a service exits without a healthcheck", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything). + Return(&container.Info{State: "exited", ExitCode: 2, Health: container.HealthNone}, nil).Once() + + rc := newRunContext(0, &serviceContainer{name: "postgres", container: service}) + require.NoError(t, rc.waitForServiceContainers()(context.Background())) + }) + + t.Run("proceeds when the container is gone", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything). + Return((*container.Info)(nil), container.ErrContainerNotFound).Once() + + rc := newRunContext(0, &serviceContainer{name: "postgres", container: service}) + require.NoError(t, rc.waitForServiceContainers()(context.Background())) + }) + + t.Run("fails right away when one service fails while another is still starting", func(t *testing.T) { + failing := &containerMock{} + failing.On("Inspect", mock.Anything). + Return(&container.Info{State: "running", Health: container.HealthUnhealthy}, nil) + failing.On("DumpLogs", mock.Anything).Return(nil).Once() + starting := &containerMock{} + starting.On("Inspect", mock.Anything). + Return(&container.Info{State: "running", Health: container.HealthStarting}, nil) + + rc := newRunContext(10*time.Second, + &serviceContainer{name: "failing", container: failing}, + &serviceContainer{name: "starting", container: starting}) + + done := make(chan error, 1) + go func() { done <- rc.waitForServiceContainers()(context.Background()) }() + + select { + case err := <-done: + require.Error(t, err) + assert.Contains(t, err.Error(), "the service 'failing' is unhealthy") + case <-time.After(2 * time.Second): + t.Fatal("waitForServiceContainers did not fail fast; it waited for the starting service") + } + }) + + t.Run("gives up once the timeout expires", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything). + Return(&container.Info{State: "running", Health: container.HealthStarting}, nil) + + rc := newRunContext(20*time.Millisecond, &serviceContainer{name: "postgres", container: service}) + err := rc.waitForServiceContainers()(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "did not become healthy within") + }) + + t.Run("gives up with the same message when the deadline stops an inspect", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything). + Run(func(args mock.Arguments) { <-args.Get(0).(context.Context).Done() }). + Return((*container.Info)(nil), errors.New("inspect aborted")) + + rc := newRunContext(20*time.Millisecond, &serviceContainer{name: "postgres", container: service}) + err := rc.waitForServiceContainers()(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "did not become healthy within") + }) + + t.Run("does not wait when the timeout is negative", func(t *testing.T) { + service := &containerMock{} + // Still described once, so the `job.services` context is filled either way. + service.On("Inspect", mock.Anything). + Return(&container.Info{ID: "id", State: "running", Health: container.HealthStarting}, nil).Once() + + svc := &serviceContainer{name: "postgres", container: service} + rc := newRunContext(-1, svc) + require.NoError(t, rc.waitForServiceContainers()(context.Background())) + service.AssertExpectations(t) + assert.Equal(t, "id", svc.info.ID) + }) + + t.Run("fails on an inspect error even when the timeout is negative", func(t *testing.T) { + service := &containerMock{} + service.On("Inspect", mock.Anything).Return((*container.Info)(nil), errors.New("daemon is gone")).Once() + + rc := newRunContext(-1, &serviceContainer{name: "postgres", container: service}) + err := rc.waitForServiceContainers()(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), "failed to inspect service 'postgres'") + }) + + t.Run("is a no-op without services", func(t *testing.T) { + require.NoError(t, newRunContext(0).waitForServiceContainers()(context.Background())) + }) +} + +func TestReportUnstartedServices(t *testing.T) { + dead := &containerMock{} + dead.On("Inspect", mock.Anything).Return(&container.Info{ID: "dead-id", State: "exited", ExitCode: 1}, nil).Once() + dead.On("DumpLogs", mock.Anything).Return(nil).Once() + running := &containerMock{} + running.On("Inspect", mock.Anything).Return(&container.Info{ID: "run-id", State: "running"}, nil).Once() + + rc := &RunContext{serviceContainers: []*serviceContainer{ + {name: "postgres", container: dead}, + {name: "redis", container: running}, + }} + + require.NoError(t, rc.reportUnstartedServices()(context.Background())) + dead.AssertExpectations(t) + running.AssertExpectations(t) +} + +func TestGetJobContextReportsContainers(t *testing.T) { + rc := &RunContext{ + jobNetworkName: "job-network", + jobContainerID: "job-container-id", + serviceContainers: []*serviceContainer{ + {name: "postgres", info: &container.Info{ID: "svc-id", Ports: map[string]string{"5432": "49153"}}}, + // A service that publishes no port reports an empty map, as GitHub does. + {name: "redis", info: &container.Info{ID: "redis-id", Ports: map[string]string{}}}, + // A service that never reported is left out rather than reported as empty. + {name: "mailhog"}, + }, + } + + jobContext := rc.getJobContext() + + assert.Equal(t, "job-container-id", jobContext.Container.ID) + assert.Equal(t, "job-network", jobContext.Container.Network) + assert.Equal(t, map[string]model.JobService{ + "postgres": {ID: "svc-id", Network: "job-network", Ports: map[string]string{"5432": "49153"}}, + "redis": {ID: "redis-id", Network: "job-network", Ports: map[string]string{}}, + }, jobContext.Services) +} + +// A job that never started a container reports an empty context, not a placeholder. +func TestGetJobContextWithoutContainer(t *testing.T) { + jobContext := (&RunContext{}).getJobContext() + + assert.Empty(t, jobContext.Container.ID) + assert.Empty(t, jobContext.Container.Network) + assert.Empty(t, jobContext.Services) +} + func TestImageOSFromImage(t *testing.T) { for _, tc := range []struct { image string diff --git a/act/runner/runner.go b/act/runner/runner.go index b92da463..1eaddccb 100644 --- a/act/runner/runner.go +++ b/act/runner/runner.go @@ -94,6 +94,7 @@ type Config struct { InsecureSkipTLS bool // whether to skip verifying TLS certificate of the Gitea instance MaxParallel int // max parallel jobs to run across all workflows (0 = no limit, uses CPU count) AllocatePTY bool // allocate a pseudo-TTY for each step's process + ServiceReadyTimeout time.Duration // how long a job waits for its service containers to report healthy (0 uses the default) RunnerName string // name this runner registered with, reported as `runner.name`, defaults to the hostname JobStartedHook string // script run inside the job environment before the job's first step; ACTIONS_RUNNER_HOOK_JOB_STARTED is read from Env when empty JobCompletedHook string // script run inside the job environment after the job's last step; ACTIONS_RUNNER_HOOK_JOB_COMPLETED is read from Env when empty diff --git a/act/runner/testdata/services/push.yaml b/act/runner/testdata/services/push.yaml index ef883da6..2ebce33b 100644 --- a/act/runner/testdata/services/push.yaml +++ b/act/runner/testdata/services/push.yaml @@ -15,3 +15,9 @@ jobs: echo "id: ${{ job.services.postgres.id }}" echo "network: ${{ job.services.postgres.network }}" echo "ports: ${{ job.services.postgres.ports }}" + - name: The job context describes the started containers + run: | + test -n "${{ job.container.id }}" + test -n "${{ job.services.postgres.id }}" + test -n "${{ job.services.postgres.ports['80'] }}" + test "${{ job.services.postgres.network }}" = "${{ job.container.network }}" diff --git a/go.mod b/go.mod index 8fe40459..269122da 100644 --- a/go.mod +++ b/go.mod @@ -37,6 +37,7 @@ require ( go.etcd.io/bbolt v1.5.0 go.yaml.in/yaml/v4 v4.0.0-rc.3 golang.org/x/net v0.57.0 + golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.0 golang.org/x/term v0.45.0 golang.org/x/text v0.40.0 @@ -104,7 +105,6 @@ require ( go.opentelemetry.io/otel/trace v1.44.0 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/crypto v0.54.0 // indirect - golang.org/x/sync v0.22.0 // indirect gopkg.in/warnings.v0 v0.1.2 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 0afd3cc6..1c7d05cc 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -510,6 +510,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. RunnerUUID: r.uuid, }, ContainerOptions: r.cfg.Container.Options, + ServiceReadyTimeout: r.cfg.Container.ServiceReadyTimeout, ContainerDaemonSocket: r.cfg.Container.DockerHost, Privileged: r.cfg.Container.Privileged, DefaultActionInstance: r.getDefaultActionsURL(task), diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index ed61aace..51322cc0 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -226,6 +226,9 @@ container: # from the DinD daemon's filesystem. When enabled, ensure the workspace parent # directory is also mounted into the runner container and listed in valid_volumes. bind_workdir: false + # How long a job waits for a service container that declares a healthcheck to become + # healthy. A negative value (e.g. -1s) starts the steps without waiting. + service_ready_timeout: 5m host: # The parent directory of a job's working directory. diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 63d97538..3dc7e5ac 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -92,6 +92,7 @@ type Container struct { RequireDocker bool `yaml:"require_docker"` // Always require a reachable docker daemon, even if not required by runner DockerTimeout time.Duration `yaml:"docker_timeout"` // Timeout to wait for the docker daemon to be reachable, if docker is required by require_docker or runner BindWorkdir bool `yaml:"bind_workdir"` // BindWorkdir binds the workspace to the host filesystem instead of using Docker volumes. Required for DinD when jobs use docker compose with bind mounts. + ServiceReadyTimeout time.Duration `yaml:"service_ready_timeout"` // ServiceReadyTimeout bounds how long a job waits for a service container that declares a healthcheck to report healthy. Negative disables waiting. } type ContainerNetworkCreateOptions struct {