fix(kubernetes): make Docker worker lifecycle reliable (#3676)

2026-09-15 06:33:02 +02:00
parent 941e7950d1
commit 54e2e21f32
16 changed files with 297 additions and 81 deletions
@@ -814,10 +814,7 @@ set -a
 . /opt/kandev/runtime.env
 . /run/kandev/auth.env
 set +a
-if [ ! -f /opt/kandev/prepared ]; then
-  sh /opt/kandev/prepare.sh
-  : > /opt/kandev/prepared
-fi
+[ -f /opt/kandev/prepared ] || { echo 'kubernetes preparation marker is missing' >&2; exit 70; }
 if [ "${` + selectedCheckoutMarker + `:-}" = "1" ]; then
  ` + selectedCheckoutCredentialScrubCommands + `
  rm -f /run/kandev/auth.env 2>/dev/null || true
@@ -11,10 +11,14 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"unicode/utf8"

 	corev1 "k8s.io/api/core/v1"

 	kubeexecutor "github.com/kandev/kandev/internal/agent/kubernetes"
+	"github.com/kandev/kandev/internal/agent/runtime/routingerr"
+	"github.com/kandev/kandev/internal/common/constants"
 	"github.com/kandev/kandev/internal/githubauth"
 	"github.com/kandev/kandev/internal/scriptengine"
 )
@@ -63,6 +67,9 @@ func (r *KubernetesExecutor) bootstrapPod(
 	if err := r.materializeKubernetesRemoteFiles(ctx, runtime, req, pod, profile.MainContainer); err != nil {
 		return err
 	}
+	if err := runKubernetesPrepareScript(ctx, runtime.streams, pod, profile.MainContainer); err != nil {
+		return err
+	}
 	if err := kubernetesWriteFile(ctx, runtime.streams, pod, profile.MainContainer,
 		kubernetesStartPath, nil, 0o600); err != nil {
 		return fmt.Errorf("kubernetes lifecycle: signal bootstrap: %w", err)
@@ -70,6 +77,79 @@ func (r *KubernetesExecutor) bootstrapPod(
 	return nil
 }

+type kubernetesPrepareOutput struct {
+	mu       sync.Mutex
+	data     []byte
+	overflow bool
+}
+
+const kubernetesPrepareCaptureBytes = 256 * 1024
+
+func (w *kubernetesPrepareOutput) Write(p []byte) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if w.overflow {
+		return len(p), nil
+	}
+	if len(p) > kubernetesPrepareCaptureBytes-len(w.data) {
+		w.data = nil
+		w.overflow = true
+		return len(p), nil
+	}
+	w.data = append(w.data, p...)
+	return len(p), nil
+}
+
+func (w *kubernetesPrepareOutput) diagnostic() string {
+	w.mu.Lock()
+	if w.overflow {
+		w.mu.Unlock()
+		return ""
+	}
+	raw := string(w.data)
+	w.mu.Unlock()
+
+	diagnostic := routingerr.SanitizeFullUnbounded(raw)
+	if len(diagnostic) > routingerr.MaxRawExcerptBytes {
+		start := len(diagnostic) - routingerr.MaxRawExcerptBytes
+		for start < len(diagnostic) && !utf8.RuneStart(diagnostic[start]) {
+			start++
+		}
+		diagnostic = diagnostic[start:]
+	}
+	return strings.TrimSpace(diagnostic)
+}
+
+func runKubernetesPrepareScript(
+	ctx context.Context,
+	streams *kubeexecutor.StreamOperations,
+	pod *corev1.Pod,
+	container string,
+) error {
+	prepareCtx, cancel := context.WithTimeout(preparationContext(ctx), constants.SetupScriptTimeout)
+	defer cancel()
+	output := &kubernetesPrepareOutput{}
+	command := `set -eu
+set -a
+. /opt/kandev/runtime.env
+. /run/kandev/auth.env
+set +a
+sh /opt/kandev/prepare.sh
+: > /opt/kandev/prepared`
+	err := streams.Exec(prepareCtx, kubeexecutor.ExecRequest{
+		Namespace: pod.Namespace, Pod: pod.Name, Container: container,
+		Command: []string{"sh", "-c", command}, Stdout: output, Stderr: output,
+	})
+	if err == nil {
+		return nil
+	}
+	diagnostic := output.diagnostic()
+	if diagnostic == "" {
+		return fmt.Errorf("kubernetes lifecycle: prepare script failed: %w", err)
+	}
+	return fmt.Errorf("kubernetes lifecycle: prepare script failed: %s: %w", diagnostic, err)
+}
+
 func kubernetesRuntimeEnvironment(req *ExecutorCreateRequest) map[string]string {
 	values := map[string]string{
 		"AGENTCTL_LISTEN_HOST":        kubernetesControlHost,
@@ -2,6 +2,7 @@ package lifecycle

 import (
 	"context"
+	"errors"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -36,6 +37,96 @@ func TestKubernetesWriteFileDoesNotOpenStdinForEmptySentinel(t *testing.T) {
 	require.Contains(t, command, "mv -f \"$temporary\" \"$destination\"")
 }

+func TestKubernetesBootstrapRunsPrepareBeforeSignallingEntrypoint(t *testing.T) {
+	execs := &recordingKubernetesExec{}
+	streams := kubeexecutor.NewStreamOperations(execs, nil)
+	pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "kandev-agents", Name: "agent-pod"}}
+	executor := NewKubernetesExecutor(nil, newTestLogger())
+	req := validKubernetesCreateRequest()
+
+	err := executor.bootstrapPod(
+		context.Background(),
+		&kubernetesRuntimeClient{streams: streams},
+		req,
+		pod,
+		kubeexecutor.ProfileConfig{MainContainer: "kandev-agent"},
+		"nonce",
+		[]byte("agentctl"),
+	)
+
+	require.NoError(t, err)
+	prepareIndex, signalIndex := -1, -1
+	for index, recorded := range execs.requests {
+		command := strings.Join(recorded.request.Command, " ")
+		if strings.Contains(command, "sh "+kubernetesPreparePath) {
+			prepareIndex = index
+		}
+		if strings.Contains(command, kubernetesStartPath) && recorded.request.Stdin == nil {
+			signalIndex = index
+		}
+	}
+	require.NotEqual(t, -1, prepareIndex)
+	require.NotEqual(t, -1, signalIndex)
+	require.Less(t, prepareIndex, signalIndex)
+	require.NotContains(t, kubernetesBootstrapCommand(), "sh "+kubernetesPreparePath)
+}
+
+func TestKubernetesBootstrapReturnsSanitizedPrepareFailure(t *testing.T) {
+	execs := &recordingKubernetesExec{}
+	execs.execFunc = func(_ context.Context, request kubeexecutor.ExecRequest) error {
+		if !strings.Contains(strings.Join(request.Command, " "), "sh "+kubernetesPreparePath) {
+			return nil
+		}
+		_, _ = request.Stderr.Write([]byte(
+			"token=do-not-persist\nDocker daemon did not become ready before preparation deadline\n",
+		))
+		return errors.New("remote command exited with status 1")
+	}
+	streams := kubeexecutor.NewStreamOperations(execs, nil)
+	pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "kandev-agents", Name: "agent-pod"}}
+	executor := NewKubernetesExecutor(nil, newTestLogger())
+
+	err := executor.bootstrapPod(
+		context.Background(),
+		&kubernetesRuntimeClient{streams: streams},
+		validKubernetesCreateRequest(),
+		pod,
+		kubeexecutor.ProfileConfig{MainContainer: "kandev-agent"},
+		"nonce",
+		[]byte("agentctl"),
+	)
+
+	require.ErrorContains(t, err, "Docker daemon did not become ready before preparation deadline")
+	require.NotContains(t, err.Error(), "do-not-persist")
+	for _, recorded := range execs.requests {
+		command := strings.Join(recorded.request.Command, " ")
+		require.False(t,
+			strings.Contains(command, kubernetesStartPath) && recorded.request.Stdin == nil,
+			"failed preparation must not signal the managed entrypoint",
+		)
+	}
+}
+
+func TestKubernetesPrepareOutputDropsOversizeDiagnosticsBeforeRedaction(t *testing.T) {
+	output := &kubernetesPrepareOutput{}
+	secret := strings.Repeat("Zx.", 100_000)
+	_, err := output.Write([]byte("token=" + secret + "\nlate failure"))
+
+	require.NoError(t, err)
+	require.Empty(t, output.diagnostic(),
+		"oversize raw output must be dropped instead of cutting away a credential marker")
+}
+
+func TestKubernetesPrepareOutputKeepsLatestSanitizedDiagnostic(t *testing.T) {
+	output := &kubernetesPrepareOutput{}
+	_, err := output.Write([]byte(strings.Repeat("ordinary setup output\n", 400) +
+		"token=do-not-persist\nlate failure"))
+
+	require.NoError(t, err)
+	require.Contains(t, output.diagnostic(), "late failure")
+	require.NotContains(t, output.diagnostic(), "do-not-persist")
+}
+
 func TestKubernetesWriteFileCommandRejectsSymlinkedAncestor(t *testing.T) {
 	root := t.TempDir()
 	outside := t.TempDir()
@@ -153,8 +244,8 @@ func TestKubernetesBootstrapPublishesManagedCredentialHelperPath(t *testing.T) {
 	)
 	require.NotContains(t, authData, "/tmp/hostile-agentctl")
 	command := kubernetesBootstrapCommand()
-	require.Less(t, strings.Index(command, ". "+kubernetesAuthEnvPath), strings.Index(command, "sh "+kubernetesPreparePath))
 	require.Less(t, strings.Index(command, ". "+kubernetesAuthEnvPath), strings.Index(command, "exec "+kubernetesAgentctlPath))
+	require.NotContains(t, command, "sh "+kubernetesPreparePath)
 }

 func TestKubernetesPrepareScriptUsesManagedAgentctlPathWithoutStartingSecondProcess(t *testing.T) {
@@ -351,9 +351,10 @@ type recordingKubernetesExec struct {
 	mu       sync.Mutex
 	requests []recordedKubernetesExec
 	err      error
+	execFunc func(context.Context, kubeexecutor.ExecRequest) error
 }

-func (r *recordingKubernetesExec) Exec(_ context.Context, request kubeexecutor.ExecRequest) error {
+func (r *recordingKubernetesExec) Exec(ctx context.Context, request kubeexecutor.ExecRequest) error {
 	var data []byte
 	if request.Stdin != nil {
 		var err error
@@ -363,8 +364,11 @@ func (r *recordingKubernetesExec) Exec(_ context.Context, request kubeexecutor.E
 		}
 	}
 	r.mu.Lock()
-	defer r.mu.Unlock()
 	r.requests = append(r.requests, recordedKubernetesExec{request: request, stdin: data})
+	r.mu.Unlock()
+	if r.execFunc != nil {
+		return r.execFunc(ctx, request)
+	}
 	return r.err
 }

@@ -148,7 +148,7 @@ func TestKubernetesCreateInstanceProvisionsBootstrapsAndForwardsAgentctl(t *test
 	require.NotNil(t, instance.Client)
 	require.Len(t, resources.createdPVCs, 1)
 	require.Len(t, resources.createdPods, 1)
-	require.Len(t, execs.requests, 4, "binary, runtime config, auth config, then start signal")
+	require.Len(t, execs.requests, 5, "binary, runtime config, auth config, prepare, then start signal")
 	require.Equal(t, []uint16{uint16(kubeexecutor.DefaultAgentctlPort), 41001}, forwards.remotePorts())
 	for _, request := range forwards.requests {
 		require.Equal(t, "127.0.0.1", request.LocalAddress)
@@ -428,7 +428,7 @@ func TestKubernetesCreateInstanceRebuildsLostPodAgainstVerifiedManagedPVC(t *tes
 	require.Equal(t, "pvc-uid", reconnected.Metadata[MetadataKeyKubernetesPVCUID])
 	require.Len(t, resources.createdPVCs, 1, "resume must reuse the recorded PVC")
 	require.Len(t, resources.createdPods, 2)
-	require.Len(t, reconnectExecs.requests, 4, "replacement Pod needs full bootstrap materialization")
+	require.Len(t, reconnectExecs.requests, 5, "replacement Pod needs full bootstrap materialization")
 }

 func TestKubernetesStopInstancePreservesOrdinaryStopAndForceCleansManagedResources(t *testing.T) {