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>.{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 <me@silverwind.io>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1107
Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com>
Co-authored-by: bircni <bircni@icloud.com>

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

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)
 			}
 		}()

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)

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)