Check for no nodes during workflow compilation

pkg/compiler/errors/compiler_error_test.go

@@ -32,6 +32,7 @@ func TestErrorCodes(t *testing.T) {
 		UnreachableNodes:           NewUnreachableNodesErr("", ""),
 		UnrecognizedValue:          NewUnrecognizedValueErr("", ""),
 		WorkflowBuildError:         NewWorkflowBuildError(errors.New("")),
+		NoNodesFound:               NewNoNodesFoundErr(""),
 	}
 
 	for key, value := range testCases {
@@ -47,6 +48,6 @@ func TestIncludeSource(t *testing.T) {
 
 	SetConfig(Config{IncludeSource: true})
 	e = NewCycleDetectedInWorkflowErr("", "")
-	assert.Equal(t, e.source, "compiler_error_test.go:49")
+	assert.Equal(t, e.source, "compiler_error_test.go:50")
 	SetConfig(Config{})
 }

pkg/compiler/errors/compiler_errors.go

@@ -72,6 +72,9 @@ const (
 
 	// A value isn't on the right syntax.
 	SyntaxError ErrorCode = "SyntaxError"
+
+	// A workflow is missing any nodes to execute
+	NoNodesFound ErrorCode = "NoNodesFound"
 )
 
 func NewBranchNodeNotSpecified(branchNodeID string) *CompileError {
@@ -246,6 +249,14 @@ func NewSyntaxError(nodeID string, element string, err error) *CompileError {
 	)
 }
 
+func NewNoNodesFoundErr(graphID string) *CompileError {
+	return newError(
+		NoNodesFound,
+		fmt.Sprintf("Can't find any nodes in workflow [%v].", graphID),
+		graphID,
+	)
+}
+
 func newError(code ErrorCode, description, nodeID string) (err *CompileError) {
 	err = &CompileError{
 		code:        code,

pkg/compiler/workflow_compiler.go

@@ -148,6 +148,11 @@ func (w workflowBuilder) AddEdges(n c.NodeBuilder, errs errors.CompileErrors) (o
 
 // Contains the main validation logic for the coreWorkflow. If successful, it'll build an executable Workflow.
 func (w workflowBuilder) ValidateWorkflow(fg *flyteWorkflow, errs errors.CompileErrors) (c.Workflow, bool) {
+	if len(fg.Template.Nodes) == 0 {
+		errs.Collect(errors.NewNoNodesFoundErr(fg.Template.Id.String()))
+		return nil, !errs.HasErrors()
+	}
+
 	// Initialize workflow
 	wf := w.newWorkflowBuilder(fg)
 	wf.updateRequiredReferences()

pkg/compiler/workflow_compiler_test.go

@@ -631,6 +631,30 @@ func TestCompileWorkflow(t *testing.T) {
 	}
 }
 
+func TestNoNodesFound(t *testing.T) {
+	inputWorkflow := &core.WorkflowTemplate{
+		Id: &core.Identifier{Name: "repo"},
+		Interface: &core.TypedInterface{
+			Inputs: createVariableMap(map[string]*core.Variable{
+				"x": {
+					Type: getIntegerLiteralType(),
+				},
+			}),
+			Outputs: createVariableMap(map[string]*core.Variable{
+				"x": {
+					Type: getIntegerLiteralType(),
+				},
+			}),
+		},
+		Nodes:   []*core.Node{},
+		Outputs: []*core.Binding{newVarBinding("node_456", "x", "x")},
+	}
+
+	_, errs := CompileWorkflow(inputWorkflow, []*core.WorkflowTemplate{},
+		mustCompileTasks(make([]*core.TaskTemplate, 0)), []common.InterfaceProvider{})
+	assert.Contains(t, errs.Error(), errors.NoNodesFound)
+}
+
 func mustCompileTasks(tasks []*core.TaskTemplate) []*core.CompiledTask {
 	res := make([]*core.CompiledTask, 0, len(tasks))
 	for _, t := range tasks {