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 <zettat123@gmail.com>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1355
Reviewed-by: Zettat123 <39446+zettat123@noreply.gitea.com>

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

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"}}}]}}`)