add option to restart prowjob when node is terminated (enabled by default)

Signed-off-by: Tim Ramlot <[email protected]>

Author: inteon

config/prow/cluster/prowjob-crd/prowjob_customresourcedefinition.yaml

@@ -484,6 +484,12 @@ spec:
                   the job is evicted. If this field is unspecified or false, a new
                   pod will be created to replace the evicted one.
                 type: boolean
+              error_on_termination:
+                description: ErrorOnTermination indicates that the ProwJob should
+                  be completed and given the ErrorState status if the pod that is
+                  executing the job is terminated. If this field is unspecified or
+                  false, a new pod will be created to replace the terminated one.
+                type: boolean
               extra_refs:
                 description: ExtraRefs are auxiliary repositories that need to be
                   cloned, determined from config

pkg/apis/prowjobs/v1/types.go

@@ -182,6 +182,11 @@ type ProwJobSpec struct {
 	// If this field is unspecified or false, a new pod will be created to replace
 	// the evicted one.
 	ErrorOnEviction bool `json:"error_on_eviction,omitempty"`
+	// ErrorOnTermination indicates that the ProwJob should be completed and given
+	// the ErrorState status if the pod that is executing the job is terminated.
+	// If this field is unspecified or false, a new pod will be created to replace
+	// the terminated one.
+	ErrorOnTermination bool `json:"error_on_termination,omitempty"`
 
 	// PodSpec provides the basis for running the test under
 	// a Kubernetes agent

pkg/config/config.go

@@ -2894,6 +2894,8 @@ func validateAgent(v JobBase, podNamespace string) error {
 		return fmt.Errorf("decoration requires agent: %s (found %q)", k, agent)
 	case v.ErrorOnEviction && agent != k:
 		return fmt.Errorf("error_on_eviction only applies to agent: %s (found %q)", k, agent)
+	case v.ErrorOnTermination && agent != k:
+		return fmt.Errorf("error_on_termination only applies to agent: %s (found %q)", k, agent)
 	case v.Namespace == nil || *v.Namespace == "":
 		return fmt.Errorf("failed to default namespace")
 	case *v.Namespace != podNamespace && agent != p:

pkg/config/config_test.go

@@ -1428,6 +1428,13 @@ func TestValidateAgent(t *testing.T) {
 			},
 			pass: true,
 		},
+		{
+			name: "error_on_termination allowed for kubernetes agent",
+			base: func(j *JobBase) {
+				j.ErrorOnTermination = true
+			},
+			pass: true,
+		},
 	}
 
 	for _, tc := range cases {

pkg/config/jobs.go

@@ -109,6 +109,11 @@ type JobBase struct {
 	// If this field is unspecified or false, a new pod will be created to replace
 	// the evicted one.
 	ErrorOnEviction bool `json:"error_on_eviction,omitempty"`
+	// ErrorOnTermination indicates that the ProwJob should be completed and given
+	// the ErrorState status if the pod that is executing the job is terminated.
+	// If this field is unspecified or false, a new pod will be created to replace
+	// the terminated one.
+	ErrorOnTermination bool `json:"error_on_termination,omitempty"`
 	// SourcePath contains the path where this job is defined
 	SourcePath string `json:"-"`
 	// Spec is the Kubernetes pod spec used if Agent is kubernetes.

pkg/pjutil/pjutil.go

@@ -228,12 +228,13 @@ func specFromJobBase(jb config.JobBase) prowapi.ProwJobSpec {
 		namespace = *jb.Namespace
 	}
 	return prowapi.ProwJobSpec{
-		Job:             jb.Name,
-		Agent:           prowapi.ProwJobAgent(jb.Agent),
-		Cluster:         jb.Cluster,
-		Namespace:       namespace,
-		MaxConcurrency:  jb.MaxConcurrency,
-		ErrorOnEviction: jb.ErrorOnEviction,
+		Job:                jb.Name,
+		Agent:              prowapi.ProwJobAgent(jb.Agent),
+		Cluster:            jb.Cluster,
+		Namespace:          namespace,
+		MaxConcurrency:     jb.MaxConcurrency,
+		ErrorOnEviction:    jb.ErrorOnEviction,
+		ErrorOnTermination: jb.ErrorOnTermination,
 
 		ExtraRefs:        DecorateExtraRefs(jb.ExtraRefs, jb),
 		DecorationConfig: jb.DecorationConfig,

pkg/plank/controller_test.go

@@ -1167,6 +1167,102 @@ func TestSyncPendingJob(t *testing.T) {
 			ExpectedNumPods:  1,
 			ExpectedURL:      "boop-42/error",
 		},
+		{
+			Name: "delete terminated pod",
+			PJ: prowapi.ProwJob{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "boop-42",
+					Namespace: "prowjobs",
+				},
+				Spec: prowapi.ProwJobSpec{
+					PodSpec: &v1.PodSpec{Containers: []v1.Container{{Name: "test-name", Env: []v1.EnvVar{}}}},
+				},
+				Status: prowapi.ProwJobStatus{
+					State:   prowapi.PendingState,
+					PodName: "boop-42",
+				},
+			},
+			Pods: []v1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "boop-42",
+						Namespace: "pods",
+					},
+					Status: v1.PodStatus{
+						Phase:  v1.PodFailed,
+						Reason: Terminated,
+					},
+				},
+			},
+			ExpectedComplete: false,
+			ExpectedState:    prowapi.PendingState,
+			ExpectedNumPods:  0,
+		},
+		{
+			Name: "delete terminated pod and remove its k8sreporter finalizer",
+			PJ: prowapi.ProwJob{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "boop-42",
+					Namespace: "prowjobs",
+				},
+				Spec: prowapi.ProwJobSpec{
+					PodSpec: &v1.PodSpec{Containers: []v1.Container{{Name: "test-name", Env: []v1.EnvVar{}}}},
+				},
+				Status: prowapi.ProwJobStatus{
+					State:   prowapi.PendingState,
+					PodName: "boop-42",
+				},
+			},
+			Pods: []v1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:       "boop-42",
+						Namespace:  "pods",
+						Finalizers: []string{"prow.x-k8s.io/gcsk8sreporter"},
+					},
+					Status: v1.PodStatus{
+						Phase:  v1.PodFailed,
+						Reason: Terminated,
+					},
+				},
+			},
+			ExpectedComplete: false,
+			ExpectedState:    prowapi.PendingState,
+			ExpectedNumPods:  0,
+		},
+		{
+			Name: "don't delete terminated pod w/ error_on_termination, complete PJ instead",
+			PJ: prowapi.ProwJob{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "boop-42",
+					Namespace: "prowjobs",
+				},
+				Spec: prowapi.ProwJobSpec{
+					ErrorOnTermination: true,
+					PodSpec:            &v1.PodSpec{Containers: []v1.Container{{Name: "test-name", Env: []v1.EnvVar{}}}},
+				},
+				Status: prowapi.ProwJobStatus{
+					State:   prowapi.PendingState,
+					PodName: "boop-42",
+				},
+			},
+			Pods: []v1.Pod{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Name:      "boop-42",
+						Namespace: "pods",
+					},
+					Status: v1.PodStatus{
+						Phase:  v1.PodFailed,
+						Reason: Terminated,
+					},
+				},
+			},
+			ExpectedComplete: true,
+			ExpectedState:    prowapi.ErrorState,
+			ExpectedNumPods:  1,
+			ExpectedURL:      "boop-42/error",
+		},
 		{
 			Name: "running pod",
 			PJ: prowapi.ProwJob{

pkg/plank/reconciler.go

@@ -61,7 +61,8 @@ const ControllerName = "plank"
 
 // PodStatus constants
 const (
-	Evicted = "Evicted"
+	Evicted    = "Evicted"
+	Terminated = "Terminated"
 )
 
 // NodeStatus constants
@@ -480,6 +481,34 @@ func (r *reconciler) syncPendingJob(ctx context.Context, pj *prowv1.ProwJob) (*r
 			r.log.WithField("name", pj.ObjectMeta.Name).Debug("Delete Pod.")
 			return nil, ctrlruntimeclient.IgnoreNotFound(client.Delete(ctx, pod))
 		}
+	} else if pod.Status.Reason == Terminated {
+		// Pod was terminated.
+		if pj.Spec.ErrorOnTermination {
+			// ErrorOnTermination is enabled, complete the PJ and mark it as
+			// errored.
+			r.log.WithField("error-on-termination", true).WithFields(pjutil.ProwJobFields(pj)).Info("Pods Node got terminated, fail job.")
+			pj.SetComplete()
+			pj.Status.State = prowv1.ErrorState
+			pj.Status.Description = "Job pod's node was terminated."
+		} else {
+			// ErrorOnTermination is disabled. Delete the pod now and recreate it in
+			// the next resync.
+			r.log.WithFields(pjutil.ProwJobFields(pj)).Info("Pods Node got terminated, deleting & next sync loop will restart pod")
+			client, ok := r.buildClients[pj.ClusterAlias()]
+			if !ok {
+				return nil, TerminalError(fmt.Errorf("terminated pod %s: unknown cluster alias %q", pod.Name, pj.ClusterAlias()))
+			}
+			if finalizers := sets.New[string](pod.Finalizers...); finalizers.Has(kubernetesreporterapi.FinalizerName) {
+				// We want the end user to not see this, so we have to remove the finalizer, otherwise the pod hangs
+				oldPod := pod.DeepCopy()
+				pod.Finalizers = finalizers.Delete(kubernetesreporterapi.FinalizerName).UnsortedList()
+				if err := client.Patch(ctx, pod, ctrlruntimeclient.MergeFrom(oldPod)); err != nil {
+					return nil, fmt.Errorf("failed to patch pod trying to remove %s finalizer: %w", kubernetesreporterapi.FinalizerName, err)
+				}
+			}
+			r.log.WithField("name", pj.ObjectMeta.Name).Debug("Delete Pod.")
+			return nil, ctrlruntimeclient.IgnoreNotFound(client.Delete(ctx, pod))
+		}
 	} else if pod.DeletionTimestamp != nil && pod.Status.Reason == NodeUnreachablePodReason {
 		// This can happen in any phase and means the node got evicted after it became unresponsive. Delete the finalizer so the pod
 		// vanishes and we will silently re-create it in the next iteration.