From 2e72cf902ae164b04878626f061403d778b26495 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Sat, 31 Oct 2026 08:00:36 +0000 Subject: [PATCH] enhance(kubernetes): fetch job pod copies from the runner over the pod network (#1354) The Kubernetes backend copies actions and workspaces into job containers with `tar` over `pods/exec`, so every byte goes through the API server. On managed clusters that often sits in other zones than the nodes, so each byte can cross zones twice, and the copies load the API server. With `kubernetes.transfer.mode: auto` or `direct`, an init container copies `gitea-runner` into the job pod, and the job container fetches directories and copies of 64 KiB or more gzipped from the runner over the pod network. `auto` falls back to the API server for a job pod that cannot reach the runner, `direct` fails the job. Copies can hold secrets, so each is served once, over TLS 1.3 pinned to a certificate the runner generates, only to the job pod's IPs. Copying setup-go@v5, cache@v4, setup-java@v4 and build-push-action@v6 sends 11.6 MB through the API server on main and 532 B with this PR. Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: silverwind Reviewed-on: https://gitea.com/gitea/runner/pulls/1354 Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: Lunny Xiao --- act/container/kubernetes.go | 77 ++++++- act/container/kubernetes_test.go | 7 + act/container/kubernetes_transfer.go | 248 ++++++++++++++++++++++ act/container/kubernetes_transfer_test.go | 185 ++++++++++++++++ docs/kubernetes.md | 54 +++++ internal/app/cmd/cmd.go | 3 + internal/app/cmd/kubernetes-transfer.go | 35 +++ internal/app/run/runner.go | 5 + internal/app/run/runner_test.go | 12 ++ internal/pkg/config/config.example.yaml | 12 ++ internal/pkg/config/config.go | 25 +++ internal/pkg/config/config_test.go | 13 ++ 12 files changed, 667 insertions(+), 9 deletions(-) create mode 100644 act/container/kubernetes_transfer.go create mode 100644 act/container/kubernetes_transfer_test.go create mode 100644 internal/app/cmd/kubernetes-transfer.go diff --git a/act/container/kubernetes.go b/act/container/kubernetes.go index 72cb0d1d..401a0253 100644 --- a/act/container/kubernetes.go +++ b/act/container/kubernetes.go @@ -10,10 +10,12 @@ import ( "compress/gzip" "context" "crypto/rand" + "encoding/json" "errors" "fmt" "io" "maps" + "net" "net/http" "net/url" "os" @@ -24,6 +26,7 @@ import ( "time" "gitea.com/gitea/runner/act/common" + "gitea.com/gitea/runner/act/common/logging" "github.com/kballard/go-shellquote" "github.com/spf13/pflag" @@ -37,6 +40,7 @@ type KubernetesOptions struct { MaxLifetime time.Duration ForcePull bool RunnerUUID string + Transfer *KubernetesTransfer } // KubernetesPod runs a job container and its service containers as one pod. @@ -47,6 +51,10 @@ type KubernetesPod struct { job *kubernetesContainer services []*kubernetesContainer arch, path string + ips []net.IP + transfer *transferServer + transferImage string + transferHelper string // empty when copies go through the API server } type kubernetesContainer struct { @@ -76,6 +84,7 @@ type podStatus struct { Spec struct{ Containers []struct{} } Status struct { Phase, Reason, Message string + PodIPs []struct{ IP string } Conditions []struct{ Type, Status, Message string } InitContainerStatuses, ContainerStatuses []containerStatus } @@ -131,14 +140,24 @@ func (p *KubernetesPod) create(ctx context.Context) error { labels[runnerUUIDLabel] = p.options.RunnerUUID } metadata := map[string]any{"name": name, "labels": labels} + var runnerImage string if hostname, err := os.Hostname(); p.client.inCluster && err == nil { - var runner struct{ Metadata struct{ Name, UID string } } + var runner struct { + Metadata struct{ Name, UID string } + Spec struct{ Containers []struct{ Image 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}} + if len(runner.Spec.Containers) == 1 { + runnerImage = runner.Spec.Containers[0].Image + } } else { common.Logger(ctx).Debugf("The runner is not pod %s, job pods get no owner: %v", hostname, err) } } + if err := p.setupTransfer(runnerImage); err != nil { + return err + } secretData := map[string]string{} manifest, err := p.manifest(name, secretData, metadata) if err != nil { @@ -207,6 +226,18 @@ func (p *KubernetesPod) manifest(secret string, secretData map[string]string, me } } spec := map[string]any{"restartPolicy": "Never", "containers": containers, "volumes": volumes} + if p.transferHelper != "" { + transfer := map[string]any{ + "name": "gitea-runner-transfer", "image": p.transferImage, "terminationMessagePolicy": "FallbackToLogsOnError", + "command": []string{"gitea-runner", KubernetesTransferCommand, "install", p.transferHelper}, + "volumeMounts": []any{map[string]any{"name": "act", "mountPath": p.job.GetActPath()}}, + "securityContext": map[string]any{"allowPrivilegeEscalation": false, "capabilities": map[string]any{"drop": []string{"ALL"}}}, + } + if p.options.ForcePull { + transfer["imagePullPolicy"] = "Always" + } + spec["initContainers"] = []any{transfer} + } if len(hostnames) > 0 { spec["hostAliases"] = []any{map[string]any{"ip": "127.0.0.1", "hostnames": hostnames}} } @@ -343,7 +374,13 @@ func (p *KubernetesPod) waitStarted(ctx context.Context) error { return err } if pod.Status.Phase == "Failed" { - return fmt.Errorf("pod %s failed: %s", p.name, cmp.Or(pod.Status.Message, pod.Status.Reason)) + reason := cmp.Or(pod.Status.Message, pod.Status.Reason) + for _, status := range append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...) { + if terminated := status.State.Terminated; reason == "" && terminated != nil && terminated.ExitCode != 0 { + reason = fmt.Sprintf("container %s exited with %d: %s", status.Name, terminated.ExitCode, strings.TrimSpace(terminated.Message)) + } + } + return fmt.Errorf("pod %s failed: %s", p.name, reason) } started := len(pod.Status.ContainerStatuses) == len(pod.Spec.Containers) for _, status := range append(pod.Status.InitContainerStatuses, pod.Status.ContainerStatuses...) { @@ -356,6 +393,9 @@ func (p *KubernetesPod) waitStarted(ctx context.Context) error { } } if started { + for _, ip := range pod.Status.PodIPs { + p.ips = append(p.ips, net.ParseIP(ip.IP)) + } return nil } for _, condition := range pod.Status.Conditions { @@ -457,22 +497,35 @@ func (c *kubernetesContainer) Exec(command []string, env map[string]string, _, w } } -// compressCopyMin is the size from which Copy compresses: below it gzip saves next to nothing on the wire and costs a process in the pod. +// compressCopyMin is the size from which Copy compresses and uses the transfer helper: below it that saves next to nothing on the wire and costs a process in the pod. const compressCopyMin = 64 << 10 -// extract streams the tar write produces into destPath over exec, which goes through the API server. -// With compress it is gzipped on the way, as copied actions shrink about fivefold and API server traffic often crosses zones. +// extract streams the tar write produces into destPath, gzipped with compress, as copied actions shrink about fivefold and API server traffic often crosses zones. func (c *kubernetesContainer) extract(ctx context.Context, destPath string, compress bool, write func(io.Writer) error) error { + if compress && c.pod.transferHelper != "" { + request, done := c.pod.transfer.offer(c.pod.ips, write) + stdin, _ := json.Marshal(request) + var stderr bytes.Buffer + err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{c.pod.transferHelper, KubernetesTransferCommand, "fetch", destPath}, bytes.NewReader(stdin), io.Discard, &stderr) + taken, writeErr := done() + if err == nil || ctx.Err() != nil { + return err + } + if taken || !c.pod.options.Transfer.Fallback { + return fmt.Errorf("copy to %s over the pod network: %w: %s", destPath, errors.Join(writeErr, err), strings.TrimSpace(stderr.String())) + } + c.pod.transferHelper = "" // a pod that cannot reach the runner once will not later + logging.Warnf("Copying into pod %s over the pod network failed, copying through the API server instead: %v: %s", c.pod.name, err, strings.TrimSpace(stderr.String())) + } compress = compress && c.gzip reader, writer := io.Pipe() defer reader.Close() go func() { - if !compress { + if compress { + writer.CloseWithError(writeGzip(writer, write)) + } else { writer.CloseWithError(write(writer)) - return } - compressed, _ := gzip.NewWriterLevel(writer, gzip.BestSpeed) // fast enough to not slow copies down, nearly as small as the default level - writer.CloseWithError(errors.Join(write(compressed), compressed.Close())) }() script := `mkdir -p "$1" && tar -xf - -C "$1"` if compress { @@ -485,6 +538,12 @@ func (c *kubernetesContainer) extract(ctx context.Context, destPath string, comp return nil } +// writeGzip gzips what write produces into w, at a level fast enough to not slow copies down and nearly as small as the default. +func writeGzip(w io.Writer, write func(io.Writer) error) error { + compressed, _ := gzip.NewWriterLevel(w, gzip.BestSpeed) + return errors.Join(write(compressed), compressed.Close()) +} + func (c *kubernetesContainer) Copy(destPath string, files ...*FileEntry) common.Executor { return func(ctx context.Context) error { size := 0 diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go index a729cbe1..04192955 100644 --- a/act/container/kubernetes_test.go +++ b/act/container/kubernetes_test.go @@ -403,6 +403,13 @@ func TestKubernetesPodFailsBeforeSteps(t *testing.T) { 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") + failed, _ := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"phase":"Failed","initContainerStatuses":[ + {"name":"gitea-runner-transfer","state":{"terminated":{"exitCode":1,"message":"exec format error\n"}}}]}}`) + pod, err = NewKubernetesPod(KubernetesOptions{Kubeconfig: failed}) + require.NoError(t, err) + job = pod.JobContainer(&NewContainerInput{Image: "node"}) + require.NoError(t, job.Create(nil, nil)(t.Context())) + require.ErrorContains(t, job.Start(false)(t.Context()), "container gitea-runner-transfer exited with 1: exec format error") plugin := filepath.Join(t.TempDir(), "kubeconfig") require.NoError(t, os.WriteFile(plugin, []byte(` diff --git a/act/container/kubernetes_transfer.go b/act/container/kubernetes_transfer.go new file mode 100644 index 00000000..6ac753c1 --- /dev/null +++ b/act/container/kubernetes_transfer.go @@ -0,0 +1,248 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "cmp" + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "math/big" + "net" + "net/http" + "os" + "os/exec" + "path" + "path/filepath" + "slices" + "strconv" + "strings" + "sync" + "time" + + "gitea.com/gitea/runner/act/common" + "gitea.com/gitea/runner/act/common/logging" +) + +// KubernetesTransfer has job pods fetch large copies from the runner over the pod network instead of through the API server. +type KubernetesTransfer struct { + Host string // empty uses the outbound IP + Port uint16 // 0 picks a free port + Image string // has gitea-runner on its PATH, for an init container to copy into each job pod + Fallback bool // copy through the API server when a job pod cannot reach the runner, else fail +} + +// KubernetesTransferCommand is the hidden runner subcommand job pods run the transfer helper with. +const KubernetesTransferCommand = "kubernetes-transfer" + +// transferRequest goes to the helper on the exec stream's stdin, keeping the token out of the command line and the audit log. +type transferRequest struct { + URL string `json:"url"` + Pin string `json:"pin"` // hex SHA-256 of the runner's certificate +} + +type transferServer struct { + base string + pin string + mu sync.Mutex + pending map[string]transferOffer +} + +type transferOffer struct { + clients []net.IP + write func(io.Writer) error + result chan error +} + +var ( + transferServersMu sync.Mutex + transferServers = map[string]*transferServer{} +) + +// startTransferServer returns the runner's server for these settings, started on first use. +func startTransferServer(host string, port uint16) (*transferServer, error) { + transferServersMu.Lock() + defer transferServersMu.Unlock() + configured := net.JoinHostPort(host, strconv.Itoa(int(port))) + if server, ok := transferServers[configured]; ok { + return server, nil + } + if host == "" { + ip := common.GetOutboundIP() + if ip == nil { + return nil, errors.New("no outbound IP for job pods to reach the runner at, set kubernetes.transfer.host") + } + host = ip.String() + } + privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + return nil, err + } + template := &x509.Certificate{SerialNumber: big.NewInt(1)} // trusted by its pin alone, so no CA, names or validity + certificate, err := x509.CreateCertificate(rand.Reader, template, template, &privateKey.PublicKey, privateKey) + if err != nil { + return nil, err + } + listener, err := net.Listen("tcp", net.JoinHostPort(host, strconv.Itoa(int(port)))) + if err != nil { + return nil, fmt.Errorf("listen for job pod transfers: %w", err) + } + pin := sha256.Sum256(certificate) + server := &transferServer{base: "https://" + listener.Addr().String() + "/", pin: hex.EncodeToString(pin[:]), pending: map[string]transferOffer{}} + httpServer := &http.Server{ + Handler: server, + TLSConfig: &tls.Config{Certificates: []tls.Certificate{{Certificate: [][]byte{certificate}, PrivateKey: privateKey}}, MinVersion: tls.VersionTLS13}, + ReadHeaderTimeout: 10 * time.Second, + IdleTimeout: 30 * time.Second, + } + go func() { _ = httpServer.ServeTLS(listener, "", "") }() + transferServers[configured] = server + return server, nil +} + +// offer serves write once to one of clients until done, which reports whether a client took it and what write failed with. +func (s *transferServer) offer(clients []net.IP, write func(io.Writer) error) (request transferRequest, done func() (taken bool, err error)) { + token := rand.Text() + result := make(chan error, 1) + s.mu.Lock() + s.pending[token] = transferOffer{clients: clients, write: write, result: result} + s.mu.Unlock() + return transferRequest{URL: s.base + token, Pin: s.pin}, func() (bool, error) { + s.mu.Lock() + _, pending := s.pending[token] + delete(s.pending, token) + s.mu.Unlock() + select { + case err := <-result: + return true, err + default: + return !pending, nil + } + } +} + +func (s *transferServer) ServeHTTP(w http.ResponseWriter, r *http.Request) { + host, _, _ := net.SplitHostPort(r.RemoteAddr) + token := strings.TrimPrefix(r.URL.Path, "/") + s.mu.Lock() + offer, ok := s.pending[token] + ok = ok && r.Method == http.MethodGet && slices.ContainsFunc(offer.clients, net.ParseIP(host).Equal) + if ok { + delete(s.pending, token) + } + s.mu.Unlock() + if !ok { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Encoding", "gzip") + err := writeGzip(w, offer.write) + offer.result <- err + if err != nil { + panic(http.ErrAbortHandler) // a cut-off body fails the fetch, a short tar might not + } +} + +// FetchKubernetesTransfer reads a transferRequest from request and extracts the tar it points to into dest. +func FetchKubernetesTransfer(ctx context.Context, request io.Reader, dest string) error { + var transfer transferRequest + if err := json.NewDecoder(request).Decode(&transfer); err != nil { + return fmt.Errorf("read transfer request: %w", err) + } + client := &http.Client{Transport: &http.Transport{ + DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext, + TLSHandshakeTimeout: 10 * time.Second, + ResponseHeaderTimeout: time.Minute, + TLSClientConfig: &tls.Config{ + MinVersion: tls.VersionTLS13, + InsecureSkipVerify: true, // VerifyConnection checks the pin instead + VerifyConnection: func(state tls.ConnectionState) error { + if sum := sha256.Sum256(state.PeerCertificates[0].Raw); hex.EncodeToString(sum[:]) != transfer.Pin { + return errors.New("runner certificate does not match the pin") + } + return nil + }, + }, + }} + httpRequest, err := http.NewRequestWithContext(ctx, http.MethodGet, transfer.URL, nil) + if err != nil { + return err + } + response, err := client.Do(httpRequest) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != http.StatusOK { + return fmt.Errorf("fetch: %s", response.Status) + } + if err := os.MkdirAll(dest, 0o777); err != nil { + return err + } + cmd := exec.CommandContext(ctx, "tar", "-xf", "-", "-C", dest) + cmd.Stdin, cmd.Stdout, cmd.Stderr = response.Body, os.Stderr, os.Stderr + if err := cmd.Run(); err != nil { + return err + } + _, err = io.Copy(io.Discard, response.Body) // tar stops at the first zero blocks, a cut-off body after them still fails + return err +} + +// InstallKubernetesTransferHelper copies the running binary to target. +func InstallKubernetesTransferHelper(target string) error { + executable, err := os.Executable() + if err != nil { + return err + } + source, err := os.Open(executable) + if err != nil { + return err + } + defer source.Close() + if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil { + return err + } + _ = os.Remove(target) // a file left in a reused act volume would keep its mode + file, err := os.OpenFile(target, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o755) + if err != nil { + return err + } + _, err = io.Copy(file, source) + return errors.Join(err, file.Close()) +} + +// setupTransfer turns on transfers for the pod, with the runner pod's image unless one is set. +func (p *KubernetesPod) setupTransfer(runnerImage string) error { + settings := p.options.Transfer + if settings == nil { + return nil + } + image := cmp.Or(settings.Image, runnerImage) + var err error + var server *transferServer + if image == "" { + err = errors.New("set kubernetes.transfer.image, the runner is no single-container pod in kubernetes.namespace to take it from") + } else { + server, err = startTransferServer(settings.Host, settings.Port) + } + if err != nil { + if !settings.Fallback { + return fmt.Errorf("kubernetes.transfer: %w", err) + } + logging.Warnf("Job pod copies go through the API server: %v", err) + return nil + } + p.transfer, p.transferImage = server, image + p.transferHelper = path.Join(p.job.GetActPath(), "bin", "gitea-runner") + return nil +} diff --git a/act/container/kubernetes_transfer_test.go b/act/container/kubernetes_transfer_test.go new file mode 100644 index 00000000..5fb9d24d --- /dev/null +++ b/act/container/kubernetes_transfer_test.go @@ -0,0 +1,185 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package container + +import ( + "bytes" + "context" + "crypto/rand" + "crypto/tls" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "os" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const transferHelperEnv = "GITEA_RUNNER_TEST_TRANSFER_HELPER" + +func TestMain(m *testing.M) { + if os.Getenv(transferHelperEnv) == "1" && len(os.Args) == 4 && os.Args[1] == KubernetesTransferCommand && os.Args[2] == "fetch" { + if err := FetchKubernetesTransfer(context.Background(), os.Stdin, os.Args[3]); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + os.Exit(0) + } + os.Exit(m.Run()) +} + +func fetch(t *testing.T, request transferRequest) error { + t.Helper() + body, err := json.Marshal(request) + require.NoError(t, err) + return FetchKubernetesTransfer(t.Context(), bytes.NewReader(body), t.TempDir()) +} + +func TestTransferServer(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("extracts with tar") + } + server, err := startTransferServer("127.0.0.1", 0) + require.NoError(t, err) + again, err := startTransferServer("127.0.0.1", 0) + require.NoError(t, err) + assert.Same(t, server, again) + loopback := []net.IP{net.ParseIP("127.0.0.1")} + nothing := func(io.Writer) error { return nil } + + t.Run("serves an offer gzipped once to an IP of its job pod", func(t *testing.T) { + request, done := server.offer([]net.IP{net.ParseIP("fd00::1"), loopback[0]}, nothing) + client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, DisableCompression: true}} + response, err := client.Get(request.URL) + require.NoError(t, err) + require.NoError(t, response.Body.Close()) + assert.Equal(t, "gzip", response.Header.Get("Content-Encoding")) + require.ErrorContains(t, fetch(t, request), "404") + taken, _ := done() + assert.True(t, taken) + }) + + t.Run("another pod can neither take nor block an offer", func(t *testing.T) { + request, done := server.offer([]net.IP{net.ParseIP("10.1.2.3")}, nothing) + require.ErrorContains(t, fetch(t, request), "404") + taken, _ := done() + assert.False(t, taken) + }) + + t.Run("a failed write fails the fetch even past the end blocks tar stops at, and done reports it", func(t *testing.T) { + request, done := server.offer(loopback, func(w io.Writer) error { + _, _ = w.Write(make([]byte, 1024)) + _, _ = io.CopyN(w, rand.Reader, 1<<20) + return errors.New("source gone") + }) + require.Error(t, fetch(t, request)) + taken, err := done() + assert.True(t, taken) + require.EqualError(t, err, "source gone") + }) + + t.Run("the helper refuses a runner whose certificate does not match its pin", func(t *testing.T) { + request, done := server.offer(loopback, nothing) + request.Pin = strings.Repeat("0", 64) + require.ErrorContains(t, fetch(t, request), "does not match the pin") + taken, _ := done() + assert.False(t, taken) + }) +} + +func TestKubernetesPodTransferImage(t *testing.T) { + newPod := func(image string, fallback bool) *KubernetesPod { + pod := &KubernetesPod{options: KubernetesOptions{Transfer: &KubernetesTransfer{Host: "127.0.0.1", Image: image, Fallback: fallback}}} + pod.JobContainer(&NewContainerInput{}) + return pod + } + + pod := newPod("", false) + require.NoError(t, pod.setupTransfer("gitea/runner:latest")) + assert.Equal(t, "gitea/runner:latest", pod.transferImage) + + pod = newPod("registry.example/runner:1", false) + require.NoError(t, pod.setupTransfer("gitea/runner:latest")) + assert.Equal(t, "registry.example/runner:1", pod.transferImage) + + pod = newPod("", true) + require.NoError(t, pod.setupTransfer("")) + assert.Empty(t, pod.transferHelper) + + pod = newPod("", false) + require.ErrorContains(t, pod.setupTransfer(""), "kubernetes.transfer.image") +} + +func TestKubernetesPodCopiesLargeFilesOverPodNetwork(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("uses POSIX shell") + } + ctx := t.Context() + workdir, err := filepath.EvalSymlinks(t.TempDir()) + require.NoError(t, err) + kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"phase":"Running","podIPs":[{"ip":"127.0.0.1"}],"containerStatuses":[ + {"name":"job","ready":true,"state":{"running":{}}}]}}`, transferHelperEnv+"=1") + pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig, ForcePull: true, Transfer: &KubernetesTransfer{Host: "127.0.0.1", Image: "gitea/runner:latest", Fallback: true}}) + require.NoError(t, err) + job := pod.JobContainer(&NewContainerInput{Name: "job", Image: "node", WorkingDir: workdir, Stdout: io.Discard, Stderr: io.Discard}) + require.NoError(t, job.Create(nil, nil)(ctx)) + + var manifest struct { + Spec struct { + InitContainers []struct { + Image, ImagePullPolicy string + Command []string + VolumeMounts []struct{ Name, MountPath string } + } + } + } + require.NoError(t, json.Unmarshal([]byte(cluster.requests["POST /api/v1/namespaces/ci/pods"][0]), &manifest)) + require.Len(t, manifest.Spec.InitContainers, 1) + initContainer := manifest.Spec.InitContainers[0] + assert.Equal(t, "gitea/runner:latest", initContainer.Image) + assert.Equal(t, "Always", initContainer.ImagePullPolicy) + assert.Equal(t, []string{"gitea-runner", KubernetesTransferCommand, "install", "/var/run/act/bin/gitea-runner"}, initContainer.Command) + assert.Equal(t, []struct{ Name, MountPath string }{{"act", "/var/run/act"}}, initContainer.VolumeMounts) + require.NoError(t, job.Start(false)(ctx)) + + helper := filepath.Join(t.TempDir(), "bin", "gitea-runner") + require.NoError(t, InstallKubernetesTransferHelper(helper)) + pod.transferHelper = helper + lastExec := func() string { return cluster.requests["exec"][len(cluster.requests["exec"])-1] } + large := &FileEntry{Name: "large", Mode: 0o644, Body: strings.Repeat("a", compressCopyMin)} + copyLarge := func(dir string) error { return job.Copy(filepath.Join(workdir, dir)+"/", large)(ctx) } + require.NoError(t, copyLarge("transferred")) + content, err := os.ReadFile(filepath.Join(workdir, "transferred", "large")) + require.NoError(t, err) + assert.Equal(t, large.Body, string(content)) + assert.Contains(t, lastExec(), "command="+KubernetesTransferCommand) + assert.NotContains(t, lastExec(), "127.0.0.1") + + require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "event.json", Mode: 0o644, Body: `{}`})(ctx)) + assert.NotContains(t, lastExec(), "command="+KubernetesTransferCommand) + + require.NoError(t, os.WriteFile(filepath.Join(workdir, "file"), nil, 0o644)) + require.ErrorContains(t, copyLarge("file"), "over the pod network") + assert.Equal(t, helper, pod.transferHelper) + + unreachable := filepath.Join(workdir, "unreachable") + require.NoError(t, os.WriteFile(unreachable, []byte("#!/bin/sh\necho 'connection refused' >&2\nexit 1\n"), 0o755)) + pod.transferHelper = unreachable + pod.options.Transfer.Fallback = false + require.ErrorContains(t, copyLarge("direct"), "connection refused") + assert.NoFileExists(t, filepath.Join(workdir, "direct", "large")) + + pod.options.Transfer.Fallback = true + require.NoError(t, copyLarge("auto")) + assert.FileExists(t, filepath.Join(workdir, "auto", "large")) + assert.Empty(t, pod.transferHelper) +} diff --git a/docs/kubernetes.md b/docs/kubernetes.md index adcd31d1..26329750 100644 --- a/docs/kubernetes.md +++ b/docs/kubernetes.md @@ -49,6 +49,60 @@ A job with `runs-on: [linux, gpu]` gets the `linux` and then the `gpu` entry of 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`. +## Copying files over the pod network + +The runner copies files into the job container, such as the actions a step uses, with `tar` over `pods/exec`. That data goes through the API server, which on a managed cluster often sits in other zones than the nodes, so each byte can cross zones twice. With `transfer.mode` set to `auto` or `direct`, the job container instead fetches directories, such as actions and workspaces, and other copies of 64 KiB or more gzipped from the runner pod. Smaller ones, like step scripts, stay on `pods/exec`, as do copies out of the job container and step output. + +```yaml +kubernetes: + transfer: + mode: direct # api (default), auto or direct + host: "" # address the runner listens on and job pods reach it at, defaults to its outbound IP, the pod IP in a cluster + port: 0 # port it listens on, defaults to a free one + image: "" # image with gitea-runner on its PATH, defaults to the image of a single-container runner pod in kubernetes.namespace +``` + +Each job pod then gets an init container named `gitea-runner-transfer` from `image`, which copies its `gitea-runner` into the `act` volume for the job container to fetch with over HTTPS. The modes differ in what happens when a job pod cannot reach the runner, for example because of a NetworkPolicy, or when no `image` is set and the runner is not a single-container pod in `kubernetes.namespace`: + +| Mode | Copies | When the pod network does not work | +|---|---|---| +| `api` | through the API server | not used | +| `auto` | over the pod network | the job pod gets its copies through the API server, after one failed attempt, and a warning is logged | +| `direct` | over the pod network | the job fails, so a broken setup shows instead of quietly costing API server traffic again | + +In both `auto` and `direct`, a job fails when its init container does, for example when `image` cannot be pulled or has no `gitea-runner` on its PATH. A private `image` needs `imagePullSecrets` in `pod_template`, and a versioned tag keeps job pods on the runner's version. An `initContainers` entry named `gitea-runner-transfer` in `pod_template` merges into the init container, for example to set its `resources`. + +The copies include secrets, such as `run` scripts with `${{ secrets.* }}` expanded, SSH keys for checkout and the runner's client key, so a runner only serves them to its own job pods: + +- The runner listens with a certificate it generates at start. Its SHA-256 pin reaches the job pod over the `pods/exec` stream, which the API server protects with TLS, and the job pod trusts no other certificate, so nobody on the pod network can read or change a copy. +- Every copy has a one-time token, sent on that stream's stdin, so it shows neither in the command line nor in the API server audit log. +- The runner serves a copy only to a connection from an IP of the job pod it is for. Other pods get a 404 and leave the copy for its job pod. This needs the runner to see the job pods' own addresses, so it does not work for a runner outside the cluster or behind a proxy that rewrites source addresses. + +To also keep other pods from connecting at all, allow the port only from the runner's own job pods, which carry its UUID in the `com.gitea.runner.uuid` label. The policy denies every other connection to the runner pod, so it also needs to allow `cache.port`, and both ports need fixed values: + +```yaml +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: gitea-runner-transfer +spec: + podSelector: + matchLabels: + app: runner # the runner pod + policyTypes: [Ingress] + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: + podSelector: + matchLabels: + com.gitea.runner.uuid: + ports: + - port: 9090 # kubernetes.transfer.port + - port: 9091 # cache.port +``` + ## 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 that is not a valid hostname is sanitized, so `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. diff --git a/internal/app/cmd/cmd.go b/internal/app/cmd/cmd.go index e6fadd9b..e7640fa4 100644 --- a/internal/app/cmd/cmd.go +++ b/internal/app/cmd/cmd.go @@ -58,6 +58,9 @@ func Execute(ctx context.Context) { // ./gitea-runner bug-report rootCmd.AddCommand(loadBugReportCmd()) + // ./gitea-runner kubernetes-transfer, run inside Kubernetes job pods + rootCmd.AddCommand(loadKubernetesTransferCmd()) + // ./gitea-runner config rootCmd.AddCommand(loadConfigCmd(&configFile)) diff --git a/internal/app/cmd/kubernetes-transfer.go b/internal/app/cmd/kubernetes-transfer.go new file mode 100644 index 00000000..adf69745 --- /dev/null +++ b/internal/app/cmd/kubernetes-transfer.go @@ -0,0 +1,35 @@ +// Copyright 2026 The Gitea Authors. All rights reserved. +// SPDX-License-Identifier: MIT + +package cmd + +import ( + "gitea.com/gitea/runner/act/container" + + "github.com/spf13/cobra" +) + +// loadKubernetesTransferCmd is what job pods run when kubernetes.transfer is on: the init container installs the binary, the job container fetches with it. +func loadKubernetesTransferCmd() *cobra.Command { + transferCmd := &cobra.Command{ + Use: container.KubernetesTransferCommand, + Short: "Copy files from the runner into a Kubernetes job pod", + Hidden: true, + } + transferCmd.AddCommand(&cobra.Command{ + Use: "install ", + Short: "Copy this binary to path", + Args: cobra.ExactArgs(1), + RunE: func(_cmd *cobra.Command, args []string) error { + return container.InstallKubernetesTransferHelper(args[0]) + }, + }, &cobra.Command{ + Use: "fetch ", + Short: "Extract the tar stream the transfer request on stdin points to into dir", + Args: cobra.ExactArgs(1), + RunE: func(cmd *cobra.Command, args []string) error { + return container.FetchKubernetesTransfer(cmd.Context(), cmd.InOrStdin(), args[0]) + }, + }) + return transferCmd +} diff --git a/internal/app/run/runner.go b/internal/app/run/runner.go index e26c48b2..61c5d74c 100644 --- a/internal/app/run/runner.go +++ b/internal/app/run/runner.go @@ -415,6 +415,11 @@ func (r *Runner) kubernetesOptions(runsOn []string) container.KubernetesOptions PodTemplates: []map[string]any{r.cfg.Kubernetes.PodTemplate}, RunnerUUID: r.uuid, } + if transfer := r.cfg.Kubernetes.Transfer; transfer.Mode == config.KubernetesTransferModeAuto || transfer.Mode == config.KubernetesTransferModeDirect { + options.Transfer = &container.KubernetesTransfer{ + Host: transfer.Host, Port: transfer.Port, Image: transfer.Image, Fallback: transfer.Mode == config.KubernetesTransferModeAuto, + } + } 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) { diff --git a/internal/app/run/runner_test.go b/internal/app/run/runner_test.go index 70087f49..3f9262df 100644 --- a/internal/app/run/runner_test.go +++ b/internal/app/run/runner_test.go @@ -21,6 +21,7 @@ import ( "gitea.com/gitea/runner/act/artifactcache" "gitea.com/gitea/runner/act/common" "gitea.com/gitea/runner/act/common/logging" + "gitea.com/gitea/runner/act/container" "gitea.com/gitea/runner/act/runner" clientmocks "gitea.com/gitea/runner/internal/pkg/client/mocks" "gitea.com/gitea/runner/internal/pkg/config" @@ -276,6 +277,17 @@ func TestRunnerKubernetesOptionsLayerTheLabelPodTemplatesInRunsOnOrder(t *testin require.Equal(t, []map[string]any{{"base": true}, {"large": true}, {"gpu": true}}, r.kubernetesOptions([]string{"linux", "large", "other", "gpu", "large"}).PodTemplates) } +func TestRunnerKubernetesOptionsTransfer(t *testing.T) { + kubernetesRunner := &Runner{cfg: &config.Config{}} + kubernetesRunner.cfg.Kubernetes.Transfer = config.KubernetesTransfer{Mode: config.KubernetesTransferModeAPI, Host: "10.0.0.5", Port: 4000, Image: "gitea/runner:latest"} + require.Nil(t, kubernetesRunner.kubernetesOptions(nil).Transfer) + + kubernetesRunner.cfg.Kubernetes.Transfer.Mode = config.KubernetesTransferModeAuto + require.Equal(t, &container.KubernetesTransfer{Host: "10.0.0.5", Port: 4000, Image: "gitea/runner:latest", Fallback: true}, kubernetesRunner.kubernetesOptions(nil).Transfer) + kubernetesRunner.cfg.Kubernetes.Transfer.Mode = config.KubernetesTransferModeDirect + require.Equal(t, &container.KubernetesTransfer{Host: "10.0.0.5", Port: 4000, Image: "gitea/runner:latest"}, kubernetesRunner.kubernetesOptions(nil).Transfer) +} + // 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) { diff --git a/internal/pkg/config/config.example.yaml b/internal/pkg/config/config.example.yaml index 22f60e29..0b8891dc 100644 --- a/internal/pkg/config/config.example.yaml +++ b/internal/pkg/config/config.example.yaml @@ -306,6 +306,18 @@ kubernetes: #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: {} + # Have job pods fetch directories, such as actions, and copies of 64 KiB or more gzipped from the runner over the pod network, + # instead of the runner streaming them through the API server. An init container copies gitea-runner from image into each job pod. + # api: through the API server (default). auto: over the pod network, through the API server for a job pod that + # cannot reach the runner. direct: over the pod network only, so a broken setup fails jobs instead of quietly + # costing API server traffic again. + #transfer: + # mode: api + # # Address and port the runner listens on and job pods reach it at, the outbound IP and a free port when empty. + # host: "" + # port: 0 + # # Image with gitea-runner on its PATH, the runner pod's image when it has a single container and runs in namespace. + # image: "" # Optional local task-admission checks. Disabled by default. When enabled, low # disk space or a failing script pauses new task fetching; existing jobs continue. diff --git a/internal/pkg/config/config.go b/internal/pkg/config/config.go index ace85d59..5686d883 100644 --- a/internal/pkg/config/config.go +++ b/internal/pkg/config/config.go @@ -210,6 +210,25 @@ type Kubernetes struct { 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. + Transfer KubernetesTransfer `yaml:"transfer"` // Transfer has job pods fetch copied files from the runner over the pod network instead of through the API server. +} + +// Kubernetes transfer modes: how files the runner copies reach job pods. +const ( + KubernetesTransferModeAPI = "api" // through the API server + KubernetesTransferModeAuto = "auto" // over the pod network, through the API server for a job pod that cannot reach the runner + KubernetesTransferModeDirect = "direct" // over the pod network only, a copy that cannot reach the runner fails +) + +// KubernetesTransferModes are the valid kubernetes.transfer.mode values. +var KubernetesTransferModes = []string{KubernetesTransferModeAPI, KubernetesTransferModeAuto, KubernetesTransferModeDirect} + +// KubernetesTransfer configures copying files into job pods over the pod network. +type KubernetesTransfer struct { + Mode string `yaml:"mode"` // Mode is one of KubernetesTransferModes, api when empty. + Host string `yaml:"host"` // Host is the address the runner listens on and job pods reach it at, empty uses the outbound IP. + Port uint16 `yaml:"port"` // Port the runner listens on for job pods, 0 picks a free one. + Image string `yaml:"image"` // Image has gitea-runner on its PATH, empty uses the image of a single-container runner pod. } // Metrics represents the configuration for the Prometheus metrics endpoint. @@ -346,6 +365,12 @@ func LoadDefault(file string) (*Config, error) { if !slices.Contains(ToolCacheModes, cfg.Runner.ToolCacheMode) { return nil, fmt.Errorf("invalid runner.tool_cache_mode %q: must be one of %q", cfg.Runner.ToolCacheMode, ToolCacheModes) } + if cfg.Kubernetes.Transfer.Mode == "" { + cfg.Kubernetes.Transfer.Mode = KubernetesTransferModeAPI + } + if !slices.Contains(KubernetesTransferModes, cfg.Kubernetes.Transfer.Mode) { + return nil, fmt.Errorf("invalid kubernetes.transfer.mode %q: must be one of %q", cfg.Kubernetes.Transfer.Mode, KubernetesTransferModes) + } if cfg.Runner.LogReportLevel != "" { if _, err := logging.ParseLevel(cfg.Runner.LogReportLevel); err != nil { return nil, fmt.Errorf("invalid runner.log_report_level: %w", err) diff --git a/internal/pkg/config/config_test.go b/internal/pkg/config/config_test.go index 845c7f99..3650a82a 100644 --- a/internal/pkg/config/config_test.go +++ b/internal/pkg/config/config_test.go @@ -140,6 +140,19 @@ func TestLoadDefault_ToolCacheMode(t *testing.T) { assert.Contains(t, err.Error(), "tool_cache_mode") } +func TestLoadDefault_KubernetesTransferMode(t *testing.T) { + cfg, err := LoadDefault(write(t, "kubernetes:\n transfer:\n host: 10.0.0.5\n")) + require.NoError(t, err) + assert.Equal(t, KubernetesTransferModeAPI, cfg.Kubernetes.Transfer.Mode, "copies stay on the API server unless asked") + + cfg, err = LoadDefault(write(t, "kubernetes:\n transfer:\n mode: direct\n")) + require.NoError(t, err) + assert.Equal(t, KubernetesTransferModeDirect, cfg.Kubernetes.Transfer.Mode) + + _, err = LoadDefault(write(t, "kubernetes:\n transfer:\n mode: fast\n")) + require.ErrorContains(t, err, "kubernetes.transfer.mode") +} + func TestLoadDefault_RejectsInvalidLogReportLevel(t *testing.T) { _, err := LoadDefault(write(t, "runner:\n log_report_level: loud\n")) require.ErrorContains(t, err, "runner.log_report_level")