Repository: GoogleCloudPlatform/spark-on-k8s-operator
Pull request #1637 (2022-11-01 16:09:31)

Add Kognita to "Who is using"

google-cla[bot]: Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this [failed invocation](https://github.com/GoogleCloudPlatform/spark-on-k8s-operator/pull/1637/checks?check_run_id=9229727318) of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

Patch:
diff --git a/docs/user-guide.md b/docs/user-guide.md
index 3cc65ef..1aeeb12 100644
--- a/docs/user-guide.md
+++ b/docs/user-guide.md
@@ -614,7 +614,7 @@ spec:
 ```
 
 ### Using Container LifeCycle Hooks
-A Spark Application can optionally specify a [Container Lifecycle Hooks](https://kubernetes.io/docs/concepts/containers/container-lifecycle-hooks/#container-hooks) for a driver. It is useful in cases where you need a PreStop or PostStart hooks to driver.
+A Spark Application can optionally specify a [Container Lifecycle Hooks](https://kubernetes.io/docs/concepts/containers/container-lifecycle-hooks/#container-hooks) for a driver. It is useful in cases where you need a PreStop or PostStart hooks to driver and executor.
 
 ```yaml
 spec:
diff --git a/docs/who-is-using.md b/docs/who-is-using.md
index 2a6167e..76a609d 100644
--- a/docs/who-is-using.md
+++ b/docs/who-is-using.md
@@ -39,3 +39,4 @@
 | [DiDi](https://www.didiglobal.com) | @Run-Lin | Evaluation | Data Infrastructure |
 | [DeepCure](https://www.deepcure.ai) | @mschroering | Production | Spark / ML |
 | [C2FO](https://www.c2fo.com/) | @vanhoale | Production | Data Platform / Data Infrastructure |
+| [Kognita](https://kognita.com.br/) | @andreclaudino | Production | MLOps, Data Platform / Data Infrastructure, ML/AI |
diff --git a/main.go b/main.go
index 2bb5496..db4861b 100644
--- a/main.go
+++ b/main.go
@@ -219,7 +219,11 @@ func main() {
 
 	if *enableLeaderElection {
 		glog.Info("Waiting to be elected leader before starting application controller goroutines")
-		<-startCh
+		select {
+		case <-signalCh:
+			os.Exit(0)
+		case <-startCh:
+		}
 	}
 
 	glog.Info("Starting application controller goroutines")
diff --git a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
index bd4aee2..2c89fe9 100644
--- a/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
+++ b/pkg/apis/sparkoperator.k8s.io/v1beta2/types.go
@@ -580,6 +580,9 @@ type ExecutorSpec struct {
 	// GC settings or other logging.
 	// +optional
 	JavaOptions *string `json:"javaOptions,omitempty"`
+	// Lifecycle for running preStop or postStart commands
+	// +optional
+	Lifecycle *apiv1.Lifecycle `json:"lifecycle,omitempty"`
 	// DeleteOnTermination specify whether executor pods should be deleted in case of failure or normal termination.
 	// Maps to `spark.kubernetes.executor.deleteOnTermination` that is available since Spark 3.0.
 	// +optional
diff --git a/pkg/webhook/patch.go b/pkg/webhook/patch.go
index 3034ebd..bc9a748 100644
--- a/pkg/webhook/patch.go
+++ b/pkg/webhook/patch.go
@@ -765,8 +765,13 @@ func addTerminationGracePeriodSeconds(pod *corev1.Pod, app *v1beta2.SparkApplica
 
 func addPodLifeCycleConfig(pod *corev1.Pod, app *v1beta2.SparkApplication) *patchOperation {
 	var lifeCycle *corev1.Lifecycle
+	var containerName string
 	if util.IsDriverPod(pod) {
 		lifeCycle = app.Spec.Driver.Lifecycle
+		containerName = config.SparkDriverContainerName
+	} else if util.IsExecutorPod(pod) {
+		lifeCycle = app.Spec.Executor.Lifecycle
+		containerName = config.SparkExecutorContainerName
 	}
 	if lifeCycle == nil {
 		return nil
@@ -775,12 +780,12 @@ func addPodLifeCycleConfig(pod *corev1.Pod, app *v1beta2.SparkApplication) *patc
 	i := 0
 	// Find the driver container in the pod.
 	for ; i < len(pod.Spec.Containers); i++ {
-		if pod.Spec.Containers[i].Name == config.SparkDriverContainerName {
+		if pod.Spec.Containers[i].Name == containerName {
 			break
 		}
 	}
 	if i == len(pod.Spec.Containers) {
-		glog.Warningf("Spark driver container not found in pod %s", pod.Name)
+		glog.Warningf("Spark container %s not found in pod %s", containerName, pod.Name)
 		return nil
 	}
 


Test patch:
diff --git a/pkg/webhook/patch_test.go b/pkg/webhook/patch_test.go
index 9efa740..4b66b62 100644
--- a/pkg/webhook/patch_test.go
+++ b/pkg/webhook/patch_test.go
@@ -1752,6 +1752,9 @@ func TestPatchSparkPod_Lifecycle(t *testing.T) {
 	preStopTest := &corev1.ExecAction{
 		Command: []string{"/bin/sh", "-c", "echo Hello from the pre stop handler > /usr/share/message"},
 	}
+	postStartTest := &corev1.ExecAction{
+		Command: []string{"/bin/sh", "-c", "echo Hello from the post start handler > /usr/share/message"},
+	}
 	app := &v1beta2.SparkApplication{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: "spark-test",
@@ -1763,6 +1766,11 @@ func TestPatchSparkPod_Lifecycle(t *testing.T) {
 					PreStop: &corev1.LifecycleHandler{Exec: preStopTest},
 				},
 			},
+			Executor: v1beta2.ExecutorSpec{
+				Lifecycle: &corev1.Lifecycle{
+					PostStart: &corev1.LifecycleHandler{Exec: postStartTest},
+				},
+			},
 		},
 	}
 
@@ -1784,11 +1792,34 @@ func TestPatchSparkPod_Lifecycle(t *testing.T) {
 		},
 	}
 
+	executorPod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "spark-executor",
+			Labels: map[string]string{
+				config.SparkRoleLabel:               config.SparkExecutorRole,
+				config.LaunchedBySparkOperatorLabel: "true",
+			},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				{
+					Name:  config.SparkExecutorContainerName,
+					Image: "spark-executor:latest",
+				},
+			},
+		},
+	}
+
 	modifiedDriverPod, err := getModifiedPod(driverPod, app)
 	if err != nil {
 		t.Fatal(err)
 	}
+	modifiedExecutorPod, err := getModifiedPod(executorPod, app)
+	if err != nil {
+		t.Fatal(err)
+	}
 	assert.Equal(t, preStopTest, modifiedDriverPod.Spec.Containers[0].Lifecycle.PreStop.Exec)
+	assert.Equal(t, postStartTest, modifiedExecutorPod.Spec.Containers[0].Lifecycle.PostStart.Exec)
 }
 
 func getModifiedPod(pod *corev1.Pod, app *v1beta2.SparkApplication) (*corev1.Pod, error) {