Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1111 (2020-12-10 15:33:02)

update executor status if pod is lost while app is still running
Hello,

This PR is related to issue #1076

Following the discussion on this issue, it seems that we can change the status of lost pods to `unknown` while the app is still `running`
Then once the driver pod is terminated, we fall back to the current implementation 



Patch:
diff --git a/pkg/controller/sparkapplication/controller.go b/pkg/controller/sparkapplication/controller.go
index cfb30c3..5baafa9 100644
--- a/pkg/controller/sparkapplication/controller.go
+++ b/pkg/controller/sparkapplication/controller.go
@@ -395,15 +395,19 @@ func (c *Controller) getAndUpdateExecutorState(app *v1beta2.SparkApplication) er
 	// Handle missing/deleted executors.
 	for name, oldStatus := range app.Status.ExecutorState {
 		_, exists := executorStateMap[name]
-		if !isExecutorTerminated(oldStatus) && !exists && !isDriverRunning(app) {
-			// If ApplicationState is COMPLETED, in other words, the driver pod has been completed
-			// successfully. The executor pods terminate and are cleaned up, so we could not found
-			// the executor pod, under this circumstances, we assume the executor pod are completed.
-			if app.Status.AppState.State == v1beta2.CompletedState {
-				app.Status.ExecutorState[name] = v1beta2.ExecutorCompletedState
+		if !isExecutorTerminated(oldStatus) && !exists {
+			if !isDriverRunning(app) {
+				// If ApplicationState is COMPLETED, in other words, the driver pod has been completed
+				// successfully. The executor pods terminate and are cleaned up, so we could not found
+				// the executor pod, under this circumstances, we assume the executor pod are completed.
+				if app.Status.AppState.State == v1beta2.CompletedState {
+					app.Status.ExecutorState[name] = v1beta2.ExecutorCompletedState
+				} else {
+					glog.Infof("Executor pod %s not found, assuming it was deleted.", name)
+					app.Status.ExecutorState[name] = v1beta2.ExecutorFailedState
+				}
 			} else {
-				glog.Infof("Executor pod %s not found, assuming it was deleted.", name)
-				app.Status.ExecutorState[name] = v1beta2.ExecutorFailedState
+				app.Status.ExecutorState[name] = v1beta2.ExecutorUnknownState
 			}
 		}
 	}


Test patch:
diff --git a/pkg/controller/sparkapplication/controller_test.go b/pkg/controller/sparkapplication/controller_test.go
index aef6697..018217e 100644
--- a/pkg/controller/sparkapplication/controller_test.go
+++ b/pkg/controller/sparkapplication/controller_test.go
@@ -1419,6 +1419,29 @@ func TestSyncSparkApplication_ExecutingState(t *testing.T) {
 				successMetricCount: 1,
 			},
 		},
+		{
+			appName:           appName,
+			oldAppStatus:      v1beta2.RunningState,
+			oldExecutorStatus: map[string]v1beta2.ExecutorState{"exec-1": v1beta2.ExecutorRunningState},
+			driverPod: &apiv1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      driverPodName,
+					Namespace: "test",
+					Labels: map[string]string{
+						config.SparkRoleLabel:    config.SparkDriverRole,
+						config.SparkAppNameLabel: appName,
+					},
+					ResourceVersion: "1",
+				},
+				Status: apiv1.PodStatus{
+					Phase: apiv1.PodRunning,
+				},
+			},
+			expectedAppState:        v1beta2.RunningState,
+			expectedExecutorState:   map[string]v1beta2.ExecutorState{"exec-1": v1beta2.ExecutorUnknownState},
+			expectedAppMetrics:      metrics{},
+			expectedExecutorMetrics: executorMetrics{},
+		},
 	}
 
 	testFn := func(test testcase, t *testing.T) {