Code structure improvements

cmd/cmd.go → cmd/wilson/cmd.go

@@ -1,19 +1,20 @@
-package cmd
+package main
 
 import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
+	"github.com/trntv/wilson/internal/watch"
 	"github.com/trntv/wilson/pkg/runner"
 	"github.com/trntv/wilson/pkg/scheduler"
 	"github.com/trntv/wilson/pkg/task"
-	"github.com/trntv/wilson/pkg/watch"
 	"io/ioutil"
 	"os"
 	"strings"
 )
 
+// todo: remove global variables
 var debug, silent bool
 var cfg *config.Config
 

cmd/completion.go → cmd/wilson/completion.go

@@ -1,4 +1,4 @@
-package cmd
+package main
 
 import (
 	"github.com/spf13/cobra"

cmd/list.go → cmd/wilson/list.go

@@ -1,10 +1,9 @@
-package cmd
+package main
 
 import (
 	"fmt"
 	"github.com/spf13/cobra"
 	"github.com/trntv/wilson/pkg/util"
-	"log"
 	"os"
 	"text/template"
 )
@@ -33,7 +32,7 @@ func NewListCommand() *cobra.Command {
 		Use:   "list",
 		Short: "List contexts, pipelines, tasks and watchers",
 		Args:  cobra.NoArgs,
-		Run: func(cmd *cobra.Command, args []string) {
+		RunE: func(cmd *cobra.Command, args []string) error {
 			t := template.Must(template.New("list").Parse(listTmpl))
 
 			data := struct {
@@ -46,9 +45,7 @@ func NewListCommand() *cobra.Command {
 			}
 
 			err := t.Execute(os.Stdout, data)
-			if err != nil {
-				log.Println("executing template:", err)
-			}
+			return err
 		},
 	}
 

cmd/run.go → cmd/wilson/run.go

@@ -1,4 +1,4 @@
-package cmd
+package main
 
 import (
 	"fmt"

cmd/task.go → cmd/wilson/task.go

@@ -1,4 +1,4 @@
-package cmd
+package main
 
 import (
 	"fmt"

cmd/watch.go → cmd/wilson/watch.go

@@ -1,11 +1,11 @@
-package cmd
+package main
 
 import (
 	"fmt"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
+	"github.com/trntv/wilson/internal/watch"
 	"github.com/trntv/wilson/pkg/util"
-	"github.com/trntv/wilson/pkg/watch"
 	"sync"
 )
 

main.go → cmd/wilson/wilson.go

@@ -2,7 +2,6 @@ package main
 
 import (
 	"github.com/sirupsen/logrus"
-	"github.com/trntv/wilson/cmd"
 	"log"
 	"os"
 	"os/signal"
@@ -16,7 +15,7 @@ func main() {
 	})
 	listenSignals()
 
-	if err := cmd.Execute(); err != nil {
+	if err := Execute(); err != nil {
 		log.Fatalf("Command execution error: %s", err)
 	}
 }
@@ -35,7 +34,7 @@ func listenSignals() {
 				case syscall.SIGTERM:
 					exit = 143
 				}
-				cmd.Abort()
+				Abort()
 				os.Exit(exit)
 			}
 		}

pkg/config/config.go → internal/config/config.go

@@ -103,7 +103,7 @@ func Get() *Config {
 
 func Load(file string) (*Config, error) {
 	var err error
-	cfg, err = loadGlobalConfig()
+	cfg, err = LoadGlobalConfig()
 	if err != nil {
 		return nil, err
 	}
@@ -130,7 +130,7 @@ func Load(file string) (*Config, error) {
 	}
 
 	file = path.Join(dir, file)
-	localCfg, err := load(file)
+	localCfg, err := LoadFile(file)
 	if err != nil {
 		return nil, err
 	}
@@ -148,7 +148,7 @@ func Load(file string) (*Config, error) {
 	return cfg, nil
 }
 
-func loadGlobalConfig() (*Config, error) {
+func LoadGlobalConfig() (*Config, error) {
 	h, err := os.UserHomeDir()
 	if err != nil {
 		return nil, err
@@ -159,10 +159,10 @@ func loadGlobalConfig() (*Config, error) {
 		return nil, nil
 	}
 
-	return load(file)
+	return LoadFile(file)
 }
 
-func load(file string) (*Config, error) {
+func LoadFile(file string) (*Config, error) {
 	loaded[file] = true
 	config, err := readFile(file)
 	if err != nil {
@@ -176,7 +176,7 @@ func load(file string) (*Config, error) {
 			continue
 		}
 
-		lconfig, err := load(importFile)
+		lconfig, err := LoadFile(importFile)
 		if err != nil {
 			return nil, err
 		}

pkg/watch/watch.go → internal/watch/watch.go

@@ -2,7 +2,7 @@ package watch
 
 import (
 	log "github.com/sirupsen/logrus"
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/runner"
 	"github.com/trntv/wilson/pkg/task"
 	"github.com/trntv/wilson/pkg/util"

pkg/runner/context.go

@@ -2,9 +2,10 @@ package runner
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	log "github.com/sirupsen/logrus"
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/util"
 	"os"
 	"os/exec"
@@ -342,24 +343,24 @@ func (c *ExecutionContext) runServiceCommand(command string) (err error) {
 	return nil
 }
 
-func (c *ExecutionContext) createCommand(ctx context.Context, command string) *exec.Cmd {
+func (c *ExecutionContext) createCommand(ctx context.Context, command string) (*exec.Cmd, error) {
 	switch c.ctxType {
 	case config.ContextTypeLocal:
-		return c.buildLocalCommand(ctx, command)
+		return c.buildLocalCommand(ctx, command), nil
 	case config.ContextTypeContainer:
 		switch c.container.provider {
 		case config.ContextContainerProviderDocker, config.ContextContainerProviderDockerCompose:
-			return c.buildDockerCommand(ctx, command)
+			return c.buildDockerCommand(ctx, command), nil
 		case config.ContextContainerProviderKubectl:
-			return c.buildKubectlCommand(ctx, command)
+			return c.buildKubectlCommand(ctx, command), nil
 		}
 	case config.ContextTypeRemote:
-		return c.buildRemoteCommand(ctx, command)
+		return c.buildRemoteCommand(ctx, command), nil
 	default:
-		log.Fatal("unknown context type")
+		return nil, errors.New("unknown context type")
 	}
 
-	return nil
+	return nil, nil
 }
 
 func (c *ExecutionContext) ScheduleForCleanup() {

pkg/runner/context_test.go

@@ -2,7 +2,7 @@ package runner
 
 import (
 	"context"
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/util"
 	"testing"
 )
@@ -18,7 +18,7 @@ func TestContext_BuildContext(t *testing.T) {
 		Env:  map[string]string{"TEST_VAR": "TEST_VAL"},
 	}, wcfg)
 
-	cmd := c.createCommand(context.Background(), "echo ${TEST_VAR}")
+	cmd, _ := c.createCommand(context.Background(), "echo ${TEST_VAR}")
 	if cmd.String() != "/bin/sh -c echo ${TEST_VAR}" {
 		t.Errorf("local build failed %s", cmd.String())
 	}
@@ -37,7 +37,7 @@ func TestContext_BuildContext(t *testing.T) {
 		},
 	}, wcfg)
 
-	cmd = c.createCommand(context.Background(), "echo ${TEST_VAR}")
+	cmd, _ = c.createCommand(context.Background(), "echo ${TEST_VAR}")
 	if cmd.String() != "/opt/docker run --rm -e TEST_VAR=TEST_VAL alpine:latest /bin/sh -c echo ${TEST_VAR}" {
 		t.Errorf("docker build failed %s", cmd.String())
 	}
@@ -57,7 +57,7 @@ func TestContext_BuildContext(t *testing.T) {
 		Up: []string{"docker-compose up -d alpine"},
 	}, wcfg)
 
-	cmd = c.createCommand(context.Background(), "echo ${TEST_VAR}")
+	cmd, _ = c.createCommand(context.Background(), "echo ${TEST_VAR}")
 	if cmd.String() != "/usr/local/bin/docker-compose --file=example/docker-compose.yaml exec -T --user=root -e TEST_VAR=TEST_VAL alpine /bin/sh -c echo ${TEST_VAR}" {
 		t.Errorf("docker-compose build failed %s", cmd.String())
 	}
@@ -75,7 +75,7 @@ func TestContext_BuildContext(t *testing.T) {
 		},
 	}, wcfg)
 
-	cmd = c.createCommand(context.Background(), "echo ${TEST_VAR}")
+	cmd, _ = c.createCommand(context.Background(), "echo ${TEST_VAR}")
 	if cmd.String() != "/usr/bin/kubectl exec deployment/geocoder -- /bin/sh -c TEST_VAR=TEST_VAL echo ${TEST_VAR}" {
 		t.Errorf("kubectl build failed %s", cmd.String())
 	}
@@ -89,7 +89,7 @@ func TestContext_BuildContext(t *testing.T) {
 		},
 	}, wcfg)
 
-	cmd = c.createCommand(context.Background(), "echo ${TEST_VAR}")
+	cmd, _ = c.createCommand(context.Background(), "echo ${TEST_VAR}")
 	if cmd.String() != "/usr/bin/ssh -6 -C -T root@host /bin/sh -c echo ${TEST_VAR}" {
 		t.Errorf("ssh build failed %s", cmd.String())
 	}

pkg/runner/output.go

@@ -13,7 +13,8 @@ import (
 	"sync"
 )
 
-const ansi = "[\u001B\u009B][[\\]()#;?]*(?:(?:(?:[a-zA-Z\\d]*(?:;[a-zA-Z\\d]*)*)?\u0007)|(?:(?:\\d{1,4}(?:;\\d{0,4})*)?[\\dA-PRZcf-ntqry=><~]))"
+// ANSI escape codes except colors
+const ansi = "[\u001B\u009B][[\\]()#;?]*(?:(?:(?:[a-zA-Z\\d]*(?:;[a-zA-Z\\d]*)*)?\u0007)|(?:(?:\\d{1,4}(?:;\\d{0,4})*)?[\\dA-LN-PRZcf-lntqry=><~]))"
 
 type taskOutput struct {
 	raw   bool
@@ -92,15 +93,15 @@ func (o *taskOutput) Scan(t *task.Task, flushed chan struct{}) {
 	} else {
 		go func() {
 			defer wg.Done()
-			err := o.streamDecoratedOutput(t, t.Stdout, lw)
+			err := o.streamDecoratedOutput(t, o.stdout, t.Stdout, lw)
 			if err != nil {
 				log.Debug(err)
 			}
 		}()
 
 		go func() {
 			defer wg.Done()
-			err := o.streamDecoratedOutput(t, t.Stderr, lw)
+			err := o.streamDecoratedOutput(t, o.stderr, t.Stderr, lw)
 			if err != nil {
 				log.Debug(err)
 			}
@@ -130,12 +131,14 @@ func (o *taskOutput) streamRawOutput(dst io.Writer, src io.ReadCloser, lw io.Wri
 	return nil
 }
 
-func (o *taskOutput) streamDecoratedOutput(t *task.Task, r io.ReadCloser, lw io.Writer) error {
-	scanner := bufio.NewScanner(r)
+func (o *taskOutput) streamDecoratedOutput(t *task.Task, dst io.Writer, src io.ReadCloser, lw io.Writer) error {
+	w := io.MultiWriter(dst, lw)
+
+	scanner := bufio.NewScanner(src)
 	for scanner.Scan() {
 		b := scanner.Bytes()
 		bs := o.ansiRegexp.ReplaceAllLiteral(b, []byte{})
-		_, err := fmt.Fprintf(o.stdout, "%s: %s\r\n", aurora.Gray(16, t.Name), bs)
+		_, err := fmt.Fprintf(w, "%s: %s\r\n", aurora.Cyan(t.Name), bs)
 		if err != nil {
 			log.Debug(err)
 		}

pkg/runner/runner.go

@@ -55,7 +55,11 @@ func (r *TaskRunner) RunWithEnv(t *task.Task, env []string) (err error) {
 			ctx, _ = context.WithTimeout(ctx, *t.Timeout)
 		}
 
-		cmd := c.createCommand(ctx, command)
+		cmd, err := c.createCommand(ctx, command)
+		if err != nil {
+			return err
+		}
+
 		cmd.Env = append(cmd.Env, env...)
 		cmd.Env = append(cmd.Env, r.env...)
 

pkg/scheduler/pipeline.go

@@ -3,8 +3,7 @@ package scheduler
 import (
 	"errors"
 	"fmt"
-	log "github.com/sirupsen/logrus"
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/task"
 	"github.com/trntv/wilson/pkg/util"
 )
@@ -87,13 +86,13 @@ func (p *Pipeline) Nodes() map[string]*Stage {
 	return p.nodes
 }
 
-func (p *Pipeline) Node(name string) *Stage {
+func (p *Pipeline) Node(name string) (*Stage, error) {
 	t, ok := p.nodes[name]
 	if !ok {
-		log.Fatalf("unknown task name %s\r\n", name)
+		return nil, fmt.Errorf("unknown task name %s\r\n", name)
 	}
 
-	return t
+	return t, nil
 }
 
 func (p *Pipeline) From(name string) []string {

pkg/scheduler/pipeline_test.go

@@ -1,7 +1,7 @@
 package scheduler
 
 import (
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/task"
 	"testing"
 )

pkg/scheduler/scheduler.go

@@ -59,7 +59,11 @@ func (s *PipelineScheduler) Schedule() {
 
 			var ready = true
 			for _, dep := range s.pipeline.To(name) {
-				depStage := s.pipeline.Node(dep)
+				depStage, err := s.pipeline.Node(dep)
+				if err != nil {
+					log.Fatal(err)
+				}
+
 				switch depStage.Task.ReadStatus() {
 				case task.StatusDone:
 					continue

pkg/task/task.go

@@ -1,7 +1,7 @@
 package task
 
 import (
-	"github.com/trntv/wilson/pkg/config"
+	"github.com/trntv/wilson/internal/config"
 	"github.com/trntv/wilson/pkg/util"
 	"io"
 	"sync"