engine/engine_impl.go: Add a mutex around pod image updates for step progress.

engine/engine_impl.go

@@ -28,10 +28,10 @@ import (
 )
 
 var backoff = wait.Backoff{
-	Steps:    5,
+	Steps:    15,
 	Duration: 500 * time.Millisecond,
 	Factor:   1.0,
-	Jitter:   0.1,
+	Jitter:   0.5,
 }
 
 // Kubernetes implements a Kubernetes pipeline engine.
@@ -246,6 +246,12 @@ func (k *Kubernetes) tail(ctx context.Context, spec *Spec, step *Step, output io
 
 func (k *Kubernetes) start(spec *Spec, step *Step) error {
 	err := retry.RetryOnConflict(backoff, func() error {
+		// We protect this read/modify/write with a mutex to reduce the
+		// chance of a self-inflicted concurrent modification error
+		// when a DAG in a pipeline is fanning out and we have a lot of
+		// steps to Start at once.
+		spec.podUpdateMutex.Lock()
+		defer spec.podUpdateMutex.Unlock()
 		pod, err := k.client.CoreV1().Pods(spec.PodSpec.Namespace).Get(spec.PodSpec.Name, metav1.GetOptions{})
 		if err != nil {
 			return err

engine/spec.go

@@ -4,6 +4,10 @@
 
 package engine
 
+import (
+	"sync"
+)
+
 type (
 	// Spec provides the pipeline spec. This provides the
 	// required instructions for reproducible pipeline
@@ -15,6 +19,10 @@ type (
 		Volumes    []*Volume          `json:"volumes,omitempty"`
 		Secrets    map[string]*Secret `json:"secrets,omitempty"`
 		PullSecret *Secret            `json:"pull_secrets,omitempty"`
+		// Runtime field to gate updating of the pod that this pipeline
+		// is running on. Helps to avoid self-inflicted 409 Conflict
+		// responses from the kubernetes api server.
+		podUpdateMutex sync.Mutex
 	}
 
 	// Step defines a pipeline step.