Return InvalidArgument for workflow compilation failures in CreateWor…

…kflow

Signed-off-by: Katrina Rogan <[email protected]>

flyteadmin/pkg/manager/impl/workflow_manager.go

@@ -146,7 +146,7 @@ func (w *WorkflowManager) CreateWorkflow(
 	workflowClosure, err := w.getCompiledWorkflow(ctx, finalizedRequest)
 	if err != nil {
 		logger.Errorf(ctx, "Failed to compile workflow with err: %v", err)
-		return nil, errors.NewFlyteAdminErrorf(codes.Internal,
+		return nil, errors.NewFlyteAdminErrorf(codes.InvalidArgument,
 			"failed to compile workflow for [%+v] with err %v", request.Id, err)
 	}
 	err = validation.ValidateCompiledWorkflow(

flyteadmin/pkg/manager/impl/workflow_manager_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"google.golang.org/grpc/status"
 	"testing"
 
 	"github.com/golang/protobuf/proto"
@@ -243,10 +244,13 @@ func TestCreateWorkflow_CompileWorkflowError(t *testing.T) {
 		getMockWorkflowConfigProvider(), mockCompiler, getMockStorage(), storagePrefix, mockScope.NewTestScope())
 	request := testutils.GetWorkflowRequest()
 	response, err := workflowManager.CreateWorkflow(context.Background(), request)
+	assert.Nil(t, response)
+	s, ok := status.FromError(err)
+	assert.True(t, ok)
+	assert.Equal(t, codes.InvalidArgument, s.Code())
 	assert.EqualError(t, err, fmt.Sprintf(
 		"failed to compile workflow for [resource_type:WORKFLOW project:\"project\" domain:\"domain\" "+
 			"name:\"name\" version:\"version\" ] with err %v", expectedErr.Error()))
-	assert.Nil(t, response)
 }
 
 func TestCreateWorkflow_DatabaseError(t *testing.T) {