[WIP] Adding methods to capture inter-step communication

README.md

@@ -32,15 +32,23 @@ class: sifter
 name: census_2010
 
 params:
-  census: ../data/census_2010_byzip.json
-  date: "2010-01-01"
-  schema: ../covid19_datadictionary/gdcdictionary/schemas/
+  census: 
+    type: file
+    default: ../data/census_2010_byzip.json
+  date: 
+    default: "2010-01-01"
+  schema: 
+    type: path
+    default: ../covid19_datadictionary/gdcdictionary/schemas/
 
 inputs:
   censusData:
     jsonLoad:
       input: "{{params.census}}"
 
+outputs:
+
+
 pipelines:
   transform:
     - from: censusData

cmd/run/main.go

@@ -14,6 +14,8 @@ var outDir string = ""
 var paramsFile string = ""
 var verbose bool = false
 var cmdParams map[string]string
+var captureDir string = ""
+var captureLimit int = 10
 
 // Cmd is the declaration of the command line
 var Cmd = &cobra.Command{
@@ -46,11 +48,11 @@ var Cmd = &cobra.Command{
 				}
 				pb := playbook.Playbook{}
 				playbook.ParseBytes(yaml, "./playbook.yaml", &pb)
-				if err := Execute(pb, "./", "./", outDir, params); err != nil {
+				if err := Execute(pb, "./", "./", outDir, params, captureDir, captureLimit); err != nil {
 					return err
 				}
 			} else {
-				if err := ExecuteFile(playFile, "./", outDir, params); err != nil {
+				if err := ExecuteFile(playFile, "./", outDir, params, captureDir, captureLimit); err != nil {
 					return err
 				}
 			}
@@ -65,4 +67,7 @@ func init() {
 	flags.BoolVarP(&verbose, "verbose", "v", verbose, "Verbose logging")
 	flags.StringToStringVarP(&cmdParams, "param", "p", cmdParams, "Parameter variable")
 	flags.StringVarP(&paramsFile, "params-file", "f", paramsFile, "Parameter file")
+	flags.StringVarP(&captureDir, "capture-dir", "d", "", "Directory for capture files (default: None)")
+	flags.IntVarP(&captureLimit, "capture-limit", "l", 10, "Max records to capture per step (0 = unlimited)")
+	flags.StringVarP(&outDir, "output", "o", outDir, "Output directory for playbook results (default: current directory or value specified in playbook)")
 }

cmd/run/run.go

@@ -1,6 +1,7 @@
 package run
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
 
@@ -9,7 +10,7 @@ import (
 	"github.com/bmeg/sifter/task"
 )
 
-func ExecuteFile(playFile string, workDir string, outDir string, inputs map[string]string) error {
+func ExecuteFile(playFile string, workDir string, outDir string, inputs map[string]string, debugDir string, debugLimit int) error {
 	logger.Info("Starting", "playFile", playFile)
 	pb := playbook.Playbook{}
 	if err := playbook.ParseFile(playFile, &pb); err != nil {
@@ -19,10 +20,10 @@ func ExecuteFile(playFile string, workDir string, outDir string, inputs map[stri
 	a, _ := filepath.Abs(playFile)
 	baseDir := filepath.Dir(a)
 	logger.Debug("parsed file", "baseDir", baseDir, "playbook", pb)
-	return Execute(pb, baseDir, workDir, outDir, inputs)
+	return Execute(pb, baseDir, workDir, outDir, inputs, debugDir, debugLimit)
 }
 
-func Execute(pb playbook.Playbook, baseDir string, workDir string, outDir string, params map[string]string) error {
+func Execute(pb playbook.Playbook, baseDir string, workDir string, outDir string, params map[string]string, debugDir string, debugLimit int) error {
 
 	if outDir == "" {
 		outDir = pb.GetDefaultOutDir()
@@ -32,13 +33,39 @@ func Execute(pb playbook.Playbook, baseDir string, workDir string, outDir string
 		os.MkdirAll(outDir, 0777)
 	}
 
+	// Setup debug capture directory if enabled
+	// Enable if: user explicitly set dir, OR user changed limit from default
+	enableDebug := debugDir != "" || (debugLimit != 10)
+	if enableDebug {
+		if debugDir == "" {
+			debugDir = filepath.Join(workDir, "debug-capture")
+		} else if !filepath.IsAbs(debugDir) {
+			debugDir = filepath.Join(workDir, debugDir)
+		}
+		if info, err := os.Stat(debugDir); err != nil {
+			if os.IsNotExist(err) {
+				if mkErr := os.MkdirAll(debugDir, 0777); mkErr != nil {
+					logger.Error("Failed to create debug directory", "error", mkErr)
+					return mkErr
+				}
+			} else {
+				logger.Error("Failed to access debug directory", "path", debugDir, "error", err)
+				return err
+			}
+		} else if !info.IsDir() {
+			logger.Error("Debug path exists but is not a directory", "path", debugDir)
+			return fmt.Errorf("debug path %s exists but is not a directory", debugDir)
+		}
+		logger.Info("Debug capture enabled", "dir", debugDir, "limit", debugLimit)
+	}
+
 	nInputs, err := pb.PrepConfig(params, workDir)
 	if err != nil {
 		return err
 	}
 	logger.Debug("Running", "outDir", outDir)
 
 	t := task.NewTask(pb.Name, baseDir, workDir, outDir, nInputs)
-	err = pb.Execute(t)
+	err = pb.ExecuteWithCapture(t, debugDir, debugLimit)
 	return err
 }

extractors/interface.go

@@ -44,6 +44,20 @@ func (ex *Extractor) Start(t task.RuntimeTask) (chan map[string]interface{}, err
 	return nil, fmt.Errorf(("Extractor not defined"))
 }
 
+func (ex *Extractor) GetType() reflect.Type {
+	v := reflect.ValueOf(ex).Elem()
+	for i := 0; i < v.NumField(); i++ {
+		f := v.Field(i)
+		x := f.Interface()
+		if _, ok := x.(Source); ok {
+			if !f.IsNil() {
+				return f.Type()
+			}
+		}
+	}
+	return nil
+}
+
 func (ex *Extractor) GetRequiredParams() []config.ParamRequest {
 	out := []config.ParamRequest{}
 	v := reflect.ValueOf(ex).Elem()