From ffd80101a317d6c3ef922dfb4856b629b3d60064 Mon Sep 17 00:00:00 2001 From: Lunny Xiao Date: Wed, 28 Oct 2026 06:00:51 +0000 Subject: [PATCH] enhance(kubernetes): gzip copies into job containers over exec (#1355) The Kubernetes backend copies actions and other files into the job container with `tar` over `pods/exec`, so the data goes through the API server, which on managed clusters often sits in other zones than the nodes (each copied byte then crosses zones twice). Actions are megabytes of JavaScript bundles that gzip about fivefold: | Action | tar | gzip | |---|---|---| | actions/setup-go@v5 | 8.0 MB | 1.5 MB | | actions/cache@v4 | 11.6 MB | 2.0 MB | | actions/setup-java@v4 | 15.3 MB | 2.5 MB | | docker/build-push-action@v6 | 12.9 MB | 3.0 MB | The startup probe now also reports whether the job container has `gzip`. If it does, `CopyDir` and copies of 64 KiB or more are gzipped at `BestSpeed` and extracted with `gzip -dc | tar -xf -`. Smaller copies (run scripts, file command files) stay uncompressed, where gzip would save next to nothing and cost a process in the pod. Images without `gzip` keep the plain tar stream, so the image requirements do not change. This is independent of #1354 and gets most of its traffic saving with no new moving parts. --- *Written by Codet (AI coding agent) on behalf of @lunny.* --------- Co-authored-by: Zettat123 Reviewed-on: https://gitea.com/gitea/runner/pulls/1355 Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com> --- act/container/kubernetes.go | 42 ++++++++++++++++++++++++------ act/container/kubernetes_test.go | 44 +++++++++++++++++++++++++++++++- 2 files changed, 77 insertions(+), 9 deletions(-) diff --git a/act/container/kubernetes.go b/act/container/kubernetes.go index 5c5bd186..72cb0d1d 100644 --- a/act/container/kubernetes.go +++ b/act/container/kubernetes.go @@ -7,6 +7,7 @@ import ( "bufio" "bytes" "cmp" + "compress/gzip" "context" "crypto/rand" "errors" @@ -55,6 +56,7 @@ type kubernetesContainer struct { name string input *NewContainerInput probe map[string]any + gzip bool // the container can gunzip, so copies into it go compressed } type containerStatus struct { @@ -391,8 +393,11 @@ func (c *kubernetesContainer) Start(_ bool) common.Executor { if err != nil { return err } - machine, path, _ := strings.Cut(output, "\n") - c.pod.arch, c.pod.path = goArchToActionArch(machine), path + probed := strings.SplitN(output, "\n", 3) + for len(probed) < 3 { + probed = append(probed, "") + } + c.pod.arch, c.gzip, c.pod.path = goArchToActionArch(probed[0]), probed[1] == "gzip", probed[2] return nil } } @@ -403,7 +408,7 @@ func (p *KubernetesPod) probe(ctx context.Context, container string) (string, er deadline := time.Now().Add(time.Minute) for delay := 100 * time.Millisecond; ; delay = min(2*delay, 2*time.Second) { var stdout bytes.Buffer - err := p.client.exec(ctx, p.name, container, []string{"sh", "-c", `printf '%s\n%s' "$(uname -m)" "$PATH"`}, nil, &stdout, io.Discard) + err := p.client.exec(ctx, p.name, container, []string{"sh", "-c", `printf '%s\n%s\n%s' "$(uname -m)" "$(command -v gzip >/dev/null && echo gzip)" "$PATH"`}, nil, &stdout, io.Discard) var exitErr ExitCodeError var apiErr *apiError if err == nil || ctx.Err() != nil || errors.As(err, &exitErr) || time.Now().After(deadline) || @@ -452,12 +457,29 @@ func (c *kubernetesContainer) Exec(command []string, env map[string]string, _, w } } -func (c *kubernetesContainer) extract(ctx context.Context, destPath string, write func(io.Writer) error) error { +// 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. +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. +func (c *kubernetesContainer) extract(ctx context.Context, destPath string, compress bool, write func(io.Writer) error) error { + compress = compress && c.gzip reader, writer := io.Pipe() defer reader.Close() - go func() { writer.CloseWithError(write(writer)) }() + go func() { + if !compress { + 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 { + script = `mkdir -p "$1" && gzip -dc | tar -xf - -C "$1"` // gzip -dc rather than tar -z, which GNU tar runs gzip for anyway + } 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 { + if err := c.pod.client.exec(ctx, c.pod.name, c.name, []string{"sh", "-c", script, "sh", destPath}, reader, io.Discard, &stderr); err != nil { return fmt.Errorf("extract to %s: %w: %s", destPath, err, strings.TrimSpace(stderr.String())) } return nil @@ -465,13 +487,17 @@ func (c *kubernetesContainer) extract(ctx context.Context, destPath string, writ 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...) }) + size := 0 + for _, file := range files { + size += len(file.Body) + } + return c.extract(ctx, destPath, size >= compressCopyMin, 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, destPath, func(writer io.Writer) error { + return c.extract(ctx, destPath, true, func(writer io.Writer) error { // directories are actions and workspaces, megabytes of compressible sources return writeDirTar(ctx, writer, "/", srcPath, useGitIgnore, skipGitDir, 0, 0) }) } diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go index 4b48a67f..a729cbe1 100644 --- a/act/container/kubernetes_test.go +++ b/act/container/kubernetes_test.go @@ -260,6 +260,7 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { ) require.NoError(t, job.Start(false)(ctx)) assert.Len(t, cluster.requests["exec"], 3) + assert.True(t, pod.job.gzip) 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)) @@ -276,7 +277,9 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { assert.NotContains(t, query, "hunter2") } + lastExec := func() string { return cluster.requests["exec"][len(cluster.requests["exec"])-1] } require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "workflow/event.json", Mode: 0o644, Body: `{"ref":"main"}`})(ctx)) + assert.NotContains(t, lastExec(), "gzip") archive, err := job.GetContainerArchive(ctx, workdir+"/act/workflow/event.json") require.NoError(t, err) reader := tar.NewReader(archive) @@ -289,6 +292,12 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { require.NoError(t, archive.Close()) _, err = job.GetContainerArchive(ctx, workdir+"/missing") require.Error(t, err) + large := strings.Repeat("a", compressCopyMin) + require.NoError(t, job.Copy(workdir+"/act/", &FileEntry{Name: "workflow/large", Mode: 0o644, Body: large})(ctx)) + assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |")) + content, err = os.ReadFile(workdir + "/act/workflow/large") + require.NoError(t, err) + assert.Equal(t, large, string(content)) actionDir := filepath.Join(t.TempDir(), "action") require.NoError(t, os.MkdirAll(actionDir, 0o755)) @@ -297,7 +306,8 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { script, err := os.ReadFile(workdir + "/act/actions/sha/index.js") require.NoError(t, err) assert.Equal(t, "main()", string(script)) - assert.Contains(t, cluster.requests["exec"][len(cluster.requests["exec"])-1], "command="+url.QueryEscape(workdir+"/act/actions/sha/")) + assert.Contains(t, lastExec(), "command="+url.QueryEscape(workdir+"/act/actions/sha/")) + assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |")) pathEnv := map[string]string{} require.NoError(t, job.UpdateFromImageEnv(&pathEnv)(ctx)) @@ -321,6 +331,38 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { assert.Empty(t, cluster.requests["DELETE /api/v1/namespaces/ci/secrets/new"]) } +func TestKubernetesCopiesUncompressedWithoutGzip(t *testing.T) { + t.Parallel() + if runtime.GOOS == "windows" { + t.Skip("uses POSIX shell") + } + ctx := t.Context() + workdir, bin := t.TempDir(), t.TempDir() + for _, tool := range []string{"uname", "tar", "mkdir"} { // all the job container needs but gzip, as on a minimal image + found, err := exec.LookPath(tool) + require.NoError(t, err) + require.NoError(t, os.Symlink(found, filepath.Join(bin, tool))) + } + kubeconfig, cluster := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"phase":"Running","containerStatuses":[ + {"name":"job","ready":true,"state":{"running":{}}}]}}`, "PATH="+bin) + pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig}) + require.NoError(t, err) + job := pod.JobContainer(&NewContainerInput{Image: "node", WorkingDir: workdir}) + require.NoError(t, job.Create(nil, nil)(ctx)) + require.NoError(t, job.Start(false)(ctx)) + assert.False(t, pod.job.gzip) + assert.Equal(t, bin, pod.path) + + actionDir := filepath.Join(t.TempDir(), "action") + require.NoError(t, os.MkdirAll(actionDir, 0o755)) + require.NoError(t, os.WriteFile(filepath.Join(actionDir, "index.js"), []byte("main()"), 0o644)) + require.NoError(t, job.CopyDir(workdir+"/act/actions/sha/", actionDir+"/", false, false)(ctx)) + script, err := os.ReadFile(workdir + "/act/actions/sha/index.js") + require.NoError(t, err) + assert.Equal(t, "main()", string(script)) + assert.NotContains(t, cluster.requests["exec"][len(cluster.requests["exec"])-1], "gzip") +} + func TestKubernetesPodFailsBeforeSteps(t *testing.T) { kubeconfig, _ := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"containerStatuses":[ {"name":"job","state":{"waiting":{"reason":"ErrImagePull","message":"no such image"}}}]}}`)