handle task validation error wip

Signed-off-by: Iaroslav Ciupin <[email protected]>

flyteadmin/pkg/manager/impl/execution_manager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/benbjohnson/clock"
@@ -862,10 +863,14 @@ func (m *ExecutionManager) launchExecutionAndPrepareModel(
 	})
 
 	if err != nil {
-		m.systemMetrics.PropellerFailures.Inc()
-		logger.Infof(ctx, "Failed to execute workflow %+v with execution id %+v and inputs %+v with err %v",
-			request, workflowExecutionID, executionInputs, err)
-		return nil, nil, err
+		if strings.Contains(err.Error(), "no node can run task") {
+			logger.Infof(ctx, "received task validation error, saving execution [%+v] as failed", workflowExecutionID)
+		} else {
+			m.systemMetrics.PropellerFailures.Inc()
+			logger.Infof(ctx, "Failed to execute workflow %+v with execution id %+v and inputs %+v with err %v",
+				request, workflowExecutionID, executionInputs, err)
+			return nil, nil, err
+		}
 	}
 	executionCreatedAt := time.Now()
 	acceptanceDelay := executionCreatedAt.Sub(requestedAt)
@@ -902,6 +907,7 @@ func (m *ExecutionManager) launchExecutionAndPrepareModel(
 		SecurityContext:       executionConfig.SecurityContext,
 		LaunchEntity:          launchPlan.Id.ResourceType,
 		Namespace:             namespace,
+		Error:                 err,
 	})
 	if err != nil {
 		logger.Infof(ctx, "Failed to create execution model in transformer for id: [%+v] with err: %v",

flyteadmin/pkg/repositories/transformers/execution.go

@@ -45,6 +45,7 @@ type CreateExecutionModelInput struct {
 	SecurityContext       *core.SecurityContext
 	LaunchEntity          core.ResourceType
 	Namespace             string
+	Error                 error
 }
 
 type ExecutionTransformerOptions struct {
@@ -91,6 +92,16 @@ func CreateExecutionModel(input CreateExecutionModelInput) (*models.Execution, e
 	if input.Phase == core.WorkflowExecution_RUNNING {
 		closure.StartedAt = createdAt
 	}
+	if input.Error != nil {
+		closure.Phase = core.WorkflowExecution_FAILED
+		closure.OutputResult = &admin.ExecutionClosure_Error{
+			Error: &core.ExecutionError{
+				Code:    "TaskValidationFailed",
+				Message: input.Error.Error(),
+				Kind:    core.ExecutionError_USER,
+			},
+		}
+	}
 
 	closureBytes, err := proto.Marshal(&closure)