PR #443: Add support to a Validation phase to the task lifecycle.

It is better if we can detect and report configuration errors of task
pipeline early, during task definition, rather than deferring such
checks to runtime (when this is possible).

To support such checks, we add a Validate() error method to
the pipeline.Node() interface which gives pipeline nodes a chance
to validate their configuration. Any pipeline containing nodes
that do not pass these test cannot be saved.

Signed-off-by: Jon Seymour <[email protected]>

pipeline/node.go

@@ -61,6 +61,9 @@ type Node interface {
 	// The type of output the node provides.
 	Provides() EdgeType
 
+	// Check that the definition of the node is consistent
+	validate() error
+
 	// Helper methods for walking DAG
 	tMark() bool
 	setTMark(b bool)
@@ -167,6 +170,10 @@ func (n *node) Provides() EdgeType {
 	return n.provides
 }
 
+func (n *node) validate() error {
+	return nil
+}
+
 func (n *node) dot(buf *bytes.Buffer) {
 	for _, c := range n.children {
 		buf.Write([]byte(fmt.Sprintf("%s -> %s;\n", n.Name(), c.Name())))

pipeline/pipeline.go

@@ -60,6 +60,12 @@ func CreatePipeline(script string, sourceEdge EdgeType, scope *tick.Scope, deadm
 			return nil, fmt.Errorf("source edge type must be either Stream or Batch not %s", sourceEdge)
 		}
 	}
+	if err = p.Walk(
+		func(n Node) error {
+			return n.validate()
+		}); err != nil {
+		return nil, err
+	}
 	return p, nil
 }