general refactor of codebase

Separated parts out into interfaced components for easier testing.

Author: Southclaws

go.mod

@@ -5,6 +5,7 @@ go 1.13
 require (
 	github.com/Southclaws/gitwatch v1.3.2
 	github.com/cpuguy83/go-md2man/v2 v2.0.0 // indirect
+	github.com/eapache/go-resiliency v1.2.0
 	github.com/frankban/quicktest v1.4.1 // indirect
 	github.com/go-test/deep v1.0.2 // indirect
 	github.com/google/go-cmp v0.3.1 // indirect
@@ -26,6 +27,7 @@ require (
 	golang.org/x/crypto v0.0.0-20200210222208-86ce3cb69678 // indirect
 	golang.org/x/lint v0.0.0-20200130185559-910be7a94367 // indirect
 	golang.org/x/net v0.0.0-20200202094626-16171245cfb2 // indirect
+	golang.org/x/sync v0.0.0-20190423024810-112230192c58
 	golang.org/x/sys v0.0.0-20200212091648-12a6c2dcc1e4 // indirect
 	golang.org/x/time v0.0.0-20191024005414-555d28b269f0 // indirect
 	golang.org/x/tools v0.0.0-20200213200052-63d1300efe97 // indirect

go.sum

@@ -31,6 +31,8 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/eapache/go-resiliency v1.2.0 h1:v7g92e/KSN71Rq7vSThKaWIq68fL4YHvWyiUKorFR1Q=
+github.com/eapache/go-resiliency v1.2.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
 github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
 github.com/emirpasic/gods v1.12.0/go.mod h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
@@ -227,6 +229,7 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58 h1:8gQV6CLnAEikrhgkHFbMAEhagSSnXWGV915qUMm9mrU=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

main.go

@@ -65,7 +65,7 @@ this repository has new commits, Picobot will automatically reconfigure.`,
 				cli.DurationFlag{Name: "check-interval", EnvVar: "CHECK_INTERVAL", Value: time.Second * 10},
 				cli.StringFlag{Name: "vault-addr", EnvVar: "VAULT_ADDR"},
 				cli.StringFlag{Name: "vault-token", EnvVar: "VAULT_TOKEN"},
-				cli.StringFlag{Name: "vault-path", EnvVar: "VAULT_PATH"},
+				cli.StringFlag{Name: "vault-path", EnvVar: "VAULT_PATH", Value: "/secret"},
 				cli.DurationFlag{Name: "vault-renew-interval", EnvVar: "VAULT_RENEW_INTERVAL", Value: time.Hour * 24},
 			},
 			Action: func(c *cli.Context) (err error) {
@@ -88,7 +88,7 @@ this repository has new commits, Picobot will automatically reconfigure.`,
 
 				zap.L().Debug("initialising service")
 
-				svc, err := service.Initialise(ctx, service.Config{
+				svc, err := service.Initialise(service.Config{
 					Target:        c.Args().First(),
 					Hostname:      hostname,
 					Directory:     c.String("directory"),
@@ -106,7 +106,7 @@ this repository has new commits, Picobot will automatically reconfigure.`,
 				zap.L().Info("service initialised")
 
 				errs := make(chan error, 1)
-				go func() { errs <- svc.Start() }()
+				go func() { errs <- svc.Start(ctx) }()
 
 				s := make(chan os.Signal, 1)
 				signal.Notify(s, os.Interrupt)

service/secret/memory/memory.go

@@ -0,0 +1,24 @@
+package memory
+
+import (
+	"context"
+
+	"github.com/picostack/picobot/service/secret"
+)
+
+// MemorySecrets implements a simple in-memory secret.Store for testing
+type MemorySecrets struct {
+	Secrets map[string]string
+}
+
+var _ secret.Store = &MemorySecrets{}
+
+// GetSecretsForTarget implements secret.Store
+func (v *MemorySecrets) GetSecretsForTarget(name string) (map[string]string, error) {
+	return v.Secrets, nil
+}
+
+// Renew implements secret.Store
+func (v *MemorySecrets) Renew(ctx context.Context) error {
+	return nil
+}

service/secret/secret.go

@@ -0,0 +1,9 @@
+package secret
+
+import "context"
+
+// Store describes a type that can securely obtain secrets for services.
+type Store interface {
+	GetSecretsForTarget(name string) (map[string]string, error)
+	Renew(ctx context.Context) error
+}

service/secret/vault/vault.go

@@ -0,0 +1,80 @@
+package vault
+
+import (
+	"context"
+	"path/filepath"
+	"time"
+
+	"github.com/hashicorp/go-cleanhttp"
+	"github.com/hashicorp/vault/api"
+	"github.com/pkg/errors"
+
+	"github.com/picostack/picobot/service/secret"
+)
+
+// VaultSecrets implements a secret.Store backed by Hashicorp Vault
+type VaultSecrets struct {
+	client  *api.Client
+	path    string
+	renewal time.Duration
+}
+
+var _ secret.Store = &VaultSecrets{}
+
+// New creates a new Vault client and pings the server
+func New(addr, path, token string, renewal time.Duration) (*VaultSecrets, error) {
+	client, err := api.NewClient(&api.Config{
+		Address:    addr,
+		HttpClient: cleanhttp.DefaultClient(),
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create vault client")
+	}
+	client.SetToken(token)
+
+	if _, err = client.Auth().Token().LookupSelf(); err != nil {
+		return nil, errors.Wrap(err, "failed to connect to vault server")
+	}
+
+	return &VaultSecrets{
+		client:  client,
+		path:    path,
+		renewal: renewal,
+	}, nil
+}
+
+// GetSecretsForTarget implements secret.Store
+func (v *VaultSecrets) GetSecretsForTarget(name string) (map[string]string, error) {
+	path := filepath.Join(v.path, name)
+	secret, err := v.client.Logical().Read(path)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to read secret")
+	}
+	if secret == nil {
+		return nil, nil
+	}
+
+	env := make(map[string]string)
+	for k, v := range secret.Data {
+		env[k] = v.(string)
+	}
+	return env, nil
+}
+
+// RenewEvery starts a renewal ticker and blocks until fatal error
+// works well with github.com/eapache/go-resiliency
+func (v *VaultSecrets) Renew(ctx context.Context) error {
+	if ctx.Err() == context.Canceled {
+		return ctx.Err()
+	}
+
+	renew := time.NewTicker(v.renewal)
+	defer renew.Stop()
+	for range renew.C {
+		_, err := v.client.Auth().Token().RenewSelf(0)
+		if err != nil {
+			return errors.Wrap(err, "failed to renew vault token")
+		}
+	}
+	return nil
+}

service/secrets.go

@@ -2,36 +2,22 @@ package service
 
 import (
 	"context"
-	"path/filepath"
-	"sync"
 	"time"
 
-	"github.com/Southclaws/gitwatch"
-	"github.com/hashicorp/go-cleanhttp"
-	"github.com/hashicorp/vault/api"
+	"github.com/eapache/go-resiliency/retrier"
 	"github.com/pkg/errors"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"gopkg.in/src-d/go-git.v4/plumbing/transport"
 	"gopkg.in/src-d/go-git.v4/plumbing/transport/ssh"
 
-	"github.com/picostack/picobot/service/config"
-	"github.com/picostack/picobot/service/task"
+	"github.com/picostack/picobot/service/secret"
+	"github.com/picostack/picobot/service/secret/memory"
+	"github.com/picostack/picobot/service/secret/vault"
+	"github.com/picostack/picobot/service/watcher"
 )
 
-// App stores application state
-type App struct {
-	config         Config
-	configWatcher  *gitwatch.Session
-	targets        []task.Target
-	targetsWatcher *gitwatch.Session
-	ssh            transport.AuthMethod
-	vault          *api.Client
-	state          config.State
-	ctx            context.Context
-	cancel         context.CancelFunc
-	errors         chan error
-}
-
+// Config specifies static configuration parameters (from CLI or environment)
 type Config struct {
 	Target        string
 	Hostname      string
@@ -44,108 +30,65 @@ type Config struct {
 	VaultRenewal  time.Duration
 }
 
+// App stores application state
+type App struct {
+	config  Config
+	watcher *watcher.Watcher
+	secrets secret.Store
+}
+
 // Initialise prepares an instance of the app to run
-func Initialise(ctx context.Context, c Config) (app *App, err error) {
+func Initialise(c Config) (app *App, err error) {
 	app = new(App)
 
-	app.ctx, app.cancel = context.WithCancel(ctx)
 	app.config = c
-	app.errors = make(chan error)
 
+	var authMethod transport.AuthMethod
 	if !c.NoSSH {
-		app.ssh, err = ssh.NewSSHAgentAuth("git")
+		authMethod, err = ssh.NewSSHAgentAuth("git")
 		if err != nil {
 			return nil, errors.Wrap(err, "failed to set up SSH authentication")
 		}
 	}
 
+	var secretStore secret.Store
 	if c.VaultAddress != "" {
-		vaultConfig := &api.Config{
-			Address:    c.VaultAddress,
-			HttpClient: cleanhttp.DefaultClient(),
-		}
-		app.vault, err = api.NewClient(vaultConfig)
+		secretStore, err = vault.New(c.VaultAddress, c.VaultPath, c.VaultToken, c.VaultRenewal)
 		if err != nil {
-			return nil, errors.Wrap(err, "failed to connect to vault")
+			return nil, err
 		}
-		app.vault.SetToken(c.VaultToken)
-
-		_, err = app.vault.Logical().List(filepath.Join("/secret", c.VaultPath, "metadata"))
-		if err != nil {
-			return nil, errors.Wrap(err, "failed to ping secrets metadata endpoint")
+	} else {
+		secretStore = &memory.MemorySecrets{
+			// TODO: pull env vars with PICO_SECRET_* or something and shove em here
 		}
 	}
 
-	err = app.reconfigure(c.Hostname)
-	if err != nil {
-		return
-	}
+	app.secrets = secretStore
+
+	app.watcher = watcher.New(
+		secretStore,
+		c.Hostname,
+		c.Directory,
+		c.Target,
+		c.CheckInterval,
+		authMethod,
+	)
 
 	return
 }
 
 // Start launches the app and blocks until fatal error
-func (app *App) Start() (final error) {
-	renew := time.NewTicker(app.config.VaultRenewal)
-	defer renew.Stop()
-
-	f := func() (err error) {
-		select {
-		case <-app.configWatcher.Events:
-			err = app.reconfigure(app.config.Hostname)
-
-		case event := <-app.targetsWatcher.Events:
-			e := app.handle(event)
-			if e != nil {
-				zap.L().Error("failed to handle event",
-					zap.String("url", event.URL),
-					zap.Error(e))
-			}
-
-		case e := <-errorMultiplex(app.configWatcher.Errors, app.targetsWatcher.Errors, app.errors):
-			zap.L().Error("git error",
-				zap.Error(e))
-
-		case <-renew.C:
-			s, e := app.vault.Auth().Token().RenewSelf(0)
-			if e != nil {
-				zap.L().Error("failed to renew vault token",
-					zap.Error(e))
-			}
-			zap.L().Debug("successfully renewed vault token",
-				zap.Any("object", s))
-		}
-		return
-	}
+func (app *App) Start(ctx context.Context) error {
+	g, ctx := errgroup.WithContext(ctx)
 
 	zap.L().Debug("starting service daemon")
 
-	for {
-		final = f()
-		if final != nil {
-			break
-		}
-	}
-	return
-}
+	g.Go(app.watcher.Start)
 
-func errorMultiplex(chans ...<-chan error) <-chan error {
-	out := make(chan error)
-	go func() {
-		var wg sync.WaitGroup
-		wg.Add(len(chans))
-
-		for _, c := range chans {
-			go func(c <-chan error) {
-				for v := range c {
-					out <- v
-				}
-				wg.Done()
-			}(c)
-		}
+	g.Go(func() error {
+		return retrier.New(retrier.ConstantBackoff(3, 100*time.Millisecond), nil).
+			RunCtx(ctx, app.secrets.Renew)
+	})
 
-		wg.Wait()
-		close(out)
-	}()
-	return out
+	return g.Wait()
 }