supporting external deletion of non-terminal subtasks (flyteorg#278)

Signed-off-by: Daniel Rammer <[email protected]>

flyteplugins/go/tasks/plugins/array/k8s/subtask.go

@@ -309,6 +309,15 @@ func getSubtaskPhaseInfo(ctx context.Context, stCtx SubTaskExecutionContext, cfg
 		return pluginsCore.PhaseInfoUndefined, err
 	}
 
+	if !phaseInfo.Phase().IsTerminal() && o.GetDeletionTimestamp() != nil {
+		// If the object has been deleted, that is, it has a deletion timestamp, but is not in a terminal state, we should
+		// mark the task as a retryable failure.  We've seen this happen when a kubelet disappears - all pods running on
+		// the node are marked with a deletionTimestamp, but our finalizers prevent the pod from being deleted.
+		// This can also happen when a user deletes a Pod directly.
+		failureReason := fmt.Sprintf("object [%s] terminated in the background, manually", nsName.String())
+		return pluginsCore.PhaseInfoSystemRetryableFailure("UnexpectedObjectDeletion", failureReason, nil), nil
+	}
+
 	return phaseInfo, err
 }