From 0cedd98051d6faf1be5637ff61b75d6014fd028f Mon Sep 17 00:00:00 2001 From: silverwind Date: Thu, 1 Oct 2026 06:59:15 +0000 Subject: [PATCH] feat: add Kubernetes backend (#1260) Adds a Kubernetes backend. A `name:kubernetes://image` label runs the job as one pod through the Kubernetes API, with services as containers of the same pod. Step env is exported by a script sent over exec stdin, keeping it out of the pod spec and process arguments, and service env goes into a per-job Secret. Docker actions and container options are rejected. See `docs/kubernetes.md`. Operators shape job pods with `kubernetes.pod_template`, with per-label `pod_templates` merged in `runs-on` order and a `$services` entry for service containers. The idle cleanup removes pods and Secrets a runner left behind. The client calls the Kubernetes API directly and runs step commands over `coder/websocket` (same dependency used in gitea) instead of pulling in the heavyweight `kubernetes/client-go`. Includes an e2e test against k3s. Replaces: https://gitea.com/gitea/runner/pulls/1000 Fixes: https://gitea.com/gitea/runner/issues/31 Assisted-by: Claude Code:claude-opus-5-5 Co-authored-by: Zettat123 Co-authored-by: bircni Reviewed-on: https://gitea.com/gitea/runner/pulls/1260 Reviewed-by: Lunny Xiao Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> Co-authored-by: silverwind --- Makefile | 3 +- README.md | 17 +- act/container/container_types.go | 3 + act/container/docker_network.go | 4 - act/container/docker_run.go | 64 +-- act/container/kubernetes.go | 576 +++++++++++++++++++ act/container/kubernetes_client.go | 400 +++++++++++++ act/container/kubernetes_test.go | 290 ++++++++++ act/container/tar.go | 69 +++ act/runner/action.go | 3 + act/runner/action_composite.go | 1 + act/runner/run_context.go | 269 +++++---- act/runner/run_context_test.go | 5 +- act/runner/runner.go | 1 + act/runner/step_action_local.go | 3 + act/runner/step_docker.go | 3 + docs/kubernetes.md | 58 ++ e2e/compatibility_test.go | 1 + e2e/gitea_fixture.go | 22 +- e2e/kubernetes_test.go | 166 ++++++ e2e/runner_harness.go | 17 +- e2e/testdata/workflows/kubernetes.yml | 20 + examples/kubernetes/README.md | 27 +- examples/kubernetes/kubernetes-pods.yaml | 83 +++ go.mod | 1 + go.sum | 2 + internal/app/cmd/exec.go | 2 +- internal/app/run/runner.go | 53 +- internal/app/run/runner_idle_cleanup_test.go | 17 +- internal/app/run/runner_test.go | 12 +- internal/pkg/config/config.example.yaml | 16 +- internal/pkg/config/config.go | 13 +- internal/pkg/labels/labels.go | 13 +- internal/pkg/labels/labels_test.go | 2 + tools/test-e2e.sh | 2 + 35 files changed, 2012 insertions(+), 226 deletions(-) create mode 100644 act/container/kubernetes.go create mode 100644 act/container/kubernetes_client.go create mode 100644 act/container/kubernetes_test.go create mode 100644 act/container/tar.go create mode 100644 docs/kubernetes.md create mode 100644 e2e/kubernetes_test.go create mode 100644 e2e/testdata/workflows/kubernetes.yml create mode 100644 examples/kubernetes/kubernetes-pods.yaml diff --git a/Makefile b/Makefile index 332b850f..b5c38f02 100644 --- a/Makefile +++ b/Makefile @@ -12,6 +12,7 @@ PODMAN_TEST_IMAGE ?= quay.io/containers/podman:v5.8.4-immutable@sha256:90405e3c6 E2E_JOB_IMAGE ?= node:24-bookworm@sha256:64af3819f9275802414d7cdc38c27e9d82bd564dec4d4da87d008255d36c63b4 # renovate: datasource=docker SERVICE_IMAGE ?= nginx:1.31.4-alpine@sha256:db35bfc6b2951e7f8a72db5db120288c127ffaeeb4a6d4b95a26fead017d5913 # renovate: datasource=docker E2E_GITEA_IMAGE ?= gitea/gitea:main-nightly +E2E_K3S_IMAGE ?= rancher/k3s:v1.36.4-k3s1@sha256:edad48e12bf81c3a09ac1c05c0c0ffaaa22145980b989d6fae84543a76b83657 # renovate: datasource=docker LINUX_ARCHS ?= linux/amd64,linux/arm64 DARWIN_ARCHS ?= darwin-12/amd64,darwin-12/arm64 @@ -170,7 +171,7 @@ E2E_CONCURRENCY ?= 8 .PHONY: test-e2e test-e2e: ## run Gitea compatibility tests against E2E_GITEA_IMAGE - @E2E_CONCURRENCY=$(E2E_CONCURRENCY) E2E_GITEA_IMAGE=$(E2E_GITEA_IMAGE) E2E_JOB_IMAGE=$(E2E_JOB_IMAGE) GO=$(GO) SERVICE_IMAGE=$(SERVICE_IMAGE) ./tools/test-e2e.sh + @E2E_CONCURRENCY=$(E2E_CONCURRENCY) E2E_GITEA_IMAGE=$(E2E_GITEA_IMAGE) E2E_K3S_IMAGE=$(E2E_K3S_IMAGE) E2E_JOB_IMAGE=$(E2E_JOB_IMAGE) GO=$(GO) SERVICE_IMAGE=$(SERVICE_IMAGE) ./tools/test-e2e.sh .PHONY: install install: $(GOFILES) ## install the runner binary via `go install` diff --git a/README.md b/README.md index 09c17a01..2667f8ee 100644 --- a/README.md +++ b/README.md @@ -172,10 +172,10 @@ A label is written as: | Part | Meaning | | --- | --- | | `name` | The name a workflow refers to in `runs-on`, e.g. `ubuntu-latest`. | -| `schema` | Either `docker` or `host`. Defaults to `host` when omitted. | -| `args` | Only used by the `docker` schema: the image to run the job in. | +| `schema` | `docker`, `kubernetes` or `host`. Defaults to `host` when omitted. | +| `args` | Only used by the `docker` and `kubernetes` schemas: the image to run the job in. | -Two schemas are supported: +Three schemas are supported: - **`docker://`** — the job runs inside a container created from ``: @@ -183,6 +183,12 @@ Two schemas are supported: ubuntu-latest:docker://docker.gitea.com/runner-images:ubuntu-latest ``` +- **`kubernetes://`** — the job runs as a Kubernetes pod created from ``, see **[docs/kubernetes.md](docs/kubernetes.md)**: + + ```text + ubuntu-latest:kubernetes://docker.gitea.com/runner-images:ubuntu-latest + ``` + - **`host`** — the job's steps run directly on the machine the runner is on, using the tools installed there: ```text @@ -197,9 +203,9 @@ ubuntu-latest:docker://docker.gitea.com/runner-images:ubuntu-latest,macos:host a workflow with `runs-on: ubuntu-latest` is executed in the `runner-images:ubuntu-latest` container, and one with `runs-on: macos` is executed directly on the host. -Names may themselves contain a colon (for example `pool:e57e18d4-10d4-406f-93bf-60f127221bdd`); only `host` and `docker` are treated as schemas. +Names may themselves contain a colon, for example `pool:e57e18d4-10d4-406f-93bf-60f127221bdd`. Only `host`, `docker` and `kubernetes` are treated as schemas. -If a job's `runs-on` matches none of the runner's labels, or sets no `runs-on` at all, it still runs: in `runner.default_image` where docker is available, on the host where it is not. Images maintained for this purpose are listed at [gitea/runner-images](https://gitea.com/gitea/runner-images). +If a job's `runs-on` matches none of the runner's labels, or sets no `runs-on` at all, it still runs in `runner.default_image`: as a container when the runner has `docker` labels, as a pod when it has `kubernetes` labels, else as a container where docker is available and on the host where it is not. Images maintained for this purpose are listed at [gitea/runner-images](https://gitea.com/gitea/runner-images). Labels are chosen at registration time (`--labels`, or the interactive prompt) and can be changed afterwards by editing `runner.labels` in the config file, or in the Gitea UI under the runner's settings. @@ -351,6 +357,7 @@ While the runner is idle it cleans up after earlier jobs: - only purely numeric subdirectories under `container.workdir_parent` are treated as task workspaces and may be removed - cleanup assumes `container.workdir_parent` is not shared across multiple runners - on runners that use docker, per-job networks left behind by jobs the runner did not live to tear down are removed, identified by the `com.gitea.runner.uuid` label carrying this runner's uuid +- on runners with `kubernetes` labels, job pods with that label older than `runner.workdir_cleanup_age` are removed the same way, together with their Secrets - cleanup runs every `runner.idle_cleanup_interval` (default: `10m`; set `0` to disable), and setting either knob to `0` disables all of the above #### Post-task script (`runner.post_task_script`) diff --git a/act/container/container_types.go b/act/container/container_types.go index cdcbdb42..50bc4a86 100644 --- a/act/container/container_types.go +++ b/act/container/container_types.go @@ -75,6 +75,9 @@ const ( HealthUnhealthy = string(container.Unhealthy) ) +// marks what a runner creates for its jobs, telling its leftovers from another runner's +const runnerUUIDLabel = "com.gitea.runner.uuid" + // 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") diff --git a/act/container/docker_network.go b/act/container/docker_network.go index f61a476b..aac7599b 100644 --- a/act/container/docker_network.go +++ b/act/container/docker_network.go @@ -24,10 +24,6 @@ import ( const ( networkCreateAttempts = 3 networkCreateRetryDelay = time.Second - - // marks the networks a runner creates for its jobs, so it can tell its own leftovers from - // those of another runner sharing the daemon - runnerUUIDLabel = "com.gitea.runner.uuid" ) // RemoveOrphanNetworks removes the networks this runner created for jobs whose teardown did diff --git a/act/container/docker_run.go b/act/container/docker_run.go index c6c70c4b..5b3f7814 100644 --- a/act/container/docker_run.go +++ b/act/container/docker_run.go @@ -7,7 +7,6 @@ package container import ( - "archive/tar" "bytes" "context" "errors" @@ -25,16 +24,12 @@ import ( "time" "gitea.com/gitea/runner/act/common" - "gitea.com/gitea/runner/act/filecollector" "dario.cat/mergo" "github.com/bmatcuk/doublestar/v4" cerrdefs "github.com/containerd/errdefs" "github.com/docker/cli/cli/compose/loader" "github.com/docker/cli/cli/connhelper" - "github.com/go-git/go-billy/v5/helper/polyfill" - "github.com/go-git/go-billy/v5/osfs" - "github.com/go-git/go-git/v5/plumbing/format/gitignore" "github.com/joho/godotenv" "github.com/moby/moby/api/pkg/stdcopy" "github.com/moby/moby/api/types/container" @@ -982,42 +977,7 @@ func (cr *containerReference) copyDir(dstPath, srcPath string, useGitIgnore, ski logger.Error(err) } }(tarFile) - tw := tar.NewWriter(tarFile) - - srcPrefix := filepath.Dir(srcPath) - if !strings.HasSuffix(srcPrefix, string(filepath.Separator)) { - srcPrefix += string(filepath.Separator) - } - logger.Debugf("Stripping prefix:%s src:%s", srcPrefix, srcPath) - - var ignorer gitignore.Matcher - if useGitIgnore { - ps, err := gitignore.ReadPatterns(polyfill.New(osfs.New(srcPath)), nil) - if err != nil { - logger.Debugf("Error loading .gitignore: %v", err) - } - - ignorer = gitignore.NewMatcher(ps) - } - - fc := &filecollector.FileCollector{ - Ignorer: ignorer, - SrcPath: srcPath, - SrcPrefix: srcPrefix, - SkipGitDir: skipGitDir, - Handler: &filecollector.TarCollector{ - TarWriter: tw, - UID: cr.UID, - GID: cr.GID, - DstDir: dstPath[1:], - }, - } - - err = filepath.Walk(srcPath, fc.CollectFiles(ctx, []string{})) - if err != nil { - return err - } - if err := tw.Close(); err != nil { + if err := writeDirTar(ctx, tarFile, dstPath, srcPath, useGitIgnore, skipGitDir, cr.UID, cr.GID); err != nil { return err } @@ -1042,30 +1002,12 @@ func (cr *containerReference) copyContent(dstPath string, files ...*FileEntry) c if cr.id == "" { return cr.missingContainerError("copy to %s", dstPath) } - logger := common.Logger(ctx) var buf bytes.Buffer - tw := tar.NewWriter(&buf) - for _, file := range files { - logger.Debugf("Writing entry to tarball %s len:%d", file.Name, len(file.Body)) - hdr := &tar.Header{ - Name: file.Name, - Mode: file.Mode, - Size: int64(len(file.Body)), - Uid: cr.UID, - Gid: cr.GID, - } - if err := tw.WriteHeader(hdr); err != nil { - return err - } - if _, err := tw.Write([]byte(file.Body)); err != nil { - return err - } - } - if err := tw.Close(); err != nil { + if err := writeFilesTar(ctx, &buf, cr.UID, cr.GID, files...); err != nil { return err } - logger.Debugf("Extracting content to '%s'", dstPath) + common.Logger(ctx).Debugf("Extracting content to '%s'", dstPath) _, err := cr.cli.CopyToContainer(ctx, cr.id, client.CopyToContainerOptions{ DestinationPath: dstPath, Content: &buf, diff --git a/act/container/kubernetes.go b/act/container/kubernetes.go new file mode 100644 index 00000000..247732a9 --- /dev/null +++ b/act/container/kubernetes.go @@ -0,0 +1,576 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "bufio" + "bytes" + "cmp" + "context" + "crypto/rand" + "errors" + "fmt" + "io" + "maps" + "net/http" + "net/url" + "os" + "path" + "regexp" + "slices" + "strings" + "time" + + "gitea.com/gitea/runner/act/common" + + "github.com/kballard/go-shellquote" + "github.com/spf13/pflag" +) + +// KubernetesOptions configures the pod a job runs in. +type KubernetesOptions struct { + Kubeconfig string + Namespace string + PodTemplates []map[string]any // merged in order, a later one winning + MaxLifetime time.Duration + ForcePull bool + RunnerUUID string +} + +// KubernetesPod runs a job container and its service containers as one pod. +type KubernetesPod struct { + options KubernetesOptions + client *kubeClient + name, secretName string + job *kubernetesContainer + services []*kubernetesContainer + arch, path string +} + +type kubernetesContainer struct { + LinuxContainerEnvironmentExtensions + pod *KubernetesPod + id string // the workflow's service id, name is its DNS label form + name string + input *NewContainerInput + probe map[string]any +} + +type containerStatus struct { + Name string + Ready bool + State struct { + Waiting *struct{ Reason, Message string } + Running *struct{} + Terminated *struct { + ExitCode int + Message string + } + } +} + +type podStatus struct { + Spec struct{ Containers []struct{} } + Status struct { + Phase, Reason, Message string + Conditions []struct{ Type, Status, Message string } + InitContainerStatuses, ContainerStatuses []containerStatus + } +} + +func NewKubernetesPod(options KubernetesOptions) (*KubernetesPod, error) { + client, err := newKubeClient(options.Kubeconfig, options.Namespace) + if err != nil { + return nil, fmt.Errorf("connect to Kubernetes: %w", err) + } + return &KubernetesPod{options: options, client: client}, nil +} + +// JobContainer returns the container of the pod the job's steps run in. +func (p *KubernetesPod) JobContainer(input *NewContainerInput) ExecutionsEnvironment { + p.job = &kubernetesContainer{pod: p, name: "job", input: input} + return p.job +} + +var invalidLabelChars = regexp.MustCompile(`[^a-z0-9-]+`) + +// KubernetesServiceName returns the container name and hostname of service id, made a DNS label as both must be. +func KubernetesServiceName(id string) string { + name := strings.Trim(invalidLabelChars.ReplaceAllString(strings.ToLower(id), "-"), "-") + return strings.TrimRight(name[:min(63, len(name))], "-") +} + +// ServiceContainer returns a container of the pod the job reaches at hostname KubernetesServiceName(id). +func (p *KubernetesPod) ServiceContainer(id string, input *NewContainerInput) ExecutionsEnvironment { + service := &kubernetesContainer{pod: p, id: id, name: KubernetesServiceName(id), input: input} + p.services = append(p.services, service) + return service +} + +func (p *KubernetesPod) create(ctx context.Context) error { + prefix := strings.Trim(strings.ToLower(p.job.input.Name), "-") + prefix = cmp.Or(strings.TrimRight(prefix[:min(40, len(prefix))], "-"), "gitea-runner") + "-" + name := prefix + strings.ToLower(rand.Text()[:8]) // chosen here, so cleanup knows it even when a create response is lost + labels := map[string]any{"app.kubernetes.io/managed-by": "gitea-runner"} + if p.options.RunnerUUID != "" { + labels[runnerUUIDLabel] = p.options.RunnerUUID + } + metadata := map[string]any{"name": name, "labels": labels} + if hostname, err := os.Hostname(); p.client.inCluster && err == nil { + var runner struct{ Metadata struct{ Name, UID string } } + if err := p.client.do(ctx, http.MethodGet, p.client.path("pods/"+hostname), nil, &runner); err == nil { + metadata["ownerReferences"] = []any{map[string]any{"apiVersion": "v1", "kind": "Pod", "name": runner.Metadata.Name, "uid": runner.Metadata.UID}} + } else { + common.Logger(ctx).Debugf("The runner is not pod %s, job pods get no owner: %v", hostname, err) + } + } + secretData := map[string]string{} + manifest, err := p.manifest(name, secretData, metadata) + if err != nil { + return err + } + for _, service := range p.services { + if !strings.EqualFold(service.id, service.name) { + common.Logger(ctx).Warnf("Service %s is reachable as %s or localhost, as a Kubernetes hostname cannot be %q", service.id, service.name, service.id) + } + } + if len(secretData) > 0 { + p.secretName = name // named like the pod, RemoveOrphanKubernetesResources finds it by that + if err := p.client.do(ctx, http.MethodPost, p.client.path("secrets"), map[string]any{"metadata": metadata, "stringData": secretData}, nil); err != nil { + return fmt.Errorf("create secret: %w", err) + } + } + p.name = name + if err := p.client.do(ctx, http.MethodPost, p.client.path("pods"), manifest, nil); err != nil { + return fmt.Errorf("create pod: %w", err) + } + return nil +} + +func (p *KubernetesPod) manifest(secret string, secretData map[string]string, metadata map[string]any) (any, error) { + template := map[string]any{"spec": map[string]any{ + "automountServiceAccountToken": false, + "enableServiceLinks": false, + "terminationGracePeriodSeconds": 0, // the job container's tail ignores SIGTERM + }} + for _, layer := range p.options.PodTemplates { + template, _ = mergeManifest(template, layer, "").(map[string]any) + } + templateSpec, _ := template["spec"].(map[string]any) + templateContainers, _ := templateSpec["containers"].([]any) + templateVolumes, _ := templateSpec["volumes"].([]any) + var serviceTemplate any + if index := slices.IndexFunc(templateContainers, func(container any) bool { return manifestField(container, "name") == "$services" }); index >= 0 { + serviceTemplate = templateContainers[index] + templateSpec["containers"] = slices.Delete(slices.Clone(templateContainers), index, index+1) + } + var containers, hostnames, volumes []any + names := map[string]bool{} + for _, container := range templateContainers { + names[manifestField(container, "name")] = true + } + for _, c := range append([]*kubernetesContainer{p.job}, p.services...) { + if c != p.job && (c.name == "" || names[c.name]) { + return nil, fmt.Errorf("service %s maps to container %q, which is empty or named like another container of the pod", c.id, c.name) + } + names[c.name] = true + spec, err := c.spec(secret, secretData) + if err != nil { + return nil, err + } + if c == p.job { + containers = append(containers, spec) + continue + } + containers = append(containers, mergeManifest(serviceTemplate, spec, "")) + hostnames = append(hostnames, c.name) + } + for _, name := range []string{"workspace", "act"} { + if !slices.ContainsFunc(templateVolumes, func(volume any) bool { return manifestField(volume, "name") == name }) { + volumes = append(volumes, map[string]any{"name": name, "emptyDir": map[string]any{}}) + } + } + spec := map[string]any{"restartPolicy": "Never", "containers": containers, "volumes": volumes} + if len(hostnames) > 0 { + spec["hostAliases"] = []any{map[string]any{"ip": "127.0.0.1", "hostnames": hostnames}} + } + if p.options.MaxLifetime > 0 { + spec["activeDeadlineSeconds"] = max(1, int64(p.options.MaxLifetime.Seconds())) + } + return mergeManifest(template, map[string]any{"apiVersion": "v1", "kind": "Pod", "metadata": metadata, "spec": spec}, ""), nil +} + +// mergeManifest merges over onto base like a strategic merge patch: maps key by key, map list items by name or volumeMounts by mountPath, env entries, volumes and other values replaced. +func mergeManifest(base, over any, field string) any { + switch over := over.(type) { + case map[string]any: + merged := map[string]any{} + if base, ok := base.(map[string]any); ok { + maps.Copy(merged, base) + } + for key, value := range over { + merged[key] = mergeManifest(merged[key], value, key) + } + return merged + case []any: + key := cmp.Or(map[string]string{"volumeMounts": "mountPath"}[field], "name") + baseList, _ := base.([]any) + merged := slices.Clone(baseList) + for _, item := range over { + if _, ok := item.(map[string]any); !ok { + return over + } + name := manifestField(item, key) + index := slices.IndexFunc(merged, func(existing any) bool { return name != "" && manifestField(existing, key) == name }) + switch { + case index < 0: + merged = append(merged, item) + case field == "env" || field == "volumes": // each holds one source, merging two makes a pod the API rejects + merged[index] = item + default: + merged[index] = mergeManifest(merged[index], item, "") + } + } + return merged + } + return over +} + +func manifestField(item any, key string) string { + fields, _ := item.(map[string]any) + value, _ := fields[key].(string) + return value +} + +func (c *kubernetesContainer) spec(secret string, secretData map[string]string) (map[string]any, error) { + env := []any{} + for index, entry := range c.input.Env { + key, value, _ := strings.Cut(entry, "=") + if c == c.pod.job { + env = append(env, map[string]any{"name": key, "value": value}) + continue + } + secretKey := fmt.Sprintf("%s.%d", c.name, index) + secretData[secretKey] = value + env = append(env, map[string]any{"name": key, "valueFrom": map[string]any{"secretKeyRef": map[string]any{"name": secret, "key": secretKey}}}) + } + spec := map[string]any{ + "name": c.name, "image": c.input.Image, "command": escapeExpansion(c.input.Entrypoint), "args": escapeExpansion(c.input.Cmd), + "workingDir": c.input.WorkingDir, "env": env, + } + if c.pod.options.ForcePull { + spec["imagePullPolicy"] = "Always" + } + if c == c.pod.job { + spec["volumeMounts"] = []any{ + map[string]any{"name": "workspace", "mountPath": c.input.WorkingDir}, + map[string]any{"name": "act", "mountPath": c.GetActPath()}, + } + } + if err := c.parseOptions(); err != nil { + return nil, fmt.Errorf("service %s: %w", c.name, err) + } + if c.probe != nil { + spec["startupProbe"] = c.probe + } + return spec, nil +} + +// escapeExpansion keeps kubelet from expanding $(VAR) and $$ in command and args, which docker passes as is. +func escapeExpansion(args []string) []string { + args = slices.Clone(args) + for index := range args { + args[index] = strings.ReplaceAll(args[index], "$", "$$") + } + return args +} + +// parseOptions turns docker health options, the only ones a pod supports, into a startup probe, which like docker gives up after the retries. +func (c *kubernetesContainer) parseOptions() error { + args, err := shellquote.Split(c.input.WorkflowOptions) + if err != nil { + return err + } + flags := pflag.NewFlagSet("options", pflag.ContinueOnError) + flags.SetOutput(io.Discard) + command := flags.String("health-cmd", "", "") + interval := flags.Duration("health-interval", 0, "") + timeout := flags.Duration("health-timeout", 0, "") + retries := flags.Int("health-retries", 0, "") + startPeriod := flags.Duration("health-start-period", 0, "") + if err := flags.Parse(args); err != nil { + return fmt.Errorf("options %q: %w, only --health-* options are supported on Kubernetes", c.input.WorkflowOptions, err) + } + if *command == "" { + return nil + } + period := max(1, int(cmp.Or(*interval, 30*time.Second).Seconds())) + c.probe = map[string]any{ + "exec": map[string]any{"command": []string{"/bin/sh", "-c", *command}}, + "periodSeconds": period, + "timeoutSeconds": max(1, int(cmp.Or(*timeout, 30*time.Second).Seconds())), + "failureThreshold": max(1, cmp.Or(*retries, 3)) + int(startPeriod.Seconds())/period, // docker does not count failures in the start period + } + return nil +} + +func (p *KubernetesPod) get(ctx context.Context) (*podStatus, error) { + var pod podStatus + return &pod, p.client.do(ctx, http.MethodGet, p.client.path("pods/"+p.name), nil, &pod) +} + +func (p *KubernetesPod) waitStarted(ctx context.Context) error { + var unscheduled string + for delay := 100 * time.Millisecond; ; delay = min(2*delay, time.Second) { + pod, err := p.get(ctx) + if err != nil { + return err + } + if pod.Status.Phase == "Failed" { + return fmt.Errorf("pod %s failed: %s", p.name, cmp.Or(pod.Status.Message, pod.Status.Reason)) + } + started := len(pod.Status.ContainerStatuses) == len(pod.Spec.Containers) + for _, status := range append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...) { + if waiting := status.State.Waiting; waiting != nil { + switch waiting.Reason { + case "ErrImagePull", "ImagePullBackOff", "InvalidImageName", "ErrImageNeverPull", "CreateContainerConfigError", "CreateContainerError", "RunContainerError": + return fmt.Errorf("container %s of pod %s: %s: %s", status.Name, p.name, waiting.Reason, waiting.Message) + } + started = false + } + } + if started { + return nil + } + for _, condition := range pod.Status.Conditions { + if condition.Type == "PodScheduled" && condition.Status == "False" && condition.Message != unscheduled { + unscheduled = condition.Message + common.Logger(ctx).Infof("Waiting for pod %s to be scheduled: %s", p.name, unscheduled) + } + } + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(delay): + } + } +} + +func (c *kubernetesContainer) Create(_, _ []string) common.Executor { + return func(ctx context.Context) error { + if c != c.pod.job { + return nil + } + return c.pod.create(ctx) + } +} + +func (c *kubernetesContainer) Start(_ bool) common.Executor { + return func(ctx context.Context) error { + if c != c.pod.job { + return nil + } + if err := c.pod.waitStarted(ctx); err != nil { + return err + } + var stdout bytes.Buffer + if err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", `printf '%s\n%s' "$(uname -m)" "$PATH"`}, nil, &stdout, io.Discard); err != nil { + return err + } + machine, path, _ := strings.Cut(stdout.String(), "\n") + c.pod.arch, c.pod.path = goArchToActionArch(machine), path + return nil + } +} + +func (*kubernetesContainer) Pull(bool) common.Executor { + return func(context.Context) error { return nil } +} + +var ( + shellName = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) + bashReadonly = []string{"BASHOPTS", "BASH_VERSINFO", "EUID", "PPID", "SHELLOPTS", "UID"} // bash as sh cannot export these +) + +// Exec feeds the command to sh on stdin, which exports the env so it stays out of the request URL and, where sh can export a name, out of process arguments. +func (c *kubernetesContainer) Exec(command []string, env map[string]string, _, workdir string) common.Executor { + return func(ctx context.Context) error { + script, args := "", []string{"env", "--"} + for name, value := range env { + if shellName.MatchString(name) && !slices.Contains(bashReadonly, name) { + script += "command export " + shellquote.Join(name+"="+value) + "\n" + } else { + args = append(args, name+"="+value) + } + } + script += "exec " + shellquote.Join(append(args, command...)...) + if workdir != "" { + script = "cd " + shellquote.Join(workdir) + " || exit\n" + script + } + defer common.FlushWriter(c.input.Stdout) + defer common.FlushWriter(c.input.Stderr) + return c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh"}, strings.NewReader(script), c.input.Stdout, c.input.Stderr) + } +} + +func (c *kubernetesContainer) extract(ctx context.Context, destPath string, write func(io.Writer) error) error { + reader, writer := io.Pipe() + defer reader.Close() + go func() { writer.CloseWithError(write(writer)) }() + var stderr bytes.Buffer + if err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", `mkdir -p "$1" && tar -xf - -C "$1"`, "sh", destPath}, reader, io.Discard, &stderr); err != nil { + return fmt.Errorf("extract to %s: %w: %s", destPath, err, strings.TrimSpace(stderr.String())) + } + return nil +} + +func (c *kubernetesContainer) Copy(destPath string, files ...*FileEntry) common.Executor { + return func(ctx context.Context) error { + return c.extract(ctx, destPath, func(writer io.Writer) error { return writeFilesTar(ctx, writer, 0, 0, files...) }) + } +} + +func (c *kubernetesContainer) CopyDir(destPath, srcPath string, useGitIgnore, skipGitDir bool) common.Executor { + return func(ctx context.Context) error { + return c.extract(ctx, "/", func(writer io.Writer) error { + return writeDirTar(ctx, writer, destPath, srcPath, useGitIgnore, skipGitDir, 0, 0) + }) + } +} + +// GetContainerArchive streams the archive, checking the path first as tar writes an empty archive for a missing one. +func (c *kubernetesContainer) GetContainerArchive(ctx context.Context, srcPath string) (io.ReadCloser, error) { + reader, writer := io.Pipe() + go func() { + var stderr bytes.Buffer + err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", `test -e "$1" -o -L "$1" && exec tar -cf - -C "$2" "$3"`, "sh", srcPath, path.Dir(srcPath), path.Base(srcPath)}, nil, writer, &stderr) + if err != nil { + err = fmt.Errorf("archive %s: %w: %s", srcPath, err, strings.TrimSpace(stderr.String())) + } + writer.CloseWithError(err) + }() + buffered := bufio.NewReader(reader) + if _, err := buffered.Peek(1); err != nil { + reader.Close() + return nil, err + } + return struct { + io.Reader + io.Closer + }{buffered, reader}, nil +} + +func (c *kubernetesContainer) UpdateFromEnv(srcPath string, env *map[string]string) common.Executor { + return parseEnvFile(c, srcPath, env) +} + +func (c *kubernetesContainer) UpdateFromImageEnv(env *map[string]string) common.Executor { + return func(context.Context) error { + (*env)["PATH"] = cmp.Or((*env)["PATH"], c.pod.path) + return nil + } +} + +func (c *kubernetesContainer) Inspect(ctx context.Context) (*Info, error) { + if c.pod.name == "" { + return nil, fmt.Errorf("container %s %w", c.name, ErrContainerNotFound) + } + pod, err := c.pod.get(ctx) + if isNotFound(err) { + return nil, fmt.Errorf("pod %s %w", c.pod.name, ErrContainerNotFound) + } else if err != nil { + return nil, err + } + info := &Info{ID: c.pod.name + "/" + c.name, State: "created", Health: HealthNone, Ports: map[string]string{}} + for port := range c.input.ExposedPorts { + info.Ports[port.Port()] = port.Port() // services share the pod's network + } + for _, status := range pod.Status.ContainerStatuses { + if status.Name != c.name { + continue + } + if status.State.Running != nil { + info.State = StateRunning + } else if terminated := status.State.Terminated; terminated != nil { + info.State, info.ExitCode, info.HealthOutput = "exited", terminated.ExitCode, terminated.Message + } + if c.probe != nil { + switch { + case status.Ready: + info.Health = HealthHealthy + case info.State == "exited": + info.Health = HealthUnhealthy + default: + info.Health = HealthStarting + } + } + } + return info, nil +} + +func (c *kubernetesContainer) DumpLogs(ctx context.Context) error { + if c.pod.name == "" { + return fmt.Errorf("container %s %w", c.name, ErrContainerNotFound) + } + defer common.FlushWriter(c.input.Stdout) + return c.pod.client.do(ctx, http.MethodGet, c.pod.client.path("pods/"+c.pod.name+"/log?container="+c.name), nil, c.input.Stdout) +} + +func (c *kubernetesContainer) Remove() common.Executor { + return func(ctx context.Context) error { + if c != c.pod.job { + return nil + } + return errors.Join(c.pod.client.delete(ctx, "pods", c.pod.name), c.pod.client.delete(ctx, "secrets", c.pod.secretName)) + } +} + +// RemoveOrphanKubernetesResources removes this runner's job pods created before createdBefore and the Secrets named like them, as listing Secrets would reveal all of the namespace's. +func RemoveOrphanKubernetesResources(ctx context.Context, options KubernetesOptions, createdBefore time.Time) error { + client, err := newKubeClient(options.Kubeconfig, options.Namespace) + if err != nil { + return fmt.Errorf("connect to Kubernetes: %w", err) + } + defer client.http.CloseIdleConnections() + var pods struct { + Items []struct { + Metadata struct { + Name string + CreationTimestamp time.Time + } + } + } + if err := client.do(ctx, http.MethodGet, client.path("pods?labelSelector="+url.QueryEscape(runnerUUIDLabel+"="+options.RunnerUUID)), nil, &pods); err != nil { + return fmt.Errorf("list pods: %w", err) + } + var errs []error + for _, item := range pods.Items { + if !item.Metadata.CreationTimestamp.Before(createdBefore) { + continue + } + if err := client.delete(ctx, "secrets", item.Metadata.Name); err != nil { + errs = append(errs, err) // leave the pod for the next cleanup, it is the only way back to its Secret + continue + } + errs = append(errs, client.delete(ctx, "pods", item.Metadata.Name)) + } + return errors.Join(errs...) +} + +func (c *kubernetesContainer) Close() common.Executor { + return func(context.Context) error { + c.pod.client.http.CloseIdleConnections() + return nil + } +} + +func (c *kubernetesContainer) ReplaceLogWriter(stdout, stderr io.Writer) (io.Writer, io.Writer) { + oldStdout, oldStderr := c.input.Stdout, c.input.Stderr + c.input.Stdout, c.input.Stderr = stdout, stderr + return oldStdout, oldStderr +} + +func (c *kubernetesContainer) GetRunnerContext(_ context.Context) map[string]any { + return map[string]any{"os": "Linux", "arch": c.pod.arch, "temp": "/tmp", "tool_cache": DefaultToolCache} +} diff --git a/act/container/kubernetes_client.go b/act/container/kubernetes_client.go new file mode 100644 index 00000000..61e1a1d6 --- /dev/null +++ b/act/container/kubernetes_client.go @@ -0,0 +1,400 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "bytes" + "cmp" + "context" + "crypto/tls" + "crypto/x509" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "time" + + "github.com/coder/websocket" + "go.yaml.in/yaml/v4" +) + +const ( + serviceAccountDir = "/var/run/secrets/kubernetes.io/serviceaccount" + execProtocol = "v5.channel.k8s.io" +) + +type kubeClient struct { + server, namespace, token string + tokenFile string // re-read on every request, service account tokens rotate + inCluster bool + http *http.Client +} + +type apiError struct { + code int + message string +} + +func (e *apiError) Error() string { return e.message } + +func isNotFound(err error) bool { + var apiErr *apiError + return errors.As(err, &apiErr) && apiErr.code == http.StatusNotFound +} + +func newKubeClient(kubeconfig, namespace string) (*kubeClient, error) { + client := &kubeClient{namespace: namespace} + transport := &http.Transport{Proxy: http.ProxyFromEnvironment, TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}} + if host := os.Getenv("KUBERNETES_SERVICE_HOST"); host != "" && kubeconfig == "" { + if err := client.loadInCluster(host, transport.TLSClientConfig); err != nil { + return nil, err + } + } else if err := client.loadKubeconfig(kubeconfig, transport); err != nil { + return nil, err + } + client.http = &http.Client{Transport: transport} + return client, nil +} + +func (c *kubeClient) loadInCluster(host string, tlsConfig *tls.Config) error { + c.server = "https://" + net.JoinHostPort(host, os.Getenv("KUBERNETES_SERVICE_PORT")) + c.tokenFile = serviceAccountDir + "/token" + ca, err := os.ReadFile(serviceAccountDir + "/ca.crt") + if err != nil { + return err + } + tlsConfig.RootCAs = x509.NewCertPool() + tlsConfig.RootCAs.AppendCertsFromPEM(ca) + if c.namespace == "" { + namespace, err := os.ReadFile(serviceAccountDir + "/namespace") + if err != nil { + return err + } + c.namespace = strings.TrimSpace(string(namespace)) + } + c.inCluster = true + return nil +} + +func (c *kubeClient) loadKubeconfig(path string, transport *http.Transport) error { + if paths := filepath.SplitList(os.Getenv("KUBECONFIG")); path == "" && len(paths) > 0 { + path = paths[0] + } + if path == "" { + home, err := os.UserHomeDir() + if err != nil { + return err + } + path = filepath.Join(home, ".kube", "config") + } + content, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("no in-cluster service account and no kubeconfig: %w", err) + } + var config struct { + CurrentContext string `yaml:"current-context"` + Contexts []struct { + Name string + Context struct{ Cluster, User, Namespace string } + } + Clusters []struct { + Name string + Cluster struct { + Server string + TLSServerName string `yaml:"tls-server-name"` + InsecureSkipTLSVerify bool `yaml:"insecure-skip-tls-verify"` + ProxyURL string `yaml:"proxy-url"` + CertificateAuthority string `yaml:"certificate-authority"` + CertificateAuthorityData string `yaml:"certificate-authority-data"` + } + } + Users []struct { + Name string + User struct { + Token string + TokenFile string `yaml:"tokenFile"` + ClientCertificate string `yaml:"client-certificate"` + ClientCertificateData string `yaml:"client-certificate-data"` + ClientKey string `yaml:"client-key"` + ClientKeyData string `yaml:"client-key-data"` + Exec any + AuthProvider any `yaml:"auth-provider"` + } + } + } + if err := yaml.Unmarshal(content, &config); err != nil { + return fmt.Errorf("parse kubeconfig %s: %w", path, err) + } + var current struct{ Cluster, User, Namespace string } + for _, entry := range config.Contexts { + if entry.Name == config.CurrentContext { + current = entry.Context + } + } + if current.Cluster == "" { + return fmt.Errorf("kubeconfig %s has no current context", path) + } + c.namespace = cmp.Or(c.namespace, current.Namespace, "default") + + resolve := func(file string) string { + if file == "" || filepath.IsAbs(file) { + return file + } + return filepath.Join(filepath.Dir(path), file) + } + read := func(data, file string) ([]byte, error) { + if data != "" { + return base64.StdEncoding.DecodeString(data) + } + if file == "" { + return nil, nil + } + return os.ReadFile(resolve(file)) + } + + tlsConfig := transport.TLSClientConfig + for _, cluster := range config.Clusters { + if cluster.Name != current.Cluster { + continue + } + c.server = strings.TrimSuffix(cluster.Cluster.Server, "/") + tlsConfig.ServerName, tlsConfig.InsecureSkipVerify = cluster.Cluster.TLSServerName, cluster.Cluster.InsecureSkipTLSVerify + if cluster.Cluster.ProxyURL != "" { + proxy, err := url.Parse(cluster.Cluster.ProxyURL) + if err != nil { + return fmt.Errorf("kubeconfig %s: %w", path, err) + } + transport.Proxy = http.ProxyURL(proxy) + } + ca, err := read(cluster.Cluster.CertificateAuthorityData, cluster.Cluster.CertificateAuthority) + if err != nil { + return err + } + if ca != nil { + tlsConfig.RootCAs = x509.NewCertPool() + tlsConfig.RootCAs.AppendCertsFromPEM(ca) + } + } + for _, user := range config.Users { + if user.Name != current.User { + continue + } + if user.User.Exec != nil || user.User.AuthProvider != nil { + return fmt.Errorf("kubeconfig %s: user %q authenticates through a credential plugin, which the runner does not support, use a token or client certificate", path, user.Name) + } + c.token, c.tokenFile = user.User.Token, resolve(user.User.TokenFile) + cert, err := read(user.User.ClientCertificateData, user.User.ClientCertificate) + if err != nil { + return err + } + key, err := read(user.User.ClientKeyData, user.User.ClientKey) + if err != nil { + return err + } + if cert != nil { + pair, err := tls.X509KeyPair(cert, key) + if err != nil { + return err + } + tlsConfig.Certificates = []tls.Certificate{pair} + } + } + if c.server == "" { + return fmt.Errorf("kubeconfig %s names no server for context %q", path, config.CurrentContext) + } + return nil +} + +func (c *kubeClient) authorize(header http.Header) error { + token := c.token + if c.tokenFile != "" { + content, err := os.ReadFile(c.tokenFile) + if err != nil { + return err + } + token = strings.TrimSpace(string(content)) + } + if token != "" { + header.Set("Authorization", "Bearer "+token) + } + return nil +} + +func (c *kubeClient) path(resource string) string { + return "/api/v1/namespaces/" + c.namespace + "/" + resource +} + +func (c *kubeClient) do(ctx context.Context, method, path string, in, out any) error { + var body io.Reader + if in != nil { + content, err := json.Marshal(in) + if err != nil { + return err + } + body = bytes.NewReader(content) + } + req, err := http.NewRequestWithContext(ctx, method, c.server+path, body) + if err != nil { + return err + } + if err := c.authorize(req.Header); err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.http.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode >= http.StatusMultipleChoices { + return statusError(resp) + } + if out == nil { + out = io.Discard // draining lets the connection be reused + } + if writer, ok := out.(io.Writer); ok { + _, err = io.Copy(writer, resp.Body) + return err + } + return json.NewDecoder(resp.Body).Decode(out) +} + +func (c *kubeClient) delete(ctx context.Context, resource, name string) error { + if name == "" { + return nil + } + if err := c.do(ctx, http.MethodDelete, c.path(resource+"/"+name), nil, nil); err != nil && !isNotFound(err) { + return fmt.Errorf("delete %s %s: %w", resource, name, err) + } + return nil +} + +func statusError(resp *http.Response) error { + content, _ := io.ReadAll(resp.Body) + var status struct{ Message string } + if json.Unmarshal(content, &status) != nil || status.Message == "" { + status.Message = strings.TrimSpace(string(content)) + } + return &apiError{code: resp.StatusCode, message: status.Message} +} + +func (c *kubeClient) exec(ctx context.Context, pod, containerName string, command []string, stdin io.Reader, stdout, stderr io.Writer) error { + query := url.Values{"container": {containerName}, "command": command, "stdout": {"true"}, "stderr": {"true"}} + if stdin != nil { + query.Set("stdin", "true") + } + header := http.Header{} + if err := c.authorize(header); err != nil { + return err + } + ctx, cancel := context.WithCancelCause(ctx) + defer cancel(nil) + conn, resp, err := websocket.Dial(ctx, c.server+c.path("pods/"+pod+"/exec?")+query.Encode(), &websocket.DialOptions{ + HTTPClient: c.http, + HTTPHeader: header, + Subprotocols: []string{execProtocol}, + }) + if err != nil { + if resp != nil && resp.StatusCode >= http.StatusMultipleChoices { + defer resp.Body.Close() + err = statusError(resp) + } + return fmt.Errorf("exec in pod %s: %w", pod, err) + } + defer func() { _ = conn.CloseNow() }() + if conn.Subprotocol() != execProtocol { + return fmt.Errorf("the Kubernetes API server does not speak %s, which needs Kubernetes 1.30 or later", execProtocol) + } + conn.SetReadLimit(-1) + + if stdin != nil { + go func() { + if err := sendStdin(ctx, conn, stdin); err != nil { + cancel(err) + } + }() + } + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + if conn.Ping(ctx) != nil { + return + } + } + } + }() + + for { + _, frame, err := conn.Read(ctx) + if err != nil { + return cmp.Or(context.Cause(ctx), fmt.Errorf("exec in pod %s ended without an exit status: %w", pod, err)) + } + if len(frame) < 2 { + continue + } + switch frame[0] { + case 1: + _, err = stdout.Write(frame[1:]) + case 2: + _, err = stderr.Write(frame[1:]) + case 3: + return cmp.Or(context.Cause(ctx), execStatus(frame[1:])) + } + if err != nil { + return err + } + } +} + +func sendStdin(ctx context.Context, conn *websocket.Conn, stdin io.Reader) error { + frame := make([]byte, 1+32*1024) // frame[0] is the stdin channel, 0 + for { + n, err := stdin.Read(frame[1:]) + if n > 0 { + if err := conn.Write(ctx, websocket.MessageBinary, frame[:1+n]); err != nil { + return err + } + } + if errors.Is(err, io.EOF) { + return conn.Write(ctx, websocket.MessageBinary, []byte{255, 0}) // v5 close signal for stdin + } + if err != nil { + return err + } + } +} + +func execStatus(content []byte) error { + var status struct { + Status, Message string + Details struct { + Causes []struct{ Reason, Message string } + } + } + if err := json.Unmarshal(content, &status); err != nil { + return fmt.Errorf("decode exec status: %w", err) + } + if status.Status == "Success" { + return nil + } + for _, cause := range status.Details.Causes { + if code, err := strconv.Atoi(cause.Message); cause.Reason == "ExitCode" && err == nil { + return ExitCodeError(code) + } + } + return fmt.Errorf("exec failed: %s", status.Message) +} diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go new file mode 100644 index 00000000..5ff668f3 --- /dev/null +++ b/act/container/kubernetes_test.go @@ -0,0 +1,290 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "archive/tar" + "bytes" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/websocket" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type podManifest struct { + Metadata struct{ Labels map[string]string } + Spec struct { + AutomountServiceAccountToken bool + EnableServiceLinks bool + HostAliases []struct{ Hostnames []string } + Volumes []map[string]any + Containers []struct { + Name, ImagePullPolicy string + Resources map[string]any + StartupProbe *struct{ PeriodSeconds, TimeoutSeconds, FailureThreshold int } + } + } +} + +type fakeCluster struct { + mu sync.Mutex + requests map[string][]string +} + +func (f *fakeCluster) record(request, content string) { + f.mu.Lock() + defer f.mu.Unlock() + f.requests[request] = append(f.requests[request], content) +} + +func fakeKubernetes(t *testing.T, podStatus string) (string, *fakeCluster) { + cluster := &fakeCluster{requests: map[string][]string{}} + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("Authorization") != "Bearer test-token" { + w.WriteHeader(http.StatusUnauthorized) + return + } + body, err := io.ReadAll(r.Body) + assert.NoError(t, err) + cluster.record(r.Method+" "+r.URL.Path, string(body)) + pods, secrets := "/api/v1/namespaces/ci/pods", "/api/v1/namespaces/ci/secrets" + switch { + case r.Method == http.MethodGet && r.URL.Path == pods: + cluster.record("list", r.URL.Query().Get("labelSelector")) + _, _ = io.WriteString(w, `{"items":[{"metadata":{"name":"old","creationTimestamp":"2020-01-01T00:00:00Z"}},{"metadata":{"name":"new","creationTimestamp":"2999-01-01T00:00:00Z"}}]}`) + case strings.HasSuffix(r.URL.Path, "/exec"): + cluster.record("exec", r.URL.RawQuery) + serveLocalExec(t, w, r) + case strings.HasSuffix(r.URL.Path, "/log"): + _, _ = io.WriteString(w, "log of "+r.URL.Query().Get("container")) + case r.Method == http.MethodGet && strings.HasPrefix(r.URL.Path, pods+"/"): + _, _ = io.WriteString(w, podStatus) + case r.Method == http.MethodPost && (r.URL.Path == pods || r.URL.Path == secrets), r.Method == http.MethodDelete: + default: + w.WriteHeader(http.StatusNotFound) + } + })) + t.Cleanup(server.Close) + kubeconfig := filepath.Join(t.TempDir(), "kubeconfig") + require.NoError(t, os.WriteFile(filepath.Join(filepath.Dir(kubeconfig), "token"), []byte("test-token\n"), 0o600)) + require.NoError(t, os.WriteFile(kubeconfig, fmt.Appendf(nil, ` +current-context: test +contexts: [{name: test, context: {cluster: test, user: test, namespace: ci}}] +clusters: [{name: test, cluster: {server: "http://kubernetes.invalid", proxy-url: %q}}] +users: [{name: test, user: {tokenFile: token}}] +`, server.URL), 0o600)) + return kubeconfig, cluster +} + +func serveLocalExec(t *testing.T, w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{execProtocol}}) + if err != nil { + t.Error(err) + return + } + defer func() { _ = conn.CloseNow() }() + conn.SetReadLimit(-1) + command := r.URL.Query()["command"] + var stdin bytes.Buffer + for r.URL.Query().Get("stdin") == "true" { + _, frame, err := conn.Read(r.Context()) + if err != nil { + t.Error(err) + return + } + if frame[0] == 255 { + break + } + stdin.Write(frame[1:]) + } + cmd := exec.Command(command[0], command[1:]...) + cmd.Env = append(os.Environ(), "COPYFILE_DISABLE=1") + var stdout, stderr bytes.Buffer + cmd.Stdin, cmd.Stdout, cmd.Stderr = &stdin, &stdout, &stderr + status := `{"status":"Success"}` + var exitErr *exec.ExitError + if err := cmd.Run(); errors.As(err, &exitErr) { + status = fmt.Sprintf(`{"status":"Failure","reason":"NonZeroExitCode","details":{"causes":[{"reason":"ExitCode","message":"%d"}]}}`, exitErr.ExitCode()) + } else { + assert.NoError(t, err) + } + for _, frame := range [][]byte{append([]byte{1}, stdout.Bytes()...), append([]byte{2}, stderr.Bytes()...), append([]byte{3}, status...)} { + _ = conn.Write(r.Context(), websocket.MessageBinary, frame) + } +} + +func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uses POSIX shell") + } + ctx := t.Context() + kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{},{}]},"status":{"phase":"Running","containerStatuses":[ + {"name":"job","ready":true,"state":{"running":{}}}, + {"name":"my-db","ready":true,"state":{"running":{}}}]}}`) + workdir, err := filepath.EvalSymlinks(t.TempDir()) + require.NoError(t, err) + options := KubernetesOptions{Kubeconfig: kubeconfig, RunnerUUID: "runner-uuid", PodTemplates: []map[string]any{ + {"spec": map[string]any{ + "automountServiceAccountToken": true, + "containers": []any{ + map[string]any{"name": "job", "resources": map[string]any{"limits": map[string]any{"cpu": "2"}}}, + map[string]any{ + "name": "$services", "imagePullPolicy": "Always", "resources": map[string]any{"limits": map[string]any{"memory": "256Mi"}}, + "env": []any{map[string]any{"name": "POSTGRES_PASSWORD", "value": "template"}}, + }, + }, + }}, + {"spec": map[string]any{ + "volumes": []any{map[string]any{"name": "workspace", "emptyDir": map[string]any{"sizeLimit": "1Gi"}}}, + "containers": []any{map[string]any{"name": "job", "resources": map[string]any{"limits": map[string]any{"memory": "1Gi"}}}}, + }}, + }} + pod, err := NewKubernetesPod(options) + require.NoError(t, err) + var jobLog, serviceLog bytes.Buffer + job := pod.JobContainer(&NewContainerInput{Name: "Job-Name", Image: "node", WorkingDir: workdir, Stdout: &jobLog, Stderr: &jobLog}) + service := pod.ServiceContainer("My_DB", &NewContainerInput{ + Image: "postgres", + Cmd: []string{"postgres", "-D", "$(PGDATA)"}, + Env: []string{"POSTGRES_PASSWORD=service-secret"}, + WorkflowOptions: "--health-cmd pg_isready --health-interval 2s --health-start-period 4s", + Stdout: &serviceLog, + }) + + require.NoError(t, job.Create(nil, nil)(ctx)) + var manifest podManifest + require.NoError(t, json.Unmarshal([]byte(cluster.requests["POST /api/v1/namespaces/ci/pods"][0]), &manifest)) + assert.Equal(t, "gitea-runner", manifest.Metadata.Labels["app.kubernetes.io/managed-by"]) + assert.Equal(t, "runner-uuid", manifest.Metadata.Labels[runnerUUIDLabel]) + assert.True(t, manifest.Spec.AutomountServiceAccountToken) + assert.False(t, manifest.Spec.EnableServiceLinks) + assert.Equal(t, []string{"my-db"}, manifest.Spec.HostAliases[0].Hostnames) + require.Len(t, manifest.Spec.Containers, 2) + assert.Equal(t, map[string]any{"limits": map[string]any{"cpu": "2", "memory": "1Gi"}}, manifest.Spec.Containers[0].Resources) + assert.Empty(t, manifest.Spec.Containers[0].ImagePullPolicy) + assert.Equal(t, "my-db", manifest.Spec.Containers[1].Name) + assert.Equal(t, "Always", manifest.Spec.Containers[1].ImagePullPolicy) + assert.Equal(t, map[string]any{"limits": map[string]any{"memory": "256Mi"}}, manifest.Spec.Containers[1].Resources) + assert.Equal(t, []map[string]any{ + {"name": "workspace", "emptyDir": map[string]any{"sizeLimit": "1Gi"}}, + {"name": "act", "emptyDir": map[string]any{}}, + }, manifest.Spec.Volumes) + assert.Equal(t, struct{ PeriodSeconds, TimeoutSeconds, FailureThreshold int }{2, 30, 5}, *manifest.Spec.Containers[1].StartupProbe) + assert.NotContains(t, cluster.requests["POST /api/v1/namespaces/ci/pods"][0], "service-secret") + assert.Contains(t, cluster.requests["POST /api/v1/namespaces/ci/secrets"][0], "service-secret") + assert.Contains(t, cluster.requests["POST /api/v1/namespaces/ci/secrets"][0], `"com.gitea.runner.uuid":"runner-uuid"`) + assert.Contains(t, cluster.requests["POST /api/v1/namespaces/ci/pods"][0], `{"name":"POSTGRES_PASSWORD","valueFrom":{"secretKeyRef":{"key":"my-db.0","name":"`+pod.secretName+`"}}}`) + assert.Equal(t, []any{"new"}, mergeManifest([]any{"old"}, []any{"new"}, "")) + mount := func(path string) any { return map[string]any{"name": "config", "mountPath": path} } + assert.Equal(t, []any{mount("/a"), mount("/b")}, mergeManifest([]any{mount("/a")}, []any{mount("/b")}, "volumeMounts")) + tmpfs := map[string]any{"name": "workspace", "emptyDir": map[string]any{"medium": "Memory"}} + ephemeral := map[string]any{"name": "workspace", "ephemeral": map[string]any{}} + assert.Equal(t, []any{ephemeral}, mergeManifest([]any{tmpfs}, []any{ephemeral}, "volumes")) + assert.Contains(t, cluster.requests["POST /api/v1/namespaces/ci/pods"][0], `"args":["postgres","-D","$$(PGDATA)"]`) + + require.NoError(t, os.WriteFile(filepath.Join(workdir, "env"), []byte("#!/bin/sh\necho \"$@\" > \"$0-args\"\nexec /usr/bin/env \"$@\"\n"), 0o755)) + t.Setenv("PATH", workdir+":"+os.Getenv("PATH")) + require.NoError(t, job.Start(false)(ctx)) + + stepEnv := map[string]string{"INPUT_WHO-TO-GREET": "it's\nme", "SECRET": "hunter2", "UID": "5"} + require.NoError(t, job.Exec([]string{"printenv", "SECRET"}, stepEnv, "", "")(ctx)) + envArgs, err := os.ReadFile(filepath.Join(workdir, "env-args")) + require.NoError(t, err) + assert.NotContains(t, string(envArgs), "hunter2") + require.NoError(t, job.Exec([]string{"printenv", "INPUT_WHO-TO-GREET"}, stepEnv, "", "")(ctx)) + require.NoError(t, job.Exec([]string{"printenv", "UID"}, stepEnv, "", "")(ctx)) + require.NoError(t, job.Exec([]string{"pwd", "-P"}, nil, "", workdir)(ctx)) + assert.Equal(t, "hunter2\nit's\nme\n5\n"+workdir+"\n", jobLog.String()) + assert.Equal(t, ExitCodeError(3), job.Exec([]string{"sh", "-c", "echo failing >&2; exit 3"}, nil, "", "")(ctx)) + assert.Contains(t, jobLog.String(), "failing\n") + for _, query := range cluster.requests["exec"] { + assert.NotContains(t, query, "hunter2") + } + + require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "workflow/event.json", Mode: 0o644, Body: `{"ref":"main"}`})(ctx)) + archive, err := job.GetContainerArchive(ctx, workdir+"/act/workflow/event.json") + require.NoError(t, err) + reader := tar.NewReader(archive) + header, err := reader.Next() + require.NoError(t, err) + content, err := io.ReadAll(reader) + require.NoError(t, err) + assert.Equal(t, "event.json", header.Name) + assert.JSONEq(t, `{"ref":"main"}`, string(content)) + require.NoError(t, archive.Close()) + _, err = job.GetContainerArchive(ctx, workdir+"/missing") + require.Error(t, err) + + pathEnv := map[string]string{} + require.NoError(t, job.UpdateFromImageEnv(&pathEnv)(ctx)) + assert.Equal(t, os.Getenv("PATH"), pathEnv["PATH"]) + + info, err := service.Inspect(ctx) + require.NoError(t, err) + assert.Equal(t, HealthHealthy, info.Health) + require.NoError(t, service.DumpLogs(ctx)) + assert.Equal(t, "log of my-db", serviceLog.String()) + + require.NoError(t, job.Remove()(ctx)) + assert.Len(t, cluster.requests["DELETE /api/v1/namespaces/ci/pods/"+pod.name], 1) + assert.Len(t, cluster.requests["DELETE /api/v1/namespaces/ci/secrets/"+pod.secretName], 1) + + require.NoError(t, RemoveOrphanKubernetesResources(ctx, options, time.Now())) + assert.Equal(t, []string{runnerUUIDLabel + "=runner-uuid"}, cluster.requests["list"]) + assert.Len(t, cluster.requests["DELETE /api/v1/namespaces/ci/pods/old"], 1) + assert.Len(t, cluster.requests["DELETE /api/v1/namespaces/ci/secrets/old"], 1) + assert.Empty(t, cluster.requests["DELETE /api/v1/namespaces/ci/pods/new"]) + assert.Empty(t, cluster.requests["DELETE /api/v1/namespaces/ci/secrets/new"]) +} + +func TestKubernetesPodFailsBeforeSteps(t *testing.T) { + kubeconfig, _ := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"containerStatuses":[ + {"name":"job","state":{"waiting":{"reason":"ErrImagePull","message":"no such image"}}}]}}`) + template := map[string]any{"spec": map[string]any{"containers": []any{map[string]any{"name": "sidecar"}}}} + for _, test := range []struct { + services []string + options, err string + }{ + {[]string{"db"}, "--hostname db", "only --health-* options are supported"}, + {[]string{"sidecar"}, "", "named like another container"}, + {[]string{"my_db", "My-DB"}, "", `service My-DB maps to container "my-db"`}, + } { + pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig, Namespace: "ci", PodTemplates: []map[string]any{template}}) + require.NoError(t, err) + job := pod.JobContainer(&NewContainerInput{Image: "node"}) + for _, service := range test.services { + pod.ServiceContainer(service, &NewContainerInput{Image: "postgres", WorkflowOptions: test.options}) + } + require.ErrorContains(t, job.Create(nil, nil)(t.Context()), test.err) + } + pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig, Namespace: "ci"}) + require.NoError(t, err) + job := pod.JobContainer(&NewContainerInput{Image: "missing"}) + require.NoError(t, job.Create(nil, nil)(t.Context())) + require.ErrorContains(t, job.Start(false)(t.Context()), "ErrImagePull: no such image") + + plugin := filepath.Join(t.TempDir(), "kubeconfig") + require.NoError(t, os.WriteFile(plugin, []byte(` +current-context: eks +contexts: [{name: eks, context: {cluster: eks, user: eks}}] +clusters: [{name: eks, cluster: {server: "https://eks.example"}}] +users: [{name: eks, user: {exec: {command: aws}}}] +`), 0o600)) + _, err = NewKubernetesPod(KubernetesOptions{Kubeconfig: plugin}) + require.ErrorContains(t, err, "credential plugin") +} diff --git a/act/container/tar.go b/act/container/tar.go new file mode 100644 index 00000000..43a355bd --- /dev/null +++ b/act/container/tar.go @@ -0,0 +1,69 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// Copyright 2020 The nektos/act Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "archive/tar" + "context" + "io" + "path/filepath" + "strings" + + "gitea.com/gitea/runner/act/common" + "gitea.com/gitea/runner/act/filecollector" + + "github.com/go-git/go-billy/v5/helper/polyfill" + "github.com/go-git/go-billy/v5/osfs" + "github.com/go-git/go-git/v5/plumbing/format/gitignore" +) + +func writeFilesTar(ctx context.Context, w io.Writer, uid, gid int, files ...*FileEntry) error { + tw := tar.NewWriter(w) + for _, file := range files { + common.Logger(ctx).Debugf("Writing entry to tarball %s len:%d", file.Name, len(file.Body)) + if err := tw.WriteHeader(&tar.Header{ + Name: file.Name, + Mode: file.Mode, + Size: int64(len(file.Body)), + Uid: uid, + Gid: gid, + }); err != nil { + return err + } + if _, err := tw.Write([]byte(file.Body)); err != nil { + return err + } + } + return tw.Close() +} + +// writeDirTar prefixes entries with dstPath for extraction at root. +func writeDirTar(ctx context.Context, w io.Writer, dstPath, srcPath string, useGitIgnore, skipGitDir bool, uid, gid int) error { + logger := common.Logger(ctx) + tw := tar.NewWriter(w) + srcPrefix := strings.TrimSuffix(filepath.Dir(srcPath), string(filepath.Separator)) + string(filepath.Separator) + logger.Debugf("Stripping prefix:%s src:%s", srcPrefix, srcPath) + + var ignorer gitignore.Matcher + if useGitIgnore { + ps, err := gitignore.ReadPatterns(polyfill.New(osfs.New(srcPath)), nil) + if err != nil { + logger.Debugf("Error loading .gitignore: %v", err) + } + + ignorer = gitignore.NewMatcher(ps) + } + + if err := filepath.Walk(srcPath, (&filecollector.FileCollector{ + Ignorer: ignorer, + SrcPath: srcPath, + SrcPrefix: srcPrefix, + SkipGitDir: skipGitDir, + Handler: &filecollector.TarCollector{TarWriter: tw, UID: uid, GID: gid, DstDir: dstPath[1:]}, + }).CollectFiles(ctx, []string{})); err != nil { + return err + } + return tw.Close() +} diff --git a/act/runner/action.go b/act/runner/action.go index 6929d18c..683dec7c 100644 --- a/act/runner/action.go +++ b/act/runner/action.go @@ -292,6 +292,9 @@ func execAsDocker(ctx context.Context, step actionStep, actionName, actionDir, b logger := common.Logger(ctx) rc := step.getRunContext() action := step.getActionModel() + if rc.kubernetes { + return errKubernetesDocker + } var prepImage common.Executor var image string diff --git a/act/runner/action_composite.go b/act/runner/action_composite.go index 609325db..299dbf48 100644 --- a/act/runner/action_composite.go +++ b/act/runner/action_composite.go @@ -104,6 +104,7 @@ func newCompositeRunContext(ctx context.Context, parent *RunContext, step action Parent: parent, EventJSON: parent.EventJSON, platformImage: parent.platformImage, + kubernetes: parent.kubernetes, jobIndex: parent.jobIndex, jobTotal: parent.jobTotal, } diff --git a/act/runner/run_context.go b/act/runner/run_context.go index f2e1a7c3..9f2c1797 100644 --- a/act/runner/run_context.go +++ b/act/runner/run_context.go @@ -7,6 +7,7 @@ package runner import ( "archive/tar" "bufio" + "cmp" "context" "crypto/rand" "crypto/sha256" @@ -22,6 +23,7 @@ import ( "regexp" "runtime" "slices" + "strconv" "strings" "sync" "time" @@ -64,6 +66,7 @@ type RunContext struct { ExprEval *expressionEvaluator JobContainer container.ExecutionsEnvironment serviceContainers []*serviceContainer + kubernetes bool containerSpec model.ContainerSpec // container:, resolved once with platformImage containerEnv yaml.Node // container.env of the kept containerSpec, decoded once the job env resolves JobName string @@ -417,11 +420,7 @@ func (rc *RunContext) startHostEnvironment() common.Executor { AllocatePTY: rc.Config.AllocatePTY, } rc.cleanUpJobContainer = rc.JobContainer.Remove() - for k, v := range rc.getRunnerContext(ctx) { - if v, ok := v.(string); ok { - rc.Env["RUNNER_"+strings.ToUpper(k)] = v - } - } + rc.setRunnerContextEnv(ctx) for _, env := range os.Environ() { if k, v, ok := strings.Cut(env, "="); ok { // don't override @@ -431,16 +430,70 @@ func (rc *RunContext) startHostEnvironment() common.Executor { } } + return rc.copyWorkflowFiles()(ctx) + } +} + +func (rc *RunContext) setRunnerContextEnv(ctx context.Context) { + for k, v := range rc.getRunnerContext(ctx) { + if v, ok := v.(string); ok { + rc.Env["RUNNER_"+strings.ToUpper(k)] = v + } + } +} + +var errKubernetesDocker = errors.New("docker actions and docker:// steps are not supported on Kubernetes") + +func (rc *RunContext) startKubernetesPod() common.Executor { + return func(ctx context.Context) error { + if rc.containerSpec.Options != "" || len(rc.containerSpec.Volumes) > 0 || rc.containerSpec.Credentials != nil { + return errors.New("container options, volumes and credentials are not supported on Kubernetes") + } + var options container.KubernetesOptions + if rc.Config.KubernetesPicker != nil { + options = rc.Config.KubernetesPicker(rc.runsOnPlatformNames(ctx)) + } + options.MaxLifetime, options.ForcePull = rc.Config.ContainerMaxLifetime, rc.Config.ForcePull + pod, err := container.NewKubernetesPod(options) + if err != nil { + return err + } + services, err := rc.jobServices(ctx, "") + if err != nil { + return err + } + for _, serviceID := range slices.Sorted(maps0.Keys(services)) { + input := services[serviceID] + if input.Username != "" || len(input.Binds) > 0 || len(input.Mounts) > 0 { + return fmt.Errorf("service %s: volumes and credentials are not supported on Kubernetes", serviceID) + } + rc.serviceContainers = append(rc.serviceContainers, &serviceContainer{name: serviceID, image: input.Image, container: pod.ServiceContainer(serviceID, input)}) + } + + logWriter := rc.commandLogWriter(ctx) + rc.JobContainer = pod.JobContainer(&container.NewContainerInput{ + Entrypoint: []string{"tail", "-f", "/dev/null"}, + WorkingDir: (&container.LinuxContainerEnvironmentExtensions{}).ToContainerPath(rc.Config.Workdir), + Image: rc.platformImage, + Name: rc.jobContainerName(), + Env: []string{"LANG=C.UTF-8"}, + Stdout: logWriter, + Stderr: logWriter, + }) + rc.cleanUpJobContainer = func(ctx context.Context) error { + rc.printServiceLogs(ctx) + return rc.JobContainer.Remove()(ctx) + } + defer printStartJobContainerGroup(ctx, rc.platformImage, rc.jobContainerName(), "")() return common.NewPipelineExecutor( - rc.JobContainer.Copy(rc.JobContainer.GetActPath()+"/", &container.FileEntry{ - Name: "workflow/event.json", - Mode: 0o644, - Body: rc.EventJSON, - }, &container.FileEntry{ - Name: "workflow/envs.txt", - Mode: 0o666, - Body: "", - }), + rc.JobContainer.Create(nil, nil), + rc.startServiceContainers(), + rc.JobContainer.Start(false), + func(ctx context.Context) error { rc.setRunnerContextEnv(ctx); return nil }, + rc.captureJobContainerInfo(), + rc.copyWorkflowFiles(), + rc.reportUnstartedServices(), + rc.waitForServiceContainers(), )(ctx) } } @@ -470,7 +523,6 @@ func WithJobVolumeCleanup(ctx context.Context, deferCleanup func(common.Executor 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) @@ -495,78 +547,13 @@ func (rc *RunContext) startJobContainer() common.Executor { networkName, createAndDeleteNetwork := rc.networkNameForGitea() rc.cleanUpJobContainer = rc.cleanupJobResources(networkName, createAndDeleteNetwork, false) - var services map[string]*model.ContainerSpec - if err := model.DecodeEvaluated("job services", rc.Run.Job().RawServices, rc.ExprEval.shared(ctx).EvaluateYamlNode, &services); err != nil { + services, err := rc.jobServices(ctx, networkName) + if err != nil { return err } for _, serviceID := range slices.Sorted(maps0.Keys(services)) { - spec := services[serviceID] - if spec == nil { - return fmt.Errorf("service %s has no container definition", serviceID) - } - spec.Image = strings.TrimPrefix(spec.Image, "docker://") - // GitHub compatibility: skip services whose image evaluates to an - // empty string, enabling conditional services via expressions - if spec.Image == "" { - logger.Infof("The service '%s' will not be started because the container definition has an empty image.", serviceID) - continue - } - envs := make([]string, 0, len(spec.Env)+len(rc.Config.ProxyEnv)) - // a service reaches the internet the way the job does; its own env still wins - for k, v := range mergeMaps(rc.Config.ProxyEnv, spec.Env) { - envs = append(envs, fmt.Sprintf("%s=%s", k, v)) - } - // keep these local: reusing username/password would overwrite the - // credentials the job container is pulled with further down - serviceUsername, servicePassword, err := registryCredentials(spec.Credentials, "") - if err != nil { - return fmt.Errorf("failed to handle service %s credentials: %w", serviceID, err) - } - cmd := spec.Cmd - if len(cmd) == 0 && spec.Command != "" { - if cmd, err = shellquote.Split(spec.Command); err != nil { - return fmt.Errorf("failed to parse service %s command: %w", serviceID, err) - } - } - var entrypoint []string - if spec.Entrypoint != "" { - entrypoint = []string{spec.Entrypoint} - } - - serviceBinds, serviceMounts, _ := splitVolumes(spec.Volumes) - - exposedPorts, portBindings, err := nat.ParsePortSpecs(spec.Ports) - if err != nil { - return fmt.Errorf("failed to parse service %s ports: %w", serviceID, err) - } - - serviceContainerName := createContainerName(rc.jobContainerName(), serviceID) - serviceLog := common.NewLineWriter(rawLogLine(ctx)) // raw, so a service's output cannot set the job's env, outputs or state - c := newContainer(&container.NewContainerInput{ - Name: serviceContainerName, - Image: spec.Image, - Username: serviceUsername, - Password: servicePassword, - Entrypoint: entrypoint, - Cmd: cmd, - Env: envs, - Mounts: serviceMounts, - Binds: serviceBinds, - Stdout: serviceLog, - Stderr: serviceLog, - Privileged: rc.Config.Privileged, - UsernsMode: rc.Config.UsernsMode, - Platform: rc.Config.ContainerArchitecture, - AutoRemove: false, // so a dead service's log survives, cleanupJobResources removes it - WorkflowOptions: spec.Options, - NetworkMode: networkName, - NetworkAliases: []string{serviceID}, - ExposedPorts: exposedPorts, - PortBindings: portBindings, - ValidVolumes: rc.Config.ValidVolumes, // not validVolumes(), a service gets no docker socket - AllocatePTY: rc.Config.AllocatePTY, - }) - rc.serviceContainers = append(rc.serviceContainers, &serviceContainer{name: serviceID, image: spec.Image, container: c}) + input := services[serviceID] + rc.serviceContainers = append(rc.serviceContainers, &serviceContainer{name: serviceID, image: input.Image, container: newContainer(input)}) } // For Gitea, `jobContainerNetwork` should be the same as `networkName` @@ -575,7 +562,7 @@ func (rc *RunContext) startJobContainer() common.Executor { ext := container.LinuxContainerEnvironmentExtensions{} containerInput := &container.NewContainerInput{ Cmd: nil, - Entrypoint: []string{"sleep", fmt.Sprint(rc.Config.ContainerMaxLifetime.Round(time.Second).Seconds())}, + Entrypoint: []string{"sleep", strconv.FormatInt(int64(rc.Config.ContainerMaxLifetime.Round(time.Second).Seconds()), 10)}, WorkingDir: ext.ToContainerPath(rc.Config.Workdir), Image: image, Username: username, @@ -623,20 +610,86 @@ func (rc *RunContext) startJobContainer() common.Executor { 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, - Body: rc.EventJSON, - }, &container.FileEntry{ - Name: "workflow/envs.txt", - Mode: 0o666, - Body: "", - }), + rc.copyWorkflowFiles(), rc.waitForServiceContainers(), )(ctx) } } +func (rc *RunContext) jobServices(ctx context.Context, networkName string) (map[string]*container.NewContainerInput, error) { + var services map[string]*model.ContainerSpec + if err := model.DecodeEvaluated("job services", rc.Run.Job().RawServices, rc.ExprEval.shared(ctx).EvaluateYamlNode, &services); err != nil { + return nil, err + } + inputs := map[string]*container.NewContainerInput{} + for _, serviceID := range slices.Sorted(maps0.Keys(services)) { + spec := services[serviceID] + if spec == nil { + return nil, fmt.Errorf("service %s has no container definition", serviceID) + } + spec.Image = strings.TrimPrefix(spec.Image, "docker://") + // GitHub compatibility: skip services whose image evaluates to an + // empty string, enabling conditional services via expressions + if spec.Image == "" { + common.Logger(ctx).Infof("The service '%s' will not be started because the container definition has an empty image.", serviceID) + continue + } + envs := make([]string, 0, len(spec.Env)+len(rc.Config.ProxyEnv)) + // a service reaches the internet the way the job does; its own env still wins + for k, v := range mergeMaps(rc.Config.ProxyEnv, spec.Env) { + envs = append(envs, fmt.Sprintf("%s=%s", k, v)) + } + serviceUsername, servicePassword, err := registryCredentials(spec.Credentials, "") + if err != nil { + return nil, fmt.Errorf("failed to handle service %s credentials: %w", serviceID, err) + } + cmd := spec.Cmd + if len(cmd) == 0 && spec.Command != "" { + if cmd, err = shellquote.Split(spec.Command); err != nil { + return nil, fmt.Errorf("failed to parse service %s command: %w", serviceID, err) + } + } + var entrypoint []string + if spec.Entrypoint != "" { + entrypoint = []string{spec.Entrypoint} + } + + serviceBinds, serviceMounts, _ := splitVolumes(spec.Volumes) + + exposedPorts, portBindings, err := nat.ParsePortSpecs(spec.Ports) + if err != nil { + return nil, fmt.Errorf("failed to parse service %s ports: %w", serviceID, err) + } + + serviceLog := common.NewLineWriter(rawLogLine(ctx)) // raw, so a service's output cannot set the job's env, outputs or state + inputs[serviceID] = &container.NewContainerInput{ + Name: createContainerName(rc.jobContainerName(), serviceID), + Image: spec.Image, + Username: serviceUsername, + Password: servicePassword, + Entrypoint: entrypoint, + Cmd: cmd, + Env: envs, + Mounts: serviceMounts, + Binds: serviceBinds, + Stdout: serviceLog, + Stderr: serviceLog, + Privileged: rc.Config.Privileged, + UsernsMode: rc.Config.UsernsMode, + Platform: rc.Config.ContainerArchitecture, + AutoRemove: false, // so a dead service's log survives, cleanupJobResources removes it + WorkflowOptions: spec.Options, + NetworkMode: networkName, + NetworkAliases: []string{serviceID}, + ExposedPorts: exposedPorts, + PortBindings: portBindings, + ValidVolumes: rc.Config.ValidVolumes, // not validVolumes(), a service gets no docker socket + AllocatePTY: rc.Config.AllocatePTY, + } + } + return inputs, nil +} + func (rc *RunContext) commandLogWriter(ctx context.Context) io.Writer { return common.NewLineWriter(rc.commandHandler(ctx), rawLogLine(ctx)) } @@ -975,6 +1028,13 @@ func (svc *serviceContainer) healthOutputSuffix() string { return ": " + svc.info.HealthOutput } +func (rc *RunContext) copyWorkflowFiles() common.Executor { + return rc.JobContainer.Copy(rc.JobContainer.GetActPath()+"/", + &container.FileEntry{Name: "workflow/event.json", Mode: 0o644, Body: rc.EventJSON}, + &container.FileEntry{Name: "workflow/envs.txt", Mode: 0o666}, + ) +} + // captureJobContainerInfo is a convenience: failing to describe the container must not // fail the job. func (rc *RunContext) captureJobContainerInfo() common.Executor { @@ -1043,9 +1103,12 @@ func (rc *RunContext) interpolateOutputs() common.Executor { func (rc *RunContext) startContainer() common.Executor { return func(ctx context.Context) error { var err error - if rc.IsHostEnv() { + switch { + case rc.kubernetes: + err = rc.startKubernetesPod()(ctx) + case rc.IsHostEnv(): err = rc.startHostEnvironment()(ctx) - } else { + default: err = rc.startJobContainer()(ctx) } if err != nil { @@ -1154,7 +1217,7 @@ func (rc *RunContext) Executor() (common.Executor, error) { } func (rc *RunContext) runsOnImage(ctx context.Context) (string, error) { - if rc.Run.Job().RunsOn() == nil { + if rc.Run.Job().RunsOn() == nil && rc.containerSpec.Image == "" { common.Logger(ctx).Errorf("'runs-on' key not defined in %s", rc.String()) } @@ -1201,13 +1264,13 @@ func (rc *RunContext) resolvePlatformImage(ctx context.Context) error { if spec.Image != "" || imagelessContainer(withoutEnv) { rc.containerSpec, rc.containerEnv = spec, env } - if rc.containerSpec.Image != "" { - rc.platformImage = rc.containerSpec.Image - return nil - } image, err := rc.runsOnImage(ctx) - rc.platformImage = image - return err + if err != nil { + return err + } + image, rc.kubernetes = strings.CutPrefix(image, "kubernetes://") + rc.platformImage = cmp.Or(rc.containerSpec.Image, image) + return nil } // imagelessContainer reports a container: mapping without an image key, which configures the runs-on image. diff --git a/act/runner/run_context_test.go b/act/runner/run_context_test.go index 901715f9..12182170 100644 --- a/act/runner/run_context_test.go +++ b/act/runner/run_context_test.go @@ -1629,10 +1629,13 @@ func createRunsOnRunContext(t *testing.T, runsOn string) *RunContext { }) } -func TestRunContextContainerImageCannotSelectHost(t *testing.T) { +func TestRunContextContainerImageCannotSelectBackend(t *testing.T) { rc := createIfTestRunContext(map[string]*model.Job{"job1": createJob(t, "container: -self-hosted", "")}) require.NoError(t, rc.resolvePlatformImage(t.Context())) assert.False(t, rc.IsHostEnv()) + + rc = createIfTestRunContext(map[string]*model.Job{"job1": createJob(t, "runs-on: ${{ fromJSON('x') }}\ncontainer: node", "")}) + require.ErrorContains(t, rc.resolvePlatformImage(t.Context()), "runs-on") } func TestRunContextImageOS(t *testing.T) { diff --git a/act/runner/runner.go b/act/runner/runner.go index dc5662ee..fc97e3a9 100644 --- a/act/runner/runner.go +++ b/act/runner/runner.go @@ -75,6 +75,7 @@ type Config struct { // AppURL. It is never set for github.com or a GithubMirror, so the token stays on-instance. DefaultActionInstanceIsSelfHosted bool PlatformPicker func(labels []string) string + KubernetesPicker func(labels []string) container.KubernetesOptions JobLoggerLevel *log.Level // the level of job logger ValidVolumes []string // only volumes (and bind mounts) in this slice can be mounted on the job container or service containers SharedToolCache bool // one tool cache for all jobs instead of one per job diff --git a/act/runner/step_action_local.go b/act/runner/step_action_local.go index ec78772e..9f6f24ba 100644 --- a/act/runner/step_action_local.go +++ b/act/runner/step_action_local.go @@ -61,11 +61,14 @@ func (sal *stepActionLocal) main() common.Executor { treader := tar.NewReader(tars) header, err := treader.Next() if errors.Is(err, io.EOF) { + tars.Close() return nil, nil, os.ErrNotExist } else if err != nil { + tars.Close() return nil, nil, err } if header.FileInfo().Mode()&os.ModeSymlink == os.ModeSymlink { + tars.Close() spath, err = symlinkJoin(spath, header.Linkname, containerActionPath) if err != nil { return nil, nil, err diff --git a/act/runner/step_docker.go b/act/runner/step_docker.go index 2a764696..e43a4802 100644 --- a/act/runner/step_docker.go +++ b/act/runner/step_docker.go @@ -56,6 +56,9 @@ func (sd *stepDocker) runUsesContainer() common.Executor { step := sd.Step return func(ctx context.Context) error { + if rc.kubernetes { + return errKubernetesDocker + } image := strings.TrimPrefix(step.Uses, "docker://") cmd, err := shellquote.Split(step.With["args"]) if err != nil { diff --git a/docs/kubernetes.md b/docs/kubernetes.md new file mode 100644 index 00000000..5f7ad615 --- /dev/null +++ b/docs/kubernetes.md @@ -0,0 +1,58 @@ +# Jobs as Kubernetes pods + +A `kubernetes` label runs each job as a pod, without a Docker daemon: + +```text +ubuntu-latest:kubernetes://docker.gitea.com/runner-images:ubuntu-latest +``` + +The pod has a `job` container from the label's image or the job's `container.image`, plus one container per service. + +## Setup + +In a cluster, the runner uses its pod's service account and namespace. [`examples/kubernetes/kubernetes-pods.yaml`](../examples/kubernetes/kubernetes-pods.yaml) deploys a runner with these permissions: + +| Resource | Verbs | +| --- | --- | +| `pods` | `create`, `get`, `list`, `delete` | +| `pods/exec` | `create`, `get` | +| `pods/log` | `get` | +| `secrets` | `create`, `delete` | + +Outside a cluster, it uses the current context of the first file in `$KUBECONFIG`, or of `~/.kube/config`, which must authenticate with a token, token file or client certificate. The cluster must run Kubernetes 1.30 or later, and job images need `sh`, `tar`, `tail` and `env`. + +## Configuration + +```yaml +kubernetes: + kubeconfig: "" # instead of the service account + namespace: "" # defaults to the service account's, the kubeconfig context's, or default + pod_template: # merged under each job pod, a container named job merges into the job container, one named $services into each service container + spec: + containers: + - name: job + resources: + limits: + memory: 4Gi + - name: $services + resources: + limits: + memory: 1Gi + pod_templates: # merged over pod_template for jobs whose runs-on has the label they are keyed by + gpu: + spec: + nodeSelector: + nvidia.com/gpu.present: "true" +``` + +A job with `runs-on: [linux, gpu]` gets the `linux` and then the `gpu` entry of `pod_templates` merged over `pod_template`, while its image comes from the first of its labels the runner has. A template volume named `workspace` or `act` replaces the `emptyDir` the runner mounts at the workspace or at `/var/run/act`, for example to limit its size. Images are pulled by the cluster's `imagePullPolicy` default, a template's, or always with `container.force_pull`. + +Job pods mount no service account token unless the template sets `automountServiceAccountToken: true`. Job pods and their Secrets are labeled `app.kubernetes.io/managed-by: gitea-runner` and `com.gitea.runner.uuid: `. The idle cleanup removes those the runner did not get to remove, once they are older than `runner.workdir_cleanup_age`. + +## Services + +Services share the pod's network, so a step reaches one by its id or `localhost`, and two services cannot listen on the same port. An id is lowercased and has characters other than letters, digits and `-` replaced by `-`, so the service `my_db` is reached as `my-db`. Pass secrets to services through `env`, as `command` is visible in the pod spec. Of `options`, only `--health-*` is supported. They become a startup probe with Docker's defaults, which stops a service that fails its retries, so the job fails as on Docker. + +## Unsupported + +Docker actions, `docker://` steps, and the `options`, `volumes` and `credentials` of the job container fail the job, as do `volumes` and `credentials` of services. Run such workflows on a [Docker in Docker](../examples/kubernetes/README.md) runner. Pull private images with `imagePullSecrets` in `pod_template`. diff --git a/e2e/compatibility_test.go b/e2e/compatibility_test.go index c6517b98..2dee668e 100644 --- a/e2e/compatibility_test.go +++ b/e2e/compatibility_test.go @@ -31,6 +31,7 @@ func TestCompatibility(t *testing.T) { t.Run("cancellation_and_log_streaming", testRunCancellation) t.Run("dispatch", testWorkflowDispatch) t.Run("ephemeral", testEphemeralRunner) + t.Run("kubernetes", testKubernetes) } func testPayloads(t *testing.T) { diff --git a/e2e/gitea_fixture.go b/e2e/gitea_fixture.go index 4f3abd82..37447274 100644 --- a/e2e/gitea_fixture.go +++ b/e2e/gitea_fixture.go @@ -23,6 +23,7 @@ import ( "gitea.com/gitea/runner/act/container" + "github.com/moby/moby/api/pkg/stdcopy" apicontainer "github.com/moby/moby/api/types/container" "github.com/moby/moby/api/types/network" mobyclient "github.com/moby/moby/client" @@ -306,9 +307,14 @@ func randomToken(n int) string { } func (f *GiteaFixture) exec(ctx context.Context, cmd []string) (string, error) { - created, err := f.cli.ExecCreate(ctx, f.id, mobyclient.ExecCreateOptions{ + return containerExec(ctx, f.cli, f.id, "git", nil, cmd...) // gitea refuses root +} + +func containerExec(ctx context.Context, cli mobyclient.APIClient, id, user string, stdin io.Reader, cmd ...string) (string, error) { + created, err := cli.ExecCreate(ctx, id, mobyclient.ExecCreateOptions{ Cmd: cmd, - User: "git", // gitea refuses root + User: user, + AttachStdin: stdin != nil, AttachStdout: true, AttachStderr: true, }) @@ -316,18 +322,24 @@ func (f *GiteaFixture) exec(ctx context.Context, cmd []string) (string, error) { return "", err } - attached, err := f.cli.ExecAttach(ctx, created.ID, mobyclient.ExecAttachOptions{}) + attached, err := cli.ExecAttach(ctx, created.ID, mobyclient.ExecAttachOptions{}) if err != nil { return "", err } defer attached.Close() + if stdin != nil { + go func() { + _, _ = io.Copy(attached.Conn, stdin) + _ = attached.CloseWrite() + }() + } var out bytes.Buffer - if _, err := io.Copy(&out, attached.Reader); err != nil { + if _, err := stdcopy.StdCopy(&out, &out, attached.Reader); err != nil { return "", err } - inspected, err := f.cli.ExecInspect(ctx, created.ID, mobyclient.ExecInspectOptions{}) + inspected, err := cli.ExecInspect(ctx, created.ID, mobyclient.ExecInspectOptions{}) if err != nil { return "", err } diff --git a/e2e/kubernetes_test.go b/e2e/kubernetes_test.go new file mode 100644 index 00000000..0e447326 --- /dev/null +++ b/e2e/kubernetes_test.go @@ -0,0 +1,166 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +//go:build e2e + +package e2e + +import ( + "bytes" + "context" + "fmt" + "net" + "net/netip" + "net/url" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" + + "gitea.com/gitea/runner/internal/pkg/config" + + "github.com/moby/moby/api/pkg/stdcopy" + apicontainer "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/network" + mobyclient "github.com/moby/moby/client" +) + +func testKubernetes(t *testing.T) { + t.Parallel() + ctx := t.Context() + k3sID, kubernetes := startK3s(t) + + images := preloadImages(t, k3sID, os.Getenv("E2E_JOB_IMAGE"), os.Getenv("SERVICE_IMAGE")) + + api, repo := newScenario(t) + if err := api.CreateVariable(ctx, repo, "E2E_SERVICE_IMAGE", images[1]); err != nil { + t.Fatalf("create service image variable: %v", err) + } + startRunner(t, repo, "e2e-kubernetes", runnerOptions{image: images[0], kubernetes: kubernetes}) + pushWorkflow(t, api, repo, "kubernetes.yml") + + wfRun := waitForRun(t, api, repo) + requireSuccess(t, api, repo, wfRun.ID) + + if out, err := containerExec(ctx, fixture.cli, k3sID, "", nil, + "kubectl", "wait", "--for=delete", "pods,secrets", "--selector=app.kubernetes.io/managed-by=gitea-runner", "--timeout=1m", + ); err != nil { + t.Errorf("job pods or secrets left behind: %v: %s", err, out) + } +} + +func startK3s(t *testing.T) (string, *config.Kubernetes) { + t.Helper() + ctx := t.Context() + cli := fixture.cli + + name := fmt.Sprintf("gitea-runner-e2e-k3s-%d", time.Now().UnixNano()) + apiPort := network.MustParsePort("6443/tcp") + server := "https://" + name + ":6443" + hostConfig := &apicontainer.HostConfig{Privileged: true} + var netConfig *network.NetworkingConfig + if fixture.network != "" { + netConfig = &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{fixture.network: {}}} + } else { + loopback := netip.MustParseAddr("127.0.0.1") + port, err := freeHostPort(loopback.String()) + if err != nil { + t.Fatalf("find a free host port: %v", err) + } + server = "https://" + net.JoinHostPort(loopback.String(), strconv.Itoa(port)) + hostConfig.PortBindings = network.PortMap{apiPort: []network.PortBinding{{HostIP: loopback, HostPort: strconv.Itoa(port)}}} + } + + created, err := cli.ContainerCreate(ctx, mobyclient.ContainerCreateOptions{ + Config: &apicontainer.Config{ + Image: os.Getenv("E2E_K3S_IMAGE"), + Cmd: []string{"server", "--disable=traefik,metrics-server,servicelb,local-storage,coredns", "--tls-san=" + name, "--kubelet-arg=fail-cgroupv1=false"}, + ExposedPorts: network.PortSet{apiPort: struct{}{}}, + }, + HostConfig: hostConfig, + NetworkingConfig: netConfig, + Name: name, + }) + if err != nil { + t.Fatalf("create k3s container: %v", err) + } + t.Cleanup(func() { + _, _ = cli.ContainerRemove(context.Background(), created.ID, mobyclient.ContainerRemoveOptions{Force: true, RemoveVolumes: true}) + }) + if _, err := cli.ContainerStart(ctx, created.ID, mobyclient.ContainerStartOptions{}); err != nil { + t.Fatalf("start k3s container: %v", err) + } + + waitCtx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + for { + _, err := containerExec(waitCtx, cli, created.ID, "", nil, "kubectl", "get", "serviceaccount", "default") + if err == nil { + break + } + inspected, inspectErr := cli.ContainerInspect(ctx, created.ID, mobyclient.ContainerInspectOptions{}) + if inspectErr != nil || !inspected.Container.State.Running || waitCtx.Err() != nil { + var logs bytes.Buffer + if reader, logsErr := cli.ContainerLogs(ctx, created.ID, mobyclient.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: "30"}); logsErr == nil { + _, _ = stdcopy.StdCopy(&logs, &logs, reader) + _ = reader.Close() + } + t.Fatalf("k3s did not start: %v, state %+v %v, last log lines:\n%s", err, inspected.Container.State, inspectErr, logs.String()) + } + select { + case <-waitCtx.Done(): + case <-time.After(pollInterval): + } + } + + kubeconfig, err := containerExec(ctx, cli, created.ID, "", nil, "cat", "/etc/rancher/k3s/k3s.yaml") + if err != nil { + t.Fatalf("read kubeconfig: %v", err) + } + path := filepath.Join(t.TempDir(), "kubeconfig") + if err := os.WriteFile(path, []byte(strings.ReplaceAll(kubeconfig, "https://127.0.0.1:6443", server)), 0o600); err != nil { + t.Fatalf("write kubeconfig: %v", err) + } + kubernetes := &config.Kubernetes{Kubeconfig: path, Namespace: "default"} + + if fixture.network != "" { + gitea, err := cli.ContainerInspect(ctx, fixture.id, mobyclient.ContainerInspectOptions{}) + if err != nil { + t.Fatalf("inspect gitea container: %v", err) + } + giteaURL, err := url.Parse(fixture.baseURL) + if err != nil { + t.Fatalf("parse gitea url: %v", err) + } + kubernetes.PodTemplate = map[string]any{"spec": map[string]any{"hostAliases": []any{map[string]any{ + "ip": gitea.Container.NetworkSettings.Networks[fixture.network].IPAddress.String(), + "hostnames": []any{giteaURL.Hostname()}, + }}}} + } + return created.ID, kubernetes +} + +func preloadImages(t *testing.T, k3sID string, images ...string) []string { + t.Helper() + ctx := t.Context() + var aliases []string + for index, image := range images { + alias := fmt.Sprintf("gitea-runner-e2e/preload-%d:%d", index, time.Now().UnixNano()) + if _, err := fixture.cli.ImageTag(ctx, mobyclient.ImageTagOptions{Source: image, Target: alias}); err != nil { + t.Fatalf("tag %s: %v", image, err) + } + t.Cleanup(func() { _, _ = fixture.cli.ImageRemove(context.Background(), alias, mobyclient.ImageRemoveOptions{}) }) + aliases = append(aliases, alias) + } + archive, err := fixture.cli.ImageSave(ctx, aliases) + if err != nil { + t.Fatalf("save %v: %v", images, err) + } + defer archive.Close() + if out, err := containerExec(ctx, fixture.cli, k3sID, "", archive, "ctr", "--namespace=k8s.io", "images", "import", "-"); err != nil { + t.Fatalf("import %v into k3s: %v: %s", images, err, out) + } + return aliases +} diff --git a/e2e/runner_harness.go b/e2e/runner_harness.go index 2c9f8807..a4f7a97f 100644 --- a/e2e/runner_harness.go +++ b/e2e/runner_harness.go @@ -6,6 +6,7 @@ package e2e import ( + "cmp" "context" "errors" "os" @@ -24,10 +25,12 @@ import ( ) type runnerOptions struct { - capacity int - ephemeral bool - cacheV2 *bool - cacheHost string + capacity int + ephemeral bool + cacheV2 *bool + cacheHost string + image string + kubernetes *config.Kubernetes } func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *poll.Poller { @@ -61,7 +64,11 @@ func startRunner(t *testing.T, repo, labelName string, options runnerOptions) *p cfg.Container.Network = fixture.network } - rawLabel := labelName + ":docker://" + os.Getenv("E2E_JOB_IMAGE") + scheme := labels.SchemeDocker + if options.kubernetes != nil { + scheme, cfg.Kubernetes = labels.SchemeKubernetes, *options.kubernetes + } + rawLabel := labelName + ":" + scheme + "://" + cmp.Or(options.image, os.Getenv("E2E_JOB_IMAGE")) label, err := labels.Parse(rawLabel) if err != nil { t.Fatalf("parse label %q: %v", rawLabel, err) diff --git a/e2e/testdata/workflows/kubernetes.yml b/e2e/testdata/workflows/kubernetes.yml new file mode 100644 index 00000000..5b8a3012 --- /dev/null +++ b/e2e/testdata/workflows/kubernetes.yml @@ -0,0 +1,20 @@ +name: kubernetes +on: push +jobs: + pod: + runs-on: e2e-kubernetes + services: + web: + image: ${{ vars.E2E_SERVICE_IMAGE }} + env: + NGINX_ENTRYPOINT_QUIET_LOGS: 1 + options: --health-cmd "wget -q -O /dev/null http://localhost/" --health-interval 1s + steps: + - uses: actions/checkout@v4 + - run: | + test -f .gitea/workflows/kubernetes.yml + echo "FROM_ENV=exported" >> "$GITHUB_ENV" + - run: | + test "$FROM_ENV" = exported + test -n "$KUBERNETES_SERVICE_HOST" -a -n "${{ runner.arch }}" + curl -fsS -o /dev/null http://web/ diff --git a/examples/kubernetes/README.md b/examples/kubernetes/README.md index d79055a6..5563a6d1 100644 --- a/examples/kubernetes/README.md +++ b/examples/kubernetes/README.md @@ -1,25 +1,8 @@ -## Kubernetes Docker in Docker Deployment with `gitea-runner` +## Kubernetes deployments of `gitea-runner` -NOTE: Docker in Docker (dind) requires elevated privileges on Kubernetes. The current way to achieve this is to set the pod `SecurityContext` to `privileged`. Keep in mind that this is a potential security issue that has the potential for a malicious application to break out of the container context. +Jobs can run on Kubernetes in two ways: -NOTE: `dind-docker.yaml` uses the native sidecar pattern (init container with `restartPolicy: Always`), which requires Kubernetes 1.29+ (or 1.28 with the `SidecarContainers` feature gate). +- **Job pods**: each job runs as its own pod, without privileged containers and under the cluster's quotas and policies. Docker actions and `docker://` steps are not supported. Deploy with [`kubernetes-pods.yaml`](kubernetes-pods.yaml) and see [docs/kubernetes.md](../../docs/kubernetes.md). +- **Docker in Docker**: jobs run in a Docker daemon next to the runner. Every workflow feature works, but the daemon is privileged, so a job can break out to the node. Deploy with [`dind-docker.yaml`](dind-docker.yaml), [`statefulset-dind.yaml`](statefulset-dind.yaml) for several replicas, [`rootless-docker.yaml`](rootless-docker.yaml) for a rootless daemon, or the [Helm chart](https://gitea.com/gitea/helm-actions). The sidecars in the first two need Kubernetes 1.29 or later. -NOTE: A helm chart for `gitea-runner` also exists for easier deployments https://gitea.com/gitea/helm-actions - -Each example persists **two** things, and it is worth knowing which is which: - -- `/data` is the runner's working directory. It holds the `.runner` registration file and, optionally, the config file — so the runner re-attaches to the server instead of registering again. -- The Docker daemon's data root holds the images pulled for jobs (`/var/lib/docker` for the dind sidecar, `/home/rootless/.local/share/docker` for `dind-rootless`). It is *not* under `/data`. If you drop this volume, the examples still work, but the image cache is discarded whenever the pod is recreated and every job re-pulls its images. - -- Kubernetes SIGKILLs a pod 30s after SIGTERM by default, long before a job finishes and reports its result, which leaves tasks the server can only reap as zombies. The manifests raise `terminationGracePeriodSeconds` to three hours, matching the systemd example and the `runner.timeout` job ceiling; set `runner.shutdown_timeout` below that so the runner drains jobs within the window rather than being killed mid-cleanup. - -Files in this directory: - -- [`dind-docker.yaml`](dind-docker.yaml) - How to create a Deployment and Persistent Volume for Kubernetes to act as a runner. The Docker credentials are re-generated each time the pod connects and does not need to be persisted. - -- [`rootless-docker.yaml`](rootless-docker.yaml) - How to create a rootless Deployment and Persistent Volume for Kubernetes to act as a runner. The Docker credentials are re-generated each time the pod connects and does not need to be persisted. - -- [`statefulset-dind.yaml`](statefulset-dind.yaml) - StatefulSet variant of the dind example. Each replica gets a stable identity and its own persistent volume via `volumeClaimTemplates`, so the runner keeps its `.runner` registration across restarts and reschedules instead of trying to register again. +All examples read the registration token from the `runner-secret` Secret and the Gitea URL from `GITEA_INSTANCE_URL`, and keep the registration in `/data`. They give a stopping pod three hours to finish its jobs, so keep `runner.shutdown_timeout` below `terminationGracePeriodSeconds`. diff --git a/examples/kubernetes/kubernetes-pods.yaml b/examples/kubernetes/kubernetes-pods.yaml new file mode 100644 index 00000000..9b3c8aac --- /dev/null +++ b/examples/kubernetes/kubernetes-pods.yaml @@ -0,0 +1,83 @@ +apiVersion: v1 +data: + token: << base64 encoded registration token >> +kind: Secret +metadata: + name: runner-secret +--- +apiVersion: v1 +kind: ServiceAccount +metadata: + name: runner +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: runner +rules: +- apiGroups: [""] + resources: ["pods"] + verbs: ["create", "get", "list", "delete"] +- apiGroups: [""] + resources: ["pods/exec"] + verbs: ["create", "get"] +- apiGroups: [""] + resources: ["pods/log"] + verbs: ["get"] +- apiGroups: [""] + resources: ["secrets"] + verbs: ["create", "delete"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: runner +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: runner +subjects: +- kind: ServiceAccount + name: runner +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: runner +spec: + serviceName: runner + selector: + matchLabels: + app: runner + template: + metadata: + labels: + app: runner + spec: + serviceAccountName: runner + terminationGracePeriodSeconds: 10800 # keep above runner.shutdown_timeout, see README + containers: + - name: runner + image: gitea/runner:nightly + env: + - name: GITEA_INSTANCE_URL + value: http://gitea-http.gitea.svc.cluster.local:3000 + - name: GITEA_RUNNER_REGISTRATION_TOKEN + valueFrom: + secretKeyRef: + name: runner-secret + key: token + - name: GITEA_RUNNER_LABELS + value: ubuntu-latest:kubernetes://docker.gitea.com/runner-images:ubuntu-latest + volumeMounts: + - name: runner-data + mountPath: /data + volumeClaimTemplates: + - metadata: + name: runner-data + spec: + accessModes: + - ReadWriteOnce + resources: + requests: + storage: 1Gi diff --git a/go.mod b/go.mod index bdae6340..2b3f61b8 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( gitea.dev/actionslib v1.2.1 github.com/avast/retry-go/v5 v5.0.0 github.com/bmatcuk/doublestar/v4 v4.10.2 + github.com/coder/websocket v1.8.15 github.com/containerd/errdefs v1.0.0 github.com/creack/pty v1.1.24 github.com/distribution/reference v0.6.0 diff --git a/go.sum b/go.sum index 5c56be4c..14e46ba3 100644 --- a/go.sum +++ b/go.sum @@ -55,6 +55,8 @@ github.com/cevatbarisyilmaz/ara v0.0.4 h1:SGH10hXpBJhhTlObuZzTuFn1rrdmjQImITXnZV github.com/cevatbarisyilmaz/ara v0.0.4/go.mod h1:BfFOxnUd6Mj6xmcvRxHN3Sr21Z1T3U2MYkYOmoQe4Ts= github.com/cloudflare/circl v1.6.5 h1:O64F26HEqNhznd/hrC5KZXVKYuKM2rx4deZDTc4ihQA= github.com/cloudflare/circl v1.6.5/go.mod h1:h5LNyxAc5nTue9DS5jT+48en2PSDYt3zdGnz5OstK6c= +github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA= +github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg= github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI= github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M= github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE= diff --git a/internal/app/cmd/exec.go b/internal/app/cmd/exec.go index e2b43ee8..1c19c6d8 100644 --- a/internal/app/cmd/exec.go +++ b/internal/app/cmd/exec.go @@ -618,7 +618,7 @@ func loadExecCmd(ctx context.Context) *cobra.Command { execCmd.PersistentFlags().BoolVarP(&execArg.noSkipCheckout, "no-skip-checkout", "", false, "Do not skip actions/checkout") execCmd.PersistentFlags().BoolVarP(&execArg.debug, "debug", "d", false, "enable debug log") execCmd.PersistentFlags().BoolVarP(&execArg.dryrun, "dryrun", "n", false, "dryrun mode") - execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", config.DefaultImage, "Docker image to use. Use \"-self-hosted\" to run directly on the host.") + execCmd.PersistentFlags().StringVarP(&execArg.image, "image", "i", config.DefaultImage, "Docker image to use. Use \"kubernetes://\" to run it as a pod, \"-self-hosted\" to run directly on the host.") execCmd.PersistentFlags().StringVarP(&execArg.toolCacheMode, "tool-cache-mode", "", config.ToolCacheModeNone, "What to mount at RUNNER_TOOL_CACHE: none, or shared to reuse one tool cache across runs") execCmd.PersistentFlags().StringVarP(&execArg.network, "network", "", "", "Specify the network to which the container will connect") execCmd.PersistentFlags().StringVarP(&execArg.githubInstance, "gitea-instance", "", "", "Gitea instance to use.") diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index 9476c892..e9efc02d 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -153,6 +153,8 @@ var removeOrphanNetworks = container.RemoveOrphanNetworks var removeOrphanJobVolumes = container.RemoveOrphanJobVolumes +var removeOrphanKubernetesResources = container.RemoveOrphanKubernetesResources + // 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 @@ -175,6 +177,13 @@ func (r *Runner) OnIdle(ctx context.Context) { r.cleanupStaleDirs(ctx, hostRoot, isHostScratchDir) } r.cleanupOrphanDockerResources(ctx) + if r.uuid != "" && r.usesKubernetes() { + ctx, cancel := context.WithTimeout(ctx, time.Minute) + defer cancel() + if err := removeOrphanKubernetesResources(ctx, r.kubernetesOptions(nil), r.now().Add(-r.cfg.Runner.WorkdirCleanupAge)); err != nil { + log.Warnf("failed to clean up job pods left behind by earlier jobs: %v", err) + } + } } func (r *Runner) cleanupOrphanDockerResources(ctx context.Context) { @@ -402,10 +411,36 @@ func (r *Runner) requiresDocker() bool { return r.labels.RequireDocker() || r.cfg.Container.RequireDocker } +func (r *Runner) usesKubernetes() bool { + return slices.ContainsFunc(r.labels, func(label *labels.Label) bool { return label.Schema == labels.SchemeKubernetes }) +} + +// kubernetesOptions layers the pod templates of the runner labels a job runs on, in runs-on order, over the shared one. +func (r *Runner) kubernetesOptions(runsOn []string) container.KubernetesOptions { + options := container.KubernetesOptions{ + Kubeconfig: r.cfg.Kubernetes.Kubeconfig, + Namespace: r.cfg.Kubernetes.Namespace, + PodTemplates: []map[string]any{r.cfg.Kubernetes.PodTemplate}, + RunnerUUID: r.uuid, + } + names := r.labels.Names() + for index, name := range runsOn { + if template, ok := r.cfg.Kubernetes.PodTemplates[name]; ok && slices.Contains(names, name) && !slices.Contains(runsOn[:index], name) { + options.PodTemplates = append(options.PodTemplates, template) + } + } + return options +} + // fallbackPlatform is where a job runs whose runs-on matches no label, as any job without a // runs-on does, since Gitea sends those to every runner. func (r *Runner) fallbackPlatform(ctx context.Context) string { - if r.requiresDocker() || dockerReachable(ctx) { + switch { + case r.requiresDocker(): + return r.cfg.Runner.DefaultImage + case r.usesKubernetes(): + return labels.SchemeKubernetes + "://" + r.cfg.Runner.DefaultImage + case dockerReachable(ctx): return r.cfg.Runner.DefaultImage } return labels.SelfHostedPlatform @@ -449,7 +484,8 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. } return fallbackPlatform() } - cacheURL, cacheContainer := r.cacheForJob(job, platformPicker) + platform := platformPicker(job.RunsOn()) + cacheURL, cacheContainer := r.cacheForJob(job, platform) if cacheURL != "" { envs["ACTIONS_CACHE_URL"] = cacheURL + "/" } @@ -458,7 +494,13 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. // act reaches them by their workflow key. var services map[string]yaml.Node _ = job.RawServices.Decode(&services) // a whole-value expression names no service before evaluation - proxyEnv := JobProxyEnv(envs, cacheURL, slices.Sorted(maps.Keys(services))) + serviceNames := slices.Sorted(maps.Keys(services)) + if strings.HasPrefix(platform, labels.SchemeKubernetes+"://") { + for index, name := range serviceNames { + serviceNames[index] = container.KubernetesServiceName(name) + } + } + proxyEnv := JobProxyEnv(envs, cacheURL, serviceNames) maps.Copy(envs, proxyEnv) if r.capabilities != "" { @@ -585,6 +627,7 @@ func (r *Runner) run(ctx context.Context, task *runnerv1.Task, reporter *report. SharedToolCache: r.cfg.Runner.ToolCacheMode == config.ToolCacheModeShared, InsecureSkipTLS: r.cfg.Runner.Insecure, RunnerName: r.name, + KubernetesPicker: r.kubernetesOptions, } rr, err := runner.New(runnerConfig) @@ -867,10 +910,10 @@ func warnIgnoredCacheSecret(cfg *config.Config) { } // cacheForJob returns the cache address for the job, and the container to attach to its network to reach it there. -func (r *Runner) cacheForJob(job *model.Job, pickPlatform func([]string) string) (string, string) { +func (r *Runner) cacheForJob(job *model.Job, platform string) (string, string) { jobContainer := job.Container() cacheContainer := "" - if jobContainer != nil && jobContainer.Image != "" || pickPlatform(job.RunsOn()) != labels.SelfHostedPlatform { + if !strings.HasPrefix(platform, labels.SchemeKubernetes+"://") && (jobContainer != nil && jobContainer.Image != "" || platform != labels.SelfHostedPlatform) { cacheContainer = r.isolatedCacheContainer() } if cacheContainer == "" { diff --git a/internal/app/run/runner_idle_cleanup_test.go b/internal/app/run/runner_idle_cleanup_test.go index 2bb73d4a..0247ea34 100644 --- a/internal/app/run/runner_idle_cleanup_test.go +++ b/internal/app/run/runner_idle_cleanup_test.go @@ -11,7 +11,9 @@ import ( "testing" "time" + "gitea.com/gitea/runner/act/container" "gitea.com/gitea/runner/internal/pkg/config" + "gitea.com/gitea/runner/internal/pkg/labels" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -329,8 +331,20 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { return nil } t.Cleanup(func() { removeOrphanJobVolumes = origRemoveOrphanJobVolumes }) + var sweptPods []string + origRemoveOrphanKubernetesResources := removeOrphanKubernetesResources + removeOrphanKubernetesResources = func(ctx context.Context, options container.KubernetesOptions, createdBefore time.Time) error { + _, hasDeadline := ctx.Deadline() + assert.True(t, hasDeadline) + sweptPods = append(sweptPods, options.RunnerUUID) + assert.Equal(t, now.Add(-24*time.Hour), createdBefore) + return nil + } + t.Cleanup(func() { removeOrphanKubernetesResources = origRemoveOrphanKubernetesResources }) + kubernetesLabel, err := labels.Parse("k8s:kubernetes://node") + require.NoError(t, err) - r := &Runner{uuid: "runner-1", cfg: cfg, now: func() time.Time { return now }} + r := &Runner{uuid: "runner-1", cfg: cfg, labels: labels.Labels{kubernetesLabel}, now: func() time.Time { return now }} r.isolatedCacheContainer = func() string { return "a1b2c3d4e5f6" } r.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1"}, swept) @@ -353,4 +367,5 @@ func TestRunnerOnIdleRemovesOrphanNetworks(t *testing.T) { hostOnly.OnIdle(context.Background()) assert.Equal(t, []string{"runner-1", "runner-2"}, sweptVolumes) assert.Equal(t, []string{"a1b2c3d4e5f6", ""}, detached) + assert.Equal(t, []string{"runner-1"}, sweptPods) } diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go index 25c8220b..aacae44c 100644 --- a/internal/app/run/runner_test.go +++ b/internal/app/run/runner_test.go @@ -245,6 +245,16 @@ func TestRunnerFallbackPlatform(t *testing.T) { } } +func TestRunnerKubernetesOptionsLayerTheLabelPodTemplatesInRunsOnOrder(t *testing.T) { + cfg := &config.Config{Kubernetes: config.Kubernetes{ + PodTemplate: map[string]any{"base": true}, + PodTemplates: map[string]map[string]any{"gpu": {"gpu": true}, "large": {"large": true}, "other": {"other": true}}, + }} + r := &Runner{cfg: cfg, labels: labels.Labels{{Name: "linux"}, {Name: "gpu"}, {Name: "large"}}} + + require.Equal(t, []map[string]any{{"base": true}, {"large": true}, {"gpu": true}}, r.kubernetesOptions([]string{"linux", "large", "other", "gpu", "large"}).PodTemplates) +} + // Proxy variables are assembled per task, because a job's service containers have to be // reached directly and they are only known once the workflow is parsed. func TestNewRunnerLeavesProxyToTheTask(t *testing.T) { @@ -350,7 +360,7 @@ func TestNewRunnerCacheServiceV2(t *testing.T) { r.isolatedCacheContainer = func() string { return "a1b2c3d4e5f6" } pickPlatform := func(runsOn []string) string { return map[string]string{"native": labels.SelfHostedPlatform}[runsOn[0]] } for job, isolated := range map[string]bool{"native": false, "linux": true, "containerized": true, "empty": false} { - cacheURL, cacheContainer := r.cacheForJob(workflow.GetJob(job), pickPlatform) + cacheURL, cacheContainer := r.cacheForJob(workflow.GetJob(job), pickPlatform(workflow.GetJob(job).RunsOn())) if isolated { assert.Equal(t, "a1b2c3d4e5f6", cacheContainer, job) assert.Equal(t, strings.Replace(r.cacheHandler.ExternalURL(), "127.0.0.1", cacheContainer, 1), cacheURL, job) diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 0503504f..89f8b78a 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -53,12 +53,13 @@ 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 and volume cleanup below. + # the docker and Kubernetes 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 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. + # workspace contents until the host is rebuilt. On runners with kubernetes labels it + # removes such jobs' pods and Secrets. #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 @@ -106,8 +107,7 @@ runner: # terminal; tools like `docker build` emit redrawing progress frames into the captured log # when a TTY is present. #allocate_pty: false - # Image for a job whose runs-on matches none of the labels above. A runner without a - # docker daemon runs such a job on the host instead. + # Image for a job whose runs-on matches none of the labels above, see the Labels section of the README for where it runs. #default_image: "docker.gitea.com/runner-images:ubuntu-latest" # What to mount at RUNNER_TOOL_CACHE (/opt/hostedtoolcache), where setup actions install tools: # none: nothing. A docker job sees what its image ships there, a host job an empty dir, and @@ -280,6 +280,14 @@ host: # If it's empty, $HOME/.cache/act/ will be used. #workdir_parent: +# Jobs that run as Kubernetes pods, see docs/kubernetes.md. +kubernetes: + #kubeconfig: "" + #namespace: "" + #pod_template: {} + # Pod templates merged over pod_template, in runs-on order, for jobs whose runs-on has the label they are keyed by. + #pod_templates: {} + # Optional local task-admission checks. Disabled by default. When enabled, low # disk space or a failing script pauses new task fetching; existing jobs continue. # No health checks run while any job is active; the last result is reused until idle. diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index 35d04bf4..712522b8 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -63,8 +63,8 @@ type Runner struct { FetchTimeout time.Duration `yaml:"fetch_timeout"` // FetchTimeout specifies the timeout duration for fetching resources. FetchInterval time.Duration `yaml:"fetch_interval"` // FetchInterval specifies the interval duration for fetching resources. FetchIntervalMax time.Duration `yaml:"fetch_interval_max"` // FetchIntervalMax specifies the maximum backoff interval when idle. - WorkdirCleanupAge time.Duration `yaml:"workdir_cleanup_age"` // WorkdirCleanupAge removes stale bind-workdir task directories, 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. + WorkdirCleanupAge time.Duration `yaml:"workdir_cleanup_age"` // WorkdirCleanupAge removes stale bind-workdir task directories, orphaned host-mode scratch dirs and orphaned docker and Kubernetes job resources older than this duration during idle cleanup. + IdleCleanupInterval time.Duration `yaml:"idle_cleanup_interval"` // IdleCleanupInterval runs the idle cleanup (stale directories, orphaned docker networks and volumes, orphaned job pods) 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. @@ -194,6 +194,14 @@ type Host struct { WorkdirParent string `yaml:"workdir_parent"` // WorkdirParent specifies the parent directory for the host's working directory. } +// Kubernetes configures the pods jobs run in. +type Kubernetes struct { + Kubeconfig string `yaml:"kubeconfig"` // Kubeconfig to use instead of the in-cluster service account. + Namespace string `yaml:"namespace"` // Namespace of job pods, empty uses the service account's, else the kubeconfig context's, else default. + PodTemplate map[string]any `yaml:"pod_template"` // PodTemplate is a pod manifest the runner merges its own onto. + PodTemplates map[string]map[string]any `yaml:"pod_templates"` // PodTemplates are merged over PodTemplate, in runs-on order, for jobs whose runs-on has the label they are keyed by. +} + // Metrics represents the configuration for the Prometheus metrics endpoint. type Metrics struct { Enabled bool `yaml:"enabled"` // Enabled indicates whether the metrics endpoint is exposed. @@ -220,6 +228,7 @@ type Config struct { Host Host `yaml:"host"` // Host represents the configuration for the host. Metrics Metrics `yaml:"metrics"` // Metrics represents the configuration for the Prometheus metrics endpoint. HealthCheck HealthCheck `yaml:"health_check"` // HealthCheck controls opt-in local task-admission checks. + Kubernetes Kubernetes `yaml:"kubernetes"` } // LoadDefault returns the default configuration. diff --git a/internal/pkg/labels/labels.go b/internal/pkg/labels/labels.go index 66e3f8e2..a4910d9e 100644 --- a/internal/pkg/labels/labels.go +++ b/internal/pkg/labels/labels.go @@ -9,8 +9,9 @@ import ( ) const ( - SchemeHost = "host" - SchemeDocker = "docker" + SchemeHost = "host" + SchemeDocker = "docker" + SchemeKubernetes = "kubernetes" // SelfHostedPlatform is the platform marker act treats as "run on the host". SelfHostedPlatform = "-self-hosted" @@ -42,8 +43,9 @@ func Parse(str string) (*Label, error) { if len(splits) >= 3 { label.Arg = splits[2] } - if label.Schema != SchemeHost && label.Schema != SchemeDocker { - // Not a schema we know: the colon belongs to the label name itself. + kubernetes := label.Schema == SchemeKubernetes && strings.HasPrefix(label.Arg, "//") && len(label.Arg) > 2 + if label.Schema != SchemeHost && label.Schema != SchemeDocker && !kubernetes { + // Not a schema we know, or kubernetes without an image: the colon belongs to the label name itself. return &Label{ Name: str, Schema: SchemeHost, @@ -72,10 +74,11 @@ func (l Labels) PickPlatform(runsOn []string) string { case SchemeDocker: // "//" will be ignored platforms[label.Name] = strings.TrimPrefix(label.Arg, "//") + case SchemeKubernetes: + platforms[label.Name] = SchemeKubernetes + ":" + label.Arg case SchemeHost: platforms[label.Name] = SelfHostedPlatform default: - // unreachable: Parse only produces host or docker schemas continue } } diff --git a/internal/pkg/labels/labels_test.go b/internal/pkg/labels/labels_test.go index d70e1c32..6a665cc0 100644 --- a/internal/pkg/labels/labels_test.go +++ b/internal/pkg/labels/labels_test.go @@ -115,6 +115,7 @@ func TestPickPlatform(t *testing.T) { ls := mustParse(t, "ubuntu:docker://node:18", "self-hosted:host", + "k8s:kubernetes://node:24", ) tests := []struct { @@ -123,6 +124,7 @@ func TestPickPlatform(t *testing.T) { want string }{ {"docker strips leading slashes", []string{"ubuntu"}, "node:18"}, + {"kubernetes keeps its scheme", []string{"k8s"}, "kubernetes://node:24"}, {"host maps to self-hosted marker", []string{"self-hosted"}, SelfHostedPlatform}, {"first match wins", []string{"self-hosted", "ubuntu"}, SelfHostedPlatform}, {"unknown label picks nothing", []string{"windows"}, ""}, diff --git a/tools/test-e2e.sh b/tools/test-e2e.sh index 5920e6a5..d4b4f104 100755 --- a/tools/test-e2e.sh +++ b/tools/test-e2e.sh @@ -4,10 +4,12 @@ set -euo pipefail : "${E2E_CONCURRENCY:?E2E_CONCURRENCY is required}" : "${E2E_JOB_IMAGE:?E2E_JOB_IMAGE is required}" : "${E2E_GITEA_IMAGE:?E2E_GITEA_IMAGE is required}" +: "${E2E_K3S_IMAGE:?E2E_K3S_IMAGE is required}" : "${SERVICE_IMAGE:?SERVICE_IMAGE is required}" docker pull "$E2E_JOB_IMAGE" docker pull "$SERVICE_IMAGE" docker pull "$E2E_GITEA_IMAGE" +docker pull "$E2E_K3S_IMAGE" exec "${GO:-go}" test -tags e2e -count=1 -parallel "$E2E_CONCURRENCY" -timeout 20m -v ./e2e/...