From c719a88f36262a9ab9d6e7b26090fcc4e647a785 Mon Sep 17 00:00:00 2001 From: Zettat123 <39446+zettat123@noreply.gitea.com> Date: Thu, 22 Oct 2026 20:59:48 +0000 Subject: [PATCH] fix(k8s): retry the startup exec until the kubelet accepts it (#1337) A new node can run pods before its kubelet has a serving certificate, so the first exec into the job container fails with `error dialing backend: remote error: tls: internal error` and the job fails during setup. That exec only reads the arch and PATH, so it now retries for up to a minute unless the API server rejects it (4xx) or the command ran. Reviewed-on: https://gitea.com/gitea/runner/pulls/1337 Reviewed-by: bircni Reviewed-by: Lunny Xiao Co-authored-by: Zettat123 <39446+zettat123@noreply.gitea.com> --- act/container/kubernetes.go | 31 +++++++++++++-- act/container/kubernetes_test.go | 66 ++++++++++++++++++++++++++++++-- 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/act/container/kubernetes.go b/act/container/kubernetes.go index aceb5d27..d3f0bb8a 100644 --- a/act/container/kubernetes.go +++ b/act/container/kubernetes.go @@ -387,16 +387,41 @@ func (c *kubernetesContainer) Start(_ bool) common.Executor { 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 { + output, err := c.pod.probe(ctx, c.name) + if err != nil { return err } - machine, path, _ := strings.Cut(stdout.String(), "\n") + machine, path, _ := strings.Cut(output, "\n") c.pod.arch, c.pod.path = goArchToActionArch(machine), path return nil } } +// probe retries exec while it fails short of the container, as a new node runs pods before its kubelet has a serving certificate. +func (p *KubernetesPod) probe(ctx context.Context, container string) (string, error) { + var failure string + 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) + var exitErr ExitCodeError + var apiErr *apiError + if err == nil || ctx.Err() != nil || errors.As(err, &exitErr) || time.Now().After(deadline) || + errors.As(err, &apiErr) && apiErr.code < http.StatusInternalServerError { + return stdout.String(), err + } + if err.Error() != failure { + failure = err.Error() + common.Logger(ctx).Infof("Waiting for exec into pod %s: %v", p.name, err) + } + select { + case <-ctx.Done(): + return "", ctx.Err() + case <-time.After(delay): + } + } +} + func (*kubernetesContainer) Pull(bool) common.Executor { return func(context.Context) error { return nil } } diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go index 48111066..33dca99b 100644 --- a/act/container/kubernetes_test.go +++ b/act/container/kubernetes_test.go @@ -43,8 +43,9 @@ type podManifest struct { } type fakeCluster struct { - mu sync.Mutex - requests map[string][]string + mu sync.Mutex + requests map[string][]string + execFailures []http.HandlerFunc } func (f *fakeCluster) record(request, content string) { @@ -53,6 +54,41 @@ func (f *fakeCluster) record(request, content string) { f.requests[request] = append(f.requests[request], content) } +func (f *fakeCluster) failExec(failures ...http.HandlerFunc) { + f.mu.Lock() + defer f.mu.Unlock() + f.execFailures = failures +} + +func (f *fakeCluster) nextExecFailure() http.HandlerFunc { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.execFailures) == 0 { + return nil + } + failure := f.execFailures[0] + f.execFailures = f.execFailures[1:] + return failure +} + +// execFailure answers like the API server, code 0 for a failure reported after the websocket upgrade. +func execFailure(t *testing.T, code int, status string) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if code != 0 { + w.WriteHeader(code) + _, _ = io.WriteString(w, status) + return + } + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{execProtocol}}) + if err != nil { + t.Error(err) + return + } + defer func() { _ = conn.CloseNow() }() + _ = conn.Write(r.Context(), websocket.MessageBinary, append([]byte{3}, status...)) + } +} + 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) { @@ -70,7 +106,11 @@ func fakeKubernetes(t *testing.T, podStatus string) (string, *fakeCluster) { _, _ = 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) + if failure := cluster.nextExecFailure(); failure != nil { + failure(w, r) + } else { + 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+"/"): @@ -213,7 +253,12 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { 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")) + cluster.failExec( + execFailure(t, http.StatusInternalServerError, `{"message":"error dialing backend: remote error: tls: internal error"}`), + execFailure(t, 0, `{"status":"Failure","reason":"InternalError","message":"Internal error occurred: error sending request: remote error: tls: internal error"}`), + ) require.NoError(t, job.Start(false)(ctx)) + assert.Len(t, cluster.requests["exec"], 3) 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)) @@ -295,6 +340,21 @@ func TestKubernetesPodFailsBeforeSteps(t *testing.T) { } require.ErrorContains(t, job.Create(nil, nil)(t.Context()), test.err) } + running, cluster := fakeKubernetes(t, `{"spec":{"containers":[{}]},"status":{"containerStatuses":[{"name":"job","state":{"running":{}}}]}}`) + for _, test := range []struct { + failure http.HandlerFunc + err string + }{ + {execFailure(t, http.StatusForbidden, `{"message":"pods \"job\" is forbidden"}`), "is forbidden"}, + {execFailure(t, 0, `{"status":"Failure","details":{"causes":[{"reason":"ExitCode","message":"126"}]}}`), ExitCodeError(126).Error()}, + } { + cluster.failExec(test.failure) + pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: running}) + 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()), test.err) + } pod, err := NewKubernetesPod(KubernetesOptions{Kubeconfig: kubeconfig, Namespace: "ci"}) require.NoError(t, err) job := pod.JobContainer(&NewContainerInput{Image: "missing"})