Split event handler from watcher (#36)

* split apart task hydration+execution from watcher/config package with a simple channel

* moved task list diff code to task package

* implemented reconfigurer

* removed env logging from target command execution debug log - it's mostly unnecessary and makes reading logs during dev annoying

* written some mocks for tests

* written tests for state transitions

* moved execution specific environment variables to the ExecutionTask structure

Author: Southclaws

.gitignore

@@ -24,3 +24,5 @@ pico
 dist
 .env
 cache
+
+*/.test/

service/executor/cmd.go

@@ -0,0 +1,56 @@
+package executor
+
+import (
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+
+	"github.com/picostack/pico/service/secret"
+	"github.com/picostack/pico/service/task"
+)
+
+var _ Executor = &CommandExecutor{}
+
+// CommandExecutor handles command invocation targets
+type CommandExecutor struct {
+	secrets secret.Store
+}
+
+// NewCommandExecutor creates a new CommandExecutor
+func NewCommandExecutor(secrets secret.Store) CommandExecutor {
+	return CommandExecutor{
+		secrets: secrets,
+	}
+}
+
+// Subscribe implements executor.Executor
+func (e *CommandExecutor) Subscribe(bus chan task.ExecutionTask) {
+	for t := range bus {
+		if err := e.execute(t.Target, t.Path, t.Shutdown); err != nil {
+			zap.L().Error("executor task unsuccessful",
+				zap.String("target", t.Target.Name),
+				zap.Bool("shutdown", t.Shutdown),
+				zap.Error(err))
+		}
+	}
+	return
+}
+
+func (e *CommandExecutor) execute(
+	target task.Target,
+	path string,
+	shutdown bool,
+) (err error) {
+	env, err := e.secrets.GetSecretsForTarget(target.Name)
+	if err != nil {
+		return errors.Wrap(err, "failed to get secrets for target")
+	}
+
+	zap.L().Debug("executing with secrets",
+		zap.String("target", target.Name),
+		zap.Strings("cmd", target.Up),
+		zap.String("url", target.RepoURL),
+		zap.String("dir", path),
+		zap.Int("secrets", len(env)))
+
+	return target.Execute(path, env, shutdown)
+}

service/executor/cmd_test.go

@@ -0,0 +1,49 @@
+package executor_test
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/picostack/pico/service/executor"
+	"github.com/picostack/pico/service/secret/memory"
+	"github.com/picostack/pico/service/task"
+)
+
+func TestMain(m *testing.M) {
+	os.Mkdir(".test", os.ModePerm) //nolint:errcheck
+	os.Exit(m.Run())
+}
+
+func TestCommandExecutor(t *testing.T) {
+	ce := executor.NewCommandExecutor(&memory.MemorySecrets{})
+	bus := make(chan task.ExecutionTask)
+
+	g := errgroup.Group{}
+
+	g.Go(func() error {
+		bus <- task.ExecutionTask{
+			Target: task.Target{
+				Name: "test_executor",
+				Up:   []string{"touch", "01"},
+			},
+			Path: "./.test",
+		}
+		return nil
+	})
+
+	go ce.Subscribe(bus)
+
+	if err := g.Wait(); err != nil {
+		t.Error(err)
+	}
+
+	// wait for the task to be consumed and executed
+	time.Sleep(time.Second)
+
+	if _, err := os.Stat(".test/01"); err == os.ErrNotExist {
+		t.Error("expected file .test/01 to exist:", err)
+	}
+}

service/executor/executor.go

@@ -0,0 +1,9 @@
+package executor
+
+import "github.com/picostack/pico/service/task"
+
+// Executor describes a type that can handle events and react to them. An
+// executor is also responsible for hydrating a target with secrets.
+type Executor interface {
+	Subscribe(chan task.ExecutionTask)
+}

service/executor/printer.go

@@ -0,0 +1,17 @@
+package executor
+
+import (
+	"fmt"
+
+	"github.com/picostack/pico/service/task"
+)
+
+// Printer implements an executor that doesn't actually execute, just prints.
+type Printer struct{}
+
+// Subscribe implements executor.Executor
+func (p *Printer) Subscribe(bus chan task.ExecutionTask) {
+	for t := range bus {
+		fmt.Printf("received task: %s\n", t.Target.Name)
+	}
+}

service/reconfigurer/git.go

@@ -0,0 +1,141 @@
+package reconfigurer
+
+import (
+	"context"
+	"path/filepath"
+	"time"
+
+	"github.com/Southclaws/gitwatch"
+	"github.com/picostack/pico/service/config"
+	"github.com/picostack/pico/service/watcher"
+	"github.com/pkg/errors"
+	"go.uber.org/zap"
+	"gopkg.in/src-d/go-git.v4/plumbing/transport"
+)
+
+var _ Provider = &GitProvider{}
+
+// GitProvider implements a Provider backed by Git. It will reconfigure its
+// watcher process upon commits to its defined configuration repository.
+type GitProvider struct {
+	directory     string
+	hostname      string
+	configRepo    string
+	checkInterval time.Duration
+	ssh           transport.AuthMethod
+
+	configWatcher *gitwatch.Session
+}
+
+// New creates a new provider with all necessary parameters
+func New(
+	directory string,
+	hostname string,
+	configRepo string,
+	checkInterval time.Duration,
+	ssh transport.AuthMethod,
+) *GitProvider {
+	return &GitProvider{
+		directory:     directory,
+		hostname:      hostname,
+		configRepo:    configRepo,
+		checkInterval: checkInterval,
+		ssh:           ssh,
+	}
+}
+
+// Configure implements Provider
+func (p *GitProvider) Configure(w watcher.Watcher) error {
+	if err := p.reconfigure(w); err != nil {
+		return err
+	}
+
+	for range p.configWatcher.Events {
+		if err := p.reconfigure(w); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// reconfigure will close the configuration watcher (unless it's the first run)
+// then create a watcher for the application's config target repo then wait for
+// the first event (either from a fresh clone, a pull, or just a noop event)
+// then update the state of the watcher it's in charge of.
+func (p *GitProvider) reconfigure(w watcher.Watcher) (err error) {
+	zap.L().Debug("reconfiguring")
+
+	err = p.watchConfig()
+	if err != nil {
+		return
+	}
+
+	// generate a new desired state from the config repo
+	path, err := gitwatch.GetRepoDirectory(p.configRepo)
+	if err != nil {
+		return
+	}
+	state := getNewState(
+		filepath.Join(p.directory, path),
+		p.hostname,
+		w.GetState(),
+	)
+
+	// Set the HOSTNAME config environment variable if necessary.
+	if p.hostname != "" {
+		state.Env["HOSTNAME"] = p.hostname
+	}
+
+	return w.SetState(state)
+}
+
+// watchConfig creates or restarts the watcher that reacts to changes to the
+// repo that contains pico configuration scripts
+func (p *GitProvider) watchConfig() (err error) {
+	if p.configWatcher != nil {
+		zap.L().Debug("closing existing watcher")
+		p.configWatcher.Close()
+	}
+
+	p.configWatcher, err = gitwatch.New(
+		context.TODO(),
+		[]gitwatch.Repository{{URL: p.configRepo}},
+		p.checkInterval,
+		p.directory,
+		p.ssh,
+		false)
+	if err != nil {
+		return errors.Wrap(err, "failed to watch config target")
+	}
+
+	go func() {
+		e := p.configWatcher.Run()
+		if e != nil && !errors.Is(e, context.Canceled) {
+			zap.L().Error("config watcher failed", zap.Error(e))
+		}
+	}()
+	zap.L().Debug("created new config watcher, awaiting setup")
+
+	<-p.configWatcher.InitialDone
+
+	return
+}
+
+// getNewState attempts to obtain a new desired state from the given path, if
+// any failures occur, it simply returns a fallback state and logs an error
+func getNewState(path, hostname string, fallback config.State) (state config.State) {
+	state, err := config.ConfigFromDirectory(path, hostname)
+	if err != nil {
+		zap.L().Error("failed to construct config from repo, falling back to original state",
+			zap.String("path", path),
+			zap.String("hostname", hostname),
+			zap.Error(err))
+
+		state = fallback
+	} else {
+		zap.L().Debug("constructed desired state",
+			zap.Int("targets", len(state.Targets)))
+	}
+	return
+}

service/reconfigurer/reconfigurer.go

@@ -0,0 +1,11 @@
+package reconfigurer
+
+import (
+	"github.com/picostack/pico/service/watcher"
+)
+
+// Provider describes a type that can provide config state events to a target
+// watcher. It will reconfigure and restart the watcher whenever necessary.
+type Provider interface {
+	Configure(watcher.Watcher) error
+}

service/reconfigurer/static.go

@@ -0,0 +1,26 @@
+package reconfigurer
+
+import (
+	"github.com/picostack/pico/service/config"
+	"github.com/picostack/pico/service/watcher"
+)
+
+var _ Provider = &Static{}
+
+// Static implements a Provider with a static config state that only sets its
+// watcher state once on initialisation.
+type Static struct {
+	state config.State
+}
+
+// NewStatic creates and calls SetState
+func NewStatic(state config.State, w watcher.Watcher) Static {
+	s := Static{state: state}
+	s.Configure(w)
+	return s
+}
+
+// Configure implements Provider
+func (s *Static) Configure(w watcher.Watcher) error {
+	return w.SetState(s.state)
+}