diff --git a/act/container/docker_run.go b/act/container/docker_run.go index b2266ee8..714c646b 100644 --- a/act/container/docker_run.go +++ b/act/container/docker_run.go @@ -895,6 +895,7 @@ func (cr *containerReference) tryReadID(opt string, cbk func(id int)) common.Exe return nil } defer resp.Close() + defer context.AfterFunc(ctx, resp.Close)() sid, err := resp.Reader.ReadString('\n') if err != nil { diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index c1fbaa61..3af5103e 100644 --- a/act/container/docker_run_test.go +++ b/act/container/docker_run_test.go @@ -378,6 +378,21 @@ func TestDockerAttachStreamClosesWithItsContext(t *testing.T) { client.AssertExpectations(t) } +func TestDockerReadIDStreamClosesWithItsContext(t *testing.T) { + conn, peer := net.Pipe() + t.Cleanup(func() { _ = peer.Close() }) + client := &mockDockerClient{} + client.On("ExecCreate", mock.Anything, "123", mock.AnythingOfType("client.ExecCreateOptions")). + Return(mobyclient.ExecCreateResult{ID: "id"}, nil) + client.On("ExecAttach", mock.Anything, "id", mobyclient.ExecAttachOptions{}). + Return(mobyclient.ExecAttachResult{HijackedResponse: mobyclient.NewHijackedResponse(conn, "")}, nil) + ctx, cancel := context.WithCancel(t.Context()) + cancel() + + require.NoError(t, (&containerReference{id: "123", cli: client}).tryReadID("-u", func(int) { t.Error("read an ID after cancellation") })(ctx)) + client.AssertExpectations(t) +} + func TestDockerWaitFailure(t *testing.T) { ctx := t.Context() statusCh := make(chan container.WaitResponse, 1) diff --git a/e2e/cli_test.go b/e2e/cli_test.go index d295701b..97d443ee 100644 --- a/e2e/cli_test.go +++ b/e2e/cli_test.go @@ -90,20 +90,29 @@ container: type cliDaemon struct { cmd *exec.Cmd log *syncBuffer - exited chan error + exited chan struct{} + err error } func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon { t.Helper() - d := &cliDaemon{log: &syncBuffer{}, exited: make(chan error, 1)} + d := &cliDaemon{log: &syncBuffer{}, exited: make(chan struct{})} d.cmd = exec.Command(c.bin, append([]string{"daemon", "-c", c.cfgFile}, args...)...) d.cmd.Stdout, d.cmd.Stderr = d.log, d.log if err := d.cmd.Start(); err != nil { t.Fatalf("start daemon: %v", err) } - go func() { d.exited <- d.cmd.Wait() }() + go func() { + d.err = d.cmd.Wait() + close(d.exited) + }() t.Cleanup(func() { - _ = d.cmd.Process.Kill() // no-op once the daemon has exited + _ = d.cmd.Process.Signal(syscall.SIGTERM) + select { + case <-d.exited: + case <-time.After(30 * time.Second): + _ = d.cmd.Process.Kill() + } }) return d } @@ -112,8 +121,8 @@ func (c *cliRunner) startDaemon(t *testing.T, args ...string) *cliDaemon { func (d *cliDaemon) wait(t *testing.T, timeout time.Duration) error { t.Helper() select { - case err := <-d.exited: - return err + case <-d.exited: + return d.err case <-time.After(timeout): t.Fatalf("daemon still running after %s\n%s", timeout, d.log) return nil @@ -140,8 +149,8 @@ func (d *cliDaemon) requireSuccess(t *testing.T, c *cliRunner, runID int64) { // testCLIRegisterDaemon drives the built binary the way a user does: register, then daemon. func testCLIRegisterDaemon(t *testing.T) { t.Parallel() - c := newCLIRunner(t, "e2e-cli", "") - d := c.startDaemon(t) + c := newCLIRunner(t, "e2e-registered", "") + d := c.startDaemon(t, "--labels", "e2e-cli:docker://"+os.Getenv("E2E_JOB_IMAGE")) pushWorkflow(t, c.api, c.repo, "cli.yml") wfRun := waitForRun(t, c.api, c.repo) @@ -150,6 +159,22 @@ func testCLIRegisterDaemon(t *testing.T) { t.Errorf("job logs are missing the marker:\n%s", logs) } + runners, err := c.api.Runners(t.Context(), c.repo) + if err != nil { + t.Fatalf("list runners: %v", err) + } + var labels []string + for _, runner := range runners { + if runner.Name == c.repo { + for _, label := range runner.Labels { + labels = append(labels, label.Name) + } + } + } + if !slices.Equal(labels, []string{"e2e-cli"}) { + t.Errorf("runner %q has labels %v on the server, want only the --labels override", c.repo, labels) + } + d.terminate(t) if err := d.wait(t, 30*time.Second); err != nil { t.Errorf("daemon did not exit cleanly on SIGTERM: %v\n%s", err, d.log) @@ -186,36 +211,6 @@ func testCLIOnce(t *testing.T) { } } -// testCLILabelsOverride checks that daemon --labels replaces the labels given at registration. -func testCLILabelsOverride(t *testing.T) { - t.Parallel() - c := newCLIRunner(t, "e2e-registered", "") - d := c.startDaemon(t, "--labels", "e2e-override:docker://"+os.Getenv("E2E_JOB_IMAGE")) - - pushWorkflow(t, c.api, c.repo, "cli_labels.yml") - wfRun := waitForRun(t, c.api, c.repo) - d.requireSuccess(t, c, wfRun.ID) - - runners, err := c.api.Runners(t.Context(), c.repo) - if err != nil { - t.Fatalf("list runners: %v", err) - } - for _, runner := range runners { - if runner.Name != c.repo { - continue - } - var labels []string - for _, label := range runner.Labels { - labels = append(labels, label.Name) - } - if !slices.Equal(labels, []string{"e2e-override"}) { - t.Errorf("runner has labels %v on the server, want only the --labels override", labels) - } - return - } - t.Errorf("runner %q not found among %+v", c.repo, runners) -} - // testCLIShutdownDrains checks that SIGTERM lets a running job finish within shutdown_timeout. func testCLIShutdownDrains(t *testing.T) { t.Parallel() @@ -282,10 +277,6 @@ func testExecArtifacts(t *testing.T) { t.Parallel() ctx := t.Context() - bin, err := runnerBinary() - if err != nil { - t.Fatal(err) - } workflow, err := os.ReadFile("testdata/workflows/exec_artifact.yml") if err != nil { t.Fatal(err) @@ -309,11 +300,9 @@ func testExecArtifacts(t *testing.T) { } artifacts := t.TempDir() - args := []string{"exec", "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--network", fixture.network} - cmd := exec.CommandContext(ctx, bin, args...) - cmd.Env = slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }) - if out, err := cmd.CombinedOutput(); err != nil { - t.Fatalf("exec: %v\n%s", err, out) + stdout, stderr, err := runExec(t, nil, "-C", workdir, "-W", workflows, "-i", os.Getenv("E2E_JOB_IMAGE"), "--artifact-server-path", artifacts, "--network", fixture.network) + if err != nil { + t.Fatalf("exec: %v\n%s%s", err, stdout, stderr) } if blobs, _ := filepath.Glob(filepath.Join(artifacts, "v4", "*", "*", "blob")); len(blobs) != 2 { t.Fatalf("want one v4 artifact blob per workflow run, got %v", blobs) diff --git a/e2e/compatibility_test.go b/e2e/compatibility_test.go index 7fa9331b..6b2f2c0d 100644 --- a/e2e/compatibility_test.go +++ b/e2e/compatibility_test.go @@ -37,7 +37,6 @@ func TestCompatibility(t *testing.T) { t.Run("job_timeout", testJobTimeout) t.Run("cli_register_daemon", testCLIRegisterDaemon) t.Run("cli_once", testCLIOnce) - t.Run("cli_labels_override", testCLILabelsOverride) t.Run("cli_shutdown_drains", testCLIShutdownDrains) t.Run("cli_shutdown_cancels", testCLIShutdownCancels) t.Run("host_mode", testHostMode) diff --git a/e2e/gitea_fixture.go b/e2e/gitea_fixture.go index 3d19da8d..135d14ec 100644 --- a/e2e/gitea_fixture.go +++ b/e2e/gitea_fixture.go @@ -80,11 +80,11 @@ func selfNetwork(ctx context.Context, cli mobyclient.APIClient) (string, string, return "", "", false } -func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, error) { - closeClient := true +func StartGitea(ctx context.Context, cli mobyclient.APIClient) (_ *GiteaFixture, err error) { + f := &GiteaFixture{cli: cli} defer func() { - if closeClient { - _ = cli.Close() + if err != nil { + _ = f.Close() } }() @@ -123,26 +123,21 @@ func StartGitea(ctx context.Context, cli mobyclient.APIClient) (*GiteaFixture, e return nil, fmt.Errorf("create gitea container: %w", err) } - f := &GiteaFixture{cli: cli, id: resp.ID, image: image, baseURL: baseURL, network: sharedNet, selfIP: selfIP} - closeClient = false + f.id, f.image, f.baseURL, f.network, f.selfIP = resp.ID, image, baseURL, sharedNet, selfIP if _, err := cli.ContainerStart(ctx, f.id, mobyclient.ContainerStartOptions{}); err != nil { - _ = f.Close(ctx) return nil, fmt.Errorf("start gitea container: %w", err) } if err := f.waitHealthy(ctx); err != nil { - _ = f.Close(ctx) return nil, fmt.Errorf("gitea did not become healthy: %w", err) } if err := f.bootstrapAdmin(ctx); err != nil { - _ = f.Close(ctx) return nil, fmt.Errorf("bootstrap gitea admin: %w", err) } if err := f.readVersion(ctx); err != nil { - _ = f.Close(ctx) return nil, fmt.Errorf("read gitea version: %w", err) } @@ -249,6 +244,7 @@ func containerExec(ctx context.Context, cli mobyclient.APIClient, id, user strin if err != nil { return "", err } + defer context.AfterFunc(ctx, attached.Close)() defer attached.Close() if stdin != nil { go func() { @@ -258,7 +254,11 @@ func containerExec(ctx context.Context, cli mobyclient.APIClient, id, user strin } var out bytes.Buffer - if _, err := stdcopy.StdCopy(&out, &out, attached.Reader); err != nil { + _, err = stdcopy.StdCopy(&out, &out, attached.Reader) + if ctx.Err() != nil { + return "", ctx.Err() + } + if err != nil { return "", err } @@ -291,10 +291,12 @@ func (f *GiteaFixture) doJSON(ctx context.Context, method, path string, reqBody, return api.doJSON(ctx, method, path, reqBody, respBody) } -func (f *GiteaFixture) Close(ctx context.Context) error { +func (f *GiteaFixture) Close() error { if f.id == "" { return f.cli.Close() } + ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) + defer cancel() _, removeErr := f.cli.ContainerRemove(ctx, f.id, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true}) return errors.Join(removeErr, f.cli.Close()) } diff --git a/e2e/kubernetes_test.go b/e2e/kubernetes_test.go index 33a35b05..05a55491 100644 --- a/e2e/kubernetes_test.go +++ b/e2e/kubernetes_test.go @@ -55,7 +55,9 @@ func checkClusterLeaks() error { if err != nil { return nil // the tests that needed the cluster already failed } - if out, err := containerExec(context.Background(), dockerCLI, cluster.id, "", nil, + ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) + defer cancel() + if out, err := containerExec(ctx, dockerCLI, cluster.id, "", nil, "kubectl", "wait", "--for=delete", "pods,secrets", "--selector=app.kubernetes.io/managed-by=gitea-runner", "--timeout=1m", ); err != nil { return fmt.Errorf("job pods or secrets left behind: %w: %s", err, out) @@ -72,7 +74,7 @@ type kubeCluster struct { var ( k3s = sync.OnceValues(startK3s) // independent of Gitea, so it can boot alongside it clusterStarted atomic.Bool - clusterCleanup []func() + clusterCleanup []func(context.Context) sharedCluster = sync.OnceValues(func() (*kubeCluster, error) { cluster, err := k3s() if err != nil { @@ -109,7 +111,9 @@ func stopCluster() { } _, _ = k3s() for _, cleanup := range slices.Backward(clusterCleanup) { - cleanup() + ctx, cancel := context.WithTimeout(context.Background(), cleanupTimeout) + cleanup(ctx) + cancel() } } @@ -138,8 +142,8 @@ func startK3s() (*kubeCluster, error) { if err != nil { return nil, fmt.Errorf("create k3s container: %w", err) } - clusterCleanup = append(clusterCleanup, func() { - _, _ = cli.ContainerRemove(context.Background(), created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true}) + clusterCleanup = append(clusterCleanup, func(ctx context.Context) { + _, _ = cli.ContainerRemove(ctx, created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true}) }) if _, err := cli.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{}); err != nil { return nil, fmt.Errorf("start k3s container: %w", err) @@ -170,7 +174,9 @@ func startK3s() (*kubeCluster, error) { inspected, inspectErr := cli.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{}) if inspectErr != nil || !inspected.Container.State.Running || ctx.Err() != nil { var logs bytes.Buffer - if reader, logsErr := cli.ContainerLogs(context.Background(), created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil { + logsCtx, logsCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer logsCancel() + if reader, logsErr := cli.ContainerLogs(logsCtx, created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil { _, _ = stdcopy.StdCopy(&logs, &logs, reader) _ = reader.Close() } @@ -226,7 +232,7 @@ func preloadImages(ctx context.Context, k3sID string, images ...string) ([]strin if _, err := dockerCLI.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil { return nil, fmt.Errorf("tag %s: %w", image, err) } - clusterCleanup = append(clusterCleanup, func() { _, _ = dockerCLI.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) }) + clusterCleanup = append(clusterCleanup, func(ctx context.Context) { _, _ = dockerCLI.ImageRemove(ctx, alias, mobyclient.ImageRemoveOptions{}) }) aliases = append(aliases, alias) } archive, err := dockerCLI.ImageSave(ctx, aliases) diff --git a/e2e/main_test.go b/e2e/main_test.go index 5500987c..c746a951 100644 --- a/e2e/main_test.go +++ b/e2e/main_test.go @@ -31,6 +31,8 @@ import ( const runTimeout = 3 * time.Minute +const cleanupTimeout = 2 * time.Minute + var nonRepoChars = regexp.MustCompile(`[^a-zA-Z0-9._-]+`) var fixture *GiteaFixture @@ -42,8 +44,10 @@ var binDir string var dockerCLI mobyclient.APIClient var runnerBinary = sync.OnceValues(func() (string, error) { + ctx, cancel := context.WithTimeout(context.Background(), runTimeout) + defer cancel() bin := filepath.Join(binDir, "gitea-runner") - if out, err := exec.Command(cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil { + if out, err := exec.CommandContext(ctx, cmp.Or(os.Getenv("GO"), "go"), "build", "-o", bin, "..").CombinedOutput(); err != nil { return "", fmt.Errorf("build runner: %w\n%s", err, out) } return bin, nil @@ -80,7 +84,7 @@ func runSuite(m *testing.M) int { return 1 } fixture = f - defer func() { _ = fixture.Close(context.Background()) }() + defer func() { _ = fixture.Close() }() fmt.Fprintf(os.Stderr, "gitea fixture: image=%s version=%s\n", fixture.image, fixture.version) diff --git a/e2e/modes_test.go b/e2e/modes_test.go index 72841b1f..e46a3b6a 100644 --- a/e2e/modes_test.go +++ b/e2e/modes_test.go @@ -9,6 +9,7 @@ import ( "bufio" "bytes" "cmp" + "context" "encoding/json" "errors" "os" @@ -138,14 +139,31 @@ func daemonWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { return result } -func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { +func runExec(t *testing.T, env []string, args ...string) (stdout, stderr []byte, err error) { t.Helper() - ctx := t.Context() - bin, err := runnerBinary() if err != nil { t.Fatal(err) } + ctx, cancel := context.WithTimeout(t.Context(), runTimeout) + defer cancel() + cmd := exec.CommandContext(ctx, bin, append([]string{"exec"}, args...)...) + cmd.Env = append(slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }), env...) + cmd.Cancel = func() error { return cmd.Process.Signal(os.Interrupt) } + cmd.WaitDelay = cleanupTimeout + var stdoutBuf, stderrBuf bytes.Buffer + cmd.Stdout, cmd.Stderr = &stdoutBuf, &stderrBuf + err = cmd.Run() + if ctx.Err() != nil { + t.Fatalf("exec did not finish within %s\n%s%s", runTimeout, stdoutBuf.Bytes(), stderrBuf.Bytes()) + } + return stdoutBuf.Bytes(), stderrBuf.Bytes(), err +} + +func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { + t.Helper() + ctx := t.Context() + workdir := t.TempDir() workflows := filepath.Join(workdir, ".gitea", "workflows") if err := os.MkdirAll(workflows, 0o755); err != nil { @@ -170,12 +188,12 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { } } - args := []string{"exec", "--json", "-C", workdir, "-W", workflows, "--artifact-server-path", t.TempDir(), "--artifact-server-port", "0"} - env := slices.DeleteFunc(os.Environ(), func(kv string) bool { return strings.HasPrefix(kv, "ACTIONS_") }) + args := []string{"--json", "-C", workdir, "-W", workflows, "--artifact-server-path", t.TempDir(), "--artifact-server-port", "0"} + var env []string if m.kubernetes { cluster := requireCluster(t) args = append(args, "-i", "kubernetes://"+cluster.jobImage) - env = append(env, "KUBECONFIG="+cluster.kubernetes.Kubeconfig) + env = []string{"KUBECONFIG=" + cluster.kubernetes.Kubeconfig} } else { args = append(args, "-i", os.Getenv("E2E_JOB_IMAGE"), "--network", fixture.network) } @@ -185,21 +203,17 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { args = append(args, "--input", name+"="+value) } } - cmd := exec.CommandContext(ctx, bin, append(args, spec.execArgs...)...) - cmd.Env = env - var stdout, stderr bytes.Buffer - cmd.Stdout, cmd.Stderr = &stdout, &stderr - runErr := cmd.Run() + stdout, stderr, runErr := runExec(t, env, append(args, spec.execArgs...)...) var exitErr *exec.ExitError if runErr != nil && !errors.As(runErr, &exitErr) { - t.Fatalf("exec: %v\n%s", runErr, stderr.String()) + t.Fatalf("exec: %v\n%s", runErr, stderr) } result := &workflowResult{conclusion: "success", jobs: map[string]string{}} if runErr != nil { result.conclusion = "failure" } - scanner := bufio.NewScanner(&stdout) + scanner := bufio.NewScanner(bytes.NewReader(stdout)) scanner.Buffer(nil, 1<<20) for scanner.Scan() { var entry struct { @@ -220,8 +234,8 @@ func execWorkflow(t *testing.T, m mode, spec workflowSpec) *workflowResult { } } } - if result.conclusion != "success" && len(result.jobs) == 0 { - t.Fatalf("exec ran no job: %v\n%s", runErr, stderr.String()) + if len(result.jobs) == 0 { + t.Fatalf("exec ran no job: %v\n%s", runErr, stderr) } return result } diff --git a/e2e/testdata/workflows/cli_labels.yml b/e2e/testdata/workflows/cli_labels.yml deleted file mode 100644 index dfc4eb7c..00000000 --- a/e2e/testdata/workflows/cli_labels.yml +++ /dev/null @@ -1,7 +0,0 @@ -name: cli-labels -on: push -jobs: - hello: - runs-on: e2e-override - steps: - - run: echo e2e-override-ran