Reporting running if the primary container status is not yet reported

flyteplugins/go/tasks/pluginmachinery/flytek8s/pod_helper.go

@@ -24,6 +24,7 @@ import (
 const PodKind = "pod"
 const OOMKilled = "OOMKilled"
 const Interrupted = "Interrupted"
+const PrimaryContainerNotFound = "PrimaryContainerNotFound"
 const SIGKILL = 137
 
 const defaultContainerTemplateName = "default"
@@ -746,7 +747,7 @@ func DeterminePrimaryContainerPhase(primaryContainerName string, statuses []v1.C
 	}
 
 	// If for some reason we can't find the primary container, always just return a permanent failure
-	return pluginsCore.PhaseInfoFailure("PrimaryContainerMissing",
+	return pluginsCore.PhaseInfoFailure(PrimaryContainerNotFound,
 		fmt.Sprintf("Primary container [%s] not found in pod's container statuses", primaryContainerName), info)
 }
 

flyteplugins/go/tasks/pluginmachinery/flytek8s/pod_helper_test.go

@@ -1682,6 +1682,7 @@ func TestDeterminePrimaryContainerPhase(t *testing.T) {
 			secondaryContainer,
 		}, info)
 		assert.Equal(t, pluginsCore.PhasePermanentFailure, phaseInfo.Phase())
+		assert.Equal(t, PrimaryContainerNotFound, phaseInfo.Err().Code)
 		assert.Equal(t, "Primary container [primary] not found in pod's container statuses", phaseInfo.Err().Message)
 	})
 }

flyteplugins/go/tasks/plugins/k8s/pod/plugin.go

@@ -211,7 +211,17 @@ func (plugin) GetTaskPhaseWithLogs(ctx context.Context, pluginContext k8s.Plugin
 		} else {
 			// if the primary container annotation exists, we use the status of the specified container
 			phaseInfo = flytek8s.DeterminePrimaryContainerPhase(primaryContainerName, pod.Status.ContainerStatuses, &info)
-			if phaseInfo.Phase() == pluginsCore.PhaseRunning && len(info.Logs) > 0 {
+			if phaseInfo.Phase() == pluginsCore.PhasePermanentFailure && phaseInfo.Err() != nil && 
+				phaseInfo.Err().GetCode() == flytek8s.PrimaryContainerNotFound {
+				// if the primary container status is not found ensure that the primary container exists.
+				// note: it should be impossible for the primary container to not exist at this point.
+				for _, container := range pod.Spec.Containers {
+					if container.Name == primaryContainerName {
+						phaseInfo = pluginsCore.PhaseInfoRunning(pluginsCore.DefaultPhaseVersion, &info)
+						break
+					}
+				}
+			} else if phaseInfo.Phase() == pluginsCore.PhaseRunning && len(info.Logs) > 0 {
 				phaseInfo = phaseInfo.WithVersion(pluginsCore.DefaultPhaseVersion + 1)
 			}
 		}

flyteplugins/go/tasks/plugins/k8s/pod/sidecar_test.go

@@ -844,6 +844,13 @@ func TestDemystifiedSidecarStatus_PrimaryRunning(t *testing.T) {
 
 func TestDemystifiedSidecarStatus_PrimaryMissing(t *testing.T) {
 	res := &v1.Pod{
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				{
+					Name: "Secondary",
+				},
+			},
+		},
 		Status: v1.PodStatus{
 			Phase: v1.PodRunning,
 			ContainerStatuses: []v1.ContainerStatus{
@@ -862,6 +869,34 @@ func TestDemystifiedSidecarStatus_PrimaryMissing(t *testing.T) {
 	assert.Equal(t, pluginsCore.PhasePermanentFailure, phaseInfo.Phase())
 }
 
+// TODO @hamersaw
+func TestDemystifiedSidecarStatus_PrimaryNotExistsYet(t *testing.T) {
+	res := &v1.Pod{
+		Spec: v1.PodSpec{
+			Containers: []v1.Container{
+				{
+					Name: "Primary",
+				},
+			},
+		},
+		Status: v1.PodStatus{
+			Phase: v1.PodRunning,
+			ContainerStatuses: []v1.ContainerStatus{
+				{
+					Name: "Secondary",
+				},
+			},
+		},
+	}
+	res.SetAnnotations(map[string]string{
+		flytek8s.PrimaryContainerKey: "Primary",
+	})
+	taskCtx := getDummySidecarTaskContext(&core.TaskTemplate{}, sidecarResourceRequirements, nil)
+	phaseInfo, err := DefaultPodPlugin.GetTaskPhase(context.TODO(), taskCtx, res)
+	assert.Nil(t, err)
+	assert.Equal(t, pluginsCore.PhaseRunning, phaseInfo.Phase())
+}
+
 func TestGetProperties(t *testing.T) {
 	expected := k8s.PluginProperties{}
 	assert.Equal(t, expected, DefaultPodPlugin.GetProperties())