Deep check on security context (flyteorg#397)

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -456,9 +456,14 @@ func mergeIntoExecConfig(workflowExecConfig *admin.WorkflowExecutionConfig, spec
 		workflowExecConfig.MaxParallelism = spec.GetMaxParallelism()
 		isChanged = true
 	}
+
 	if workflowExecConfig.GetSecurityContext() == nil && spec.GetSecurityContext() != nil {
-		workflowExecConfig.SecurityContext = spec.GetSecurityContext()
-		isChanged = true
+		if spec.GetSecurityContext().GetRunAs() != nil &&
+			(len(spec.GetSecurityContext().GetRunAs().GetK8SServiceAccount()) > 0 ||
+				len(spec.GetSecurityContext().GetRunAs().GetIamRole()) > 0) {
+			workflowExecConfig.SecurityContext = spec.GetSecurityContext()
+			isChanged = true
+		}
 	}
 	// Launchplan spec has label, annotation and rawOutputDataConfig initialized with empty values.
 	// Hence we do a deep check in the following conditions before assignment

flyteadmin/pkg/manager/impl/execution_manager_test.go

@@ -3814,6 +3814,39 @@ func TestGetExecutionConfigOverrides(t *testing.T) {
 		assert.Equal(t, requestOutputLocationPrefix, execConfig.RawOutputDataConfig.OutputLocationPrefix)
 		assert.Equal(t, requestLabels, execConfig.GetLabels().Values)
 	})
+	t.Run("request with empty security context", func(t *testing.T) {
+		request := &admin.ExecutionCreateRequest{
+			Project: workflowIdentifier.Project,
+			Domain:  workflowIdentifier.Domain,
+			Spec: &admin.ExecutionSpec{
+				SecurityContext: &core.SecurityContext{
+					RunAs: &core.Identity{
+						K8SServiceAccount: "",
+						IamRole:           "",
+					},
+				},
+			},
+		}
+		launchPlan := &admin.LaunchPlan{
+			Spec: &admin.LaunchPlanSpec{
+				Annotations:         &admin.Annotations{Values: launchPlanAnnotations},
+				Labels:              &admin.Labels{Values: launchPlanLabels},
+				RawOutputDataConfig: &admin.RawOutputDataConfig{OutputLocationPrefix: launchPlanOutputLocationPrefix},
+				SecurityContext: &core.SecurityContext{
+					RunAs: &core.Identity{
+						K8SServiceAccount: launchPlanK8sServiceAccount,
+					},
+				},
+				MaxParallelism: launchPlanMaxParallelism,
+			},
+		}
+		execConfig, err := executionManager.getExecutionConfig(context.TODO(), request, launchPlan)
+		assert.NoError(t, err)
+		assert.Equal(t, launchPlanMaxParallelism, execConfig.MaxParallelism)
+		assert.Equal(t, launchPlanK8sServiceAccount, execConfig.SecurityContext.RunAs.K8SServiceAccount)
+		assert.Equal(t, launchPlanOutputLocationPrefix, execConfig.RawOutputDataConfig.OutputLocationPrefix)
+		assert.Equal(t, launchPlanLabels, execConfig.GetLabels().Values)
+	})
 	t.Run("request with no config", func(t *testing.T) {
 		request := &admin.ExecutionCreateRequest{
 			Project: workflowIdentifier.Project,