Use PATCH instead of UPDATE in StopSidecars to avoid 409 conflicts

Author: prad9192

examples/v1/pipelineruns/pipelinerun-with-task-timeout-override.yaml

@@ -21,8 +21,8 @@ spec:
     script: |
       #!/bin/sh
       echo "- Pipeline task timeout: 60s (defined in pipeline spec)"
-      echo "- TaskRunSpecs override: 20s (defined in pipelinerun spec)"
-      echo "- Actual task duration: 10s (within 20s override timeout)"
+      echo "- TaskRunSpecs override: 40s (defined in pipelinerun spec)"
+      echo "- Actual task duration: 10s (within 40s override timeout)"
 ---
 apiVersion: tekton.dev/v1
 kind: Pipeline
@@ -50,4 +50,4 @@ spec:
     name: taskrun-timeout-override
   taskRunSpecs:
   - pipelineTaskName: long-running-task
-    timeout: "20s"      # 20s timeout (can't actually test with a timeout lesser than  10s else the task fails)
+    timeout: "40s"      # 40s timeout (accounts for image pull time + 10s sleep)

pkg/pod/entrypoint.go

@@ -271,6 +271,43 @@ func init() {
 	}
 }
 
+// buildSidecarStopPatch creates a JSON Patch to replace sidecar container images with nop image
+func buildSidecarStopPatch(pod *corev1.Pod, nopImage string, ctx context.Context) ([]byte, error) {
+	var patchOps []jsonpatch.JsonPatchOperation
+
+	// Iterate over container statuses to find running sidecars
+	for _, s := range pod.Status.ContainerStatuses {
+		// If the results-from is set to sidecar logs,
+		// a sidecar container with name `sidecar-log-results` is injected by the reconciler.
+		// Do not kill this sidecar. Let it exit gracefully.
+		if config.FromContextOrDefaults(ctx).FeatureFlags.ResultExtractionMethod == config.ResultExtractionMethodSidecarLogs && s.Name == pipeline.ReservedResultsSidecarContainerName {
+			continue
+		}
+		// Stop any running container that isn't a step.
+		// An injected sidecar container might not have the
+		// "sidecar-" prefix, so we can't just look for that prefix.
+		if !IsContainerStep(s.Name) && s.State.Running != nil {
+			// Find the corresponding container in the spec by name to get the correct index
+			for i, c := range pod.Spec.Containers {
+				if c.Name == s.Name && c.Image != nopImage {
+					patchOps = append(patchOps, jsonpatch.JsonPatchOperation{
+						Operation: "replace",
+						Path:      fmt.Sprintf("/spec/containers/%d/image", i),
+						Value:     nopImage,
+					})
+					break
+				}
+			}
+		}
+	}
+
+	if len(patchOps) == 0 {
+		return nil, nil
+	}
+
+	return json.Marshal(patchOps)
+}
+
 // CancelPod cancels the pod
 func CancelPod(ctx context.Context, kubeClient kubernetes.Interface, namespace, podName string) error {
 	// PATCH the Pod's annotations to replace the cancel annotation with the
@@ -296,43 +333,37 @@ func UpdateReady(ctx context.Context, kubeclient kubernetes.Interface, pod corev
 // StopSidecars updates sidecar containers in the Pod to a nop image, which
 // exits successfully immediately.
 func StopSidecars(ctx context.Context, nopImage string, kubeclient kubernetes.Interface, namespace, name string) (*corev1.Pod, error) {
-	newPod, err := kubeclient.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{})
+	pod, err := kubeclient.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{})
 	if k8serrors.IsNotFound(err) {
 		// return NotFound as-is, since the K8s error checks don't handle wrapping.
 		return nil, err
 	} else if err != nil {
 		return nil, fmt.Errorf("error getting Pod %q when stopping sidecars: %w", name, err)
 	}
 
-	updated := false
-	if newPod.Status.Phase == corev1.PodRunning {
-		for _, s := range newPod.Status.ContainerStatuses {
-			// If the results-from is set to sidecar logs,
-			// a sidecar container with name `sidecar-log-results` is injected by the reconiler.
-			// Do not kill this sidecar. Let it exit gracefully.
-			if config.FromContextOrDefaults(ctx).FeatureFlags.ResultExtractionMethod == config.ResultExtractionMethodSidecarLogs && s.Name == pipeline.ReservedResultsSidecarContainerName {
-				continue
-			}
-			// Stop any running container that isn't a step.
-			// An injected sidecar container might not have the
-			// "sidecar-" prefix, so we can't just look for that
-			// prefix.
-			if !IsContainerStep(s.Name) && s.State.Running != nil {
-				for j, c := range newPod.Spec.Containers {
-					if c.Name == s.Name && c.Image != nopImage {
-						updated = true
-						newPod.Spec.Containers[j].Image = nopImage
-					}
-				}
-			}
-		}
+	// Only attempt to stop sidecars if pod is running
+	if pod.Status.Phase != corev1.PodRunning {
+		return pod, nil
 	}
-	if updated {
-		if newPod, err = kubeclient.CoreV1().Pods(newPod.Namespace).Update(ctx, newPod, metav1.UpdateOptions{}); err != nil {
-			return nil, fmt.Errorf("error stopping sidecars of Pod %q: %w", name, err)
-		}
+
+	// Build JSON Patch operations to replace sidecar images
+	patchBytes, err := buildSidecarStopPatch(pod, nopImage, ctx)
+	if err != nil {
+		return nil, fmt.Errorf("error building patch for stopping sidecars of Pod %q: %w", name, err)
 	}
-	return newPod, nil
+
+	// If no sidecars need to be stopped, return early
+	if patchBytes == nil {
+		return pod, nil
+	}
+
+	// PATCH the Pod's container images to stop sidecars, using the same pattern as UpdateReady and CancelPod
+	patchedPod, err := kubeclient.CoreV1().Pods(namespace).Patch(ctx, name, types.JSONPatchType, patchBytes, metav1.PatchOptions{})
+	if err != nil {
+		return nil, fmt.Errorf("error stopping sidecars of Pod %q: %w", name, err)
+	}
+
+	return patchedPod, nil
 }
 
 // IsSidecarStatusRunning determines if any SidecarStatus on a TaskRun

pkg/pod/entrypoint_test.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	corev1 "k8s.io/api/core/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
@@ -1097,6 +1098,222 @@ func TestStopSidecars(t *testing.T) {
 	}
 }
 
+// TestStopSidecarsUsesPatch verifies that StopSidecars uses PATCH instead of UPDATE.
+func TestStopSidecarsUsesPatch(t *testing.T) {
+	stepContainer := corev1.Container{
+		Name:  stepPrefix + "my-step",
+		Image: "foo",
+	}
+	sidecarContainer := corev1.Container{
+		Name:  sidecarPrefix + "my-sidecar",
+		Image: "original-sidecar-image",
+	}
+	injectedSidecar := corev1.Container{
+		Name:  "injected-sidecar",
+		Image: "original-injected-image",
+	}
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "test-pod",
+			Namespace:       "default",
+			ResourceVersion: "1000",
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{stepContainer, sidecarContainer, injectedSidecar},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+			ContainerStatuses: []corev1.ContainerStatus{{
+				Name: stepContainer.Name,
+				State: corev1.ContainerState{
+					Terminated: &corev1.ContainerStateTerminated{
+						ExitCode: 0,
+					},
+				},
+			}, {
+				Name: sidecarContainer.Name,
+				State: corev1.ContainerState{
+					Running: &corev1.ContainerStateRunning{
+						StartedAt: metav1.NewTime(time.Now()),
+					},
+				},
+			}, {
+				Name: injectedSidecar.Name,
+				State: corev1.ContainerState{
+					Running: &corev1.ContainerStateRunning{
+						StartedAt: metav1.NewTime(time.Now()),
+					},
+				},
+			}},
+		},
+	}
+
+	kubeclient := fakek8s.NewSimpleClientset(pod)
+
+	var patchCalled, updateCalled bool
+	var patchType string
+	var patchBytes []byte
+
+	kubeclient.PrependReactor("patch", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		patchAction := action.(k8stesting.PatchAction)
+		patchCalled = true
+		patchType = string(patchAction.GetPatchType())
+		patchBytes = patchAction.GetPatch()
+
+		patchedPod := pod.DeepCopy()
+		patchedPod.Spec.Containers[1].Image = nopImage
+		patchedPod.Spec.Containers[2].Image = nopImage
+		return true, patchedPod, nil
+	})
+	kubeclient.PrependReactor("update", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		updateCalled = true
+		return true, nil, nil
+	})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	got, err := StopSidecars(ctx, nopImage, kubeclient, pod.Namespace, pod.Name)
+	if err != nil {
+		t.Fatalf("StopSidecars failed: %v", err)
+	}
+
+	if !patchCalled {
+		t.Error("expected PATCH to be called")
+	}
+	if updateCalled {
+		t.Error("expected UPDATE not to be called")
+	}
+
+	if patchType != "application/json-patch+json" {
+		t.Errorf("expected patch type 'application/json-patch+json', got %q", patchType)
+	}
+
+	patchStr := string(patchBytes)
+	if !containsSubstr(patchStr, "/spec/containers/1/image") {
+		t.Errorf("patch should target container 1, got: %s", patchStr)
+	}
+	if !containsSubstr(patchStr, "/spec/containers/2/image") {
+		t.Errorf("patch should target container 2, got: %s", patchStr)
+	}
+	if containsSubstr(patchStr, "/spec/containers/0/image") {
+		t.Errorf("patch should not target step container, got: %s", patchStr)
+	}
+
+	if got.Spec.Containers[1].Image != nopImage {
+		t.Errorf("expected sidecar image %q, got %q", nopImage, got.Spec.Containers[1].Image)
+	}
+	if got.Spec.Containers[2].Image != nopImage {
+		t.Errorf("expected injected sidecar image %q, got %q", nopImage, got.Spec.Containers[2].Image)
+	}
+	if got.Spec.Containers[0].Image != stepContainer.Image {
+		t.Errorf("step container should be unchanged, got %q", got.Spec.Containers[0].Image)
+	}
+}
+
+// TestStopSidecarsWithConflictSimulation simulates concurrent pod updates
+// that cause 409 conflicts with UPDATE but succeed with PATCH.
+func TestStopSidecarsWithConflictSimulation(t *testing.T) {
+	stepContainer := corev1.Container{
+		Name:  stepPrefix + "my-step",
+		Image: "foo",
+	}
+	sidecarContainer := corev1.Container{
+		Name:  sidecarPrefix + "my-sidecar",
+		Image: "original-image",
+	}
+
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:            "test-pod",
+			Namespace:       "default",
+			ResourceVersion: "1000",
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{stepContainer, sidecarContainer},
+		},
+		Status: corev1.PodStatus{
+			Phase: corev1.PodRunning,
+			ContainerStatuses: []corev1.ContainerStatus{{
+				Name: stepContainer.Name,
+				State: corev1.ContainerState{
+					Terminated: &corev1.ContainerStateTerminated{},
+				},
+			}, {
+				Name: sidecarContainer.Name,
+				State: corev1.ContainerState{
+					Running: &corev1.ContainerStateRunning{
+						StartedAt: metav1.NewTime(time.Now()),
+					},
+				},
+			}},
+		},
+	}
+
+	kubeclient := fakek8s.NewSimpleClientset(pod)
+
+	podUpdated := false
+
+	kubeclient.PrependReactor("get", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		p := pod.DeepCopy()
+		if !podUpdated {
+			p.ResourceVersion = "1000"
+			podUpdated = true
+		} else {
+			p.ResourceVersion = "1001"
+			p.Status.ContainerStatuses[0].State.Terminated.FinishedAt = metav1.NewTime(time.Now())
+		}
+		return true, p, nil
+	})
+
+	kubeclient.PrependReactor("update", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		updateAction := action.(k8stesting.UpdateAction)
+		updatedPod := updateAction.GetObject().(*corev1.Pod)
+
+		if podUpdated && updatedPod.ResourceVersion == "1000" {
+			return true, nil, k8serrors.NewConflict(
+				corev1.Resource("pods"),
+				pod.Name,
+				errors.New("the object has been modified; please apply your changes to the latest version"),
+			)
+		}
+		return false, nil, nil
+	})
+
+	kubeclient.PrependReactor("patch", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		patchedPod := pod.DeepCopy()
+		patchedPod.Spec.Containers[1].Image = nopImage
+		patchedPod.ResourceVersion = "1002"
+		return true, patchedPod, nil
+	})
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+
+	got, err := StopSidecars(ctx, nopImage, kubeclient, pod.Namespace, pod.Name)
+	if err != nil {
+		if k8serrors.IsConflict(err) {
+			t.Fatalf("got 409 conflict, this indicates UPDATE is being used instead of PATCH: %v", err)
+		}
+		t.Fatalf("unexpected error: %v", err)
+	}
+
+	if got.Spec.Containers[1].Image != nopImage {
+		t.Errorf("expected sidecar image %q, got %q", nopImage, got.Spec.Containers[1].Image)
+	}
+}
+
+// Helper function to check if a string contains a substring
+func containsSubstr(s, substr string) bool {
+	for i := 0; i <= len(s)-len(substr); i++ {
+		if s[i:i+len(substr)] == substr {
+			return true
+		}
+	}
+	return false
+}
+
 func TestCancelPod(t *testing.T) {
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{