feat: receive primary container name in K8SPod

Signed-off-by: machichima <[email protected]>

flyteplugins/go/tasks/plugins/k8s/spark/spark.go

@@ -208,11 +208,9 @@ func createDriverSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCont
 				"Unable to unmarshal driver pod spec [%v], Err: [%v]", driverPod.GetPodSpec(), err.Error())
 		}
 
-		if len(customPodSpec.Containers) != 0 {
-			// As only the container with the name of primary container set in driver pod will be passed into
-			// here, we can directly take the container name here as the primaryContainerName
-			primaryContainerName = customPodSpec.Containers[0].Name
-		}
+        if driverPod.GetPrimaryContainerName() != "" {
+            primaryContainerName = driverPod.PrimaryContainerName
+        }
 
 		podSpec, err = flytek8s.MergePodSpecs(podSpec, customPodSpec, primaryContainerName, "")
 		if err != nil {
@@ -261,11 +259,9 @@ func createExecutorSpec(ctx context.Context, taskCtx pluginsCore.TaskExecutionCo
 				"Unable to unmarshal executor pod spec [%v], Err: [%v]", executorPod.GetPodSpec(), err.Error())
 		}
 
-		if len(customPodSpec.Containers) != 0 {
-			// As only the container with the name of primary container set in executor pod will be passed into
-			// here, we can directly take the container name here as the primaryContainerName
-			primaryContainerName = customPodSpec.Containers[0].Name
-		}
+        if executorPod.GetPrimaryContainerName() != "" {
+            primaryContainerName = executorPod.PrimaryContainerName
+        }
 
 		podSpec, err = flytek8s.MergePodSpecs(podSpec, customPodSpec, primaryContainerName, "")
 		if err != nil {