implement Executor pattern

Makefile

@@ -8,7 +8,7 @@ LDFLAGS=-ldflags "-X main.Version=${VERSION} -X main.BuildTime=${BUILD_TIME}"
 all: build-linux
 
 test:
-	go test ./...
+	go test -v ./...
 
 build-linux:
 	GOOS=linux GOARCH=amd64 go build ${LDFLAGS} -o bin/thanks ./cmd/thanks

internal/executor/executor.go

@@ -0,0 +1,11 @@
+package executor
+
+import (
+	"context"
+	"os/exec"
+)
+
+type Executor interface {
+	CombinedOutput(context.Context, string, ...string) ([]byte, error)
+	CommandContext(context.Context, string, ...string) *exec.Cmd
+}

internal/executor/local.go

@@ -0,0 +1,16 @@
+package executor
+
+import (
+	"context"
+	"os/exec"
+)
+
+type LocalExecutor struct{}
+
+func (l LocalExecutor) CombinedOutput(ctx context.Context, name string, arg ...string) ([]byte, error) {
+	return l.CommandContext(ctx, name, arg...).CombinedOutput()
+}
+
+func (l LocalExecutor) CommandContext(ctx context.Context, name string, arg ...string) *exec.Cmd {
+	return exec.CommandContext(ctx, name, arg...)
+}

internal/executor/ssh.go

@@ -0,0 +1,21 @@
+package executor
+
+import (
+	"context"
+	"os/exec"
+)
+
+type SSHExecutor struct {
+	SSHTarget string
+}
+
+func (s SSHExecutor) CombinedOutput(ctx context.Context, name string, arg ...string) ([]byte, error) {
+	return s.CommandContext(ctx, name, arg...).CombinedOutput()
+}
+
+func (s SSHExecutor) CommandContext(ctx context.Context, name string, arg ...string) *exec.Cmd {
+	newArg := append([]string{s.SSHTarget, name}, arg...)
+	cmd := exec.CommandContext(ctx, "ssh", newArg...)
+
+	return cmd
+}

internal/jobs/jobs.go

@@ -7,71 +7,58 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"os"
 	"os/exec"
-	"strconv"
 	"strings"
 	"time"
 
+	"git.gentoo.party/sam/thanks/internal/executor"
 	"git.gentoo.party/sam/thanks/internal/zfs"
 )
 
-type BackupJob struct {
-	Source     string `yaml:"source"`     // the source dataset (e.g., zroot)
-	TargetHost string `yaml:"targetHost"` // SSH-compatible host
-	Target     string `yaml:"target"`     // the target dataset
-	Keep       int    `yaml:"keep"`       // number of snapshots to keep
-	Recursive  bool   `yaml:"recursive"`  // create recursive snapshots
-	Prefix     string `yaml:"prefix"`     // name each snapshot with this prefix
+type BackupHookCommand struct {
+	Command string   `yaml:"command"`
+	Args    []string `yaml:"args"`
+	Env     []string `yaml:"env"`
 }
 
-//	func (j *BackupJob) getBaseSnap() {
-//		params := "zfs get -j -d 1 -t snapshot guid,creation %s"
-//		srcCmd := fmt.Sprintf(params, j.Source)
-//		dstCmd := fmt.Sprintf(params, j.Target)
-//	}
-func (j *BackupJob) listSnapshots(ctx context.Context) ([]zfs.Snapshot, error) {
-	out, err := zfs.Cmd(ctx, "zfs list -Hp -o name,creation -t snapshot -s creation %s | grep '%s'", j.Source, j.Prefix)
-	if err != nil {
-		execErr, ok := err.(*exec.ExitError)
-		if !ok {
-			log.Printf("%s\n", out)
-			return nil, err
-		}
+type BackupHooks struct {
+	Completed BackupHookCommand `yaml:"completed"` // called upon successful completion of a backup job
+	Error     BackupHookCommand `yaml:"error"`
+}
 
-		if execErr.ExitCode() == 1 {
-			return nil, nil
-		}
+type BackupJob struct {
+	Source     string        `yaml:"source"`     // the source dataset (e.g., zroot)
+	TargetHost string        `yaml:"targetHost"` // SSH-compatible host
+	Target     string        `yaml:"target"`     // the target dataset
+	Keep       int           `yaml:"keep"`       // number of snapshots to keep
+	MaxAge     time.Duration `yaml:"maxAge"`     // age at which to delete snapshot (if Keep is met)
+	Recursive  bool          `yaml:"recursive"`  // create recursive snapshots
+	Prefix     string        `yaml:"prefix"`     // name each snapshot with this prefix
+	Hooks      BackupHooks   `yaml:"hooks"`      // external programs (libnotify, sendmail, etc) to call
+
+	localExecutor  executor.Executor
+	remoteExecutor executor.Executor
+}
+
+func (j *BackupJob) listSnapshots(ctx context.Context) ([]zfs.Snapshot, error) {
+	allSnaps, err := zfs.Snapshots(ctx, j.localExecutor, j.Source)
+	if err != nil {
+		return nil, err
 	}
 
-	snapList := strings.Split(string(out), "\n")
-	snaps := make([]zfs.Snapshot, len(snapList)-1)
-
-	for i, snap := range snapList {
-		params := strings.Split(snap, "\t") // "zroot@foo\tTIME" -> ["zroot@foo", "TIME"]
-		if len(params) != 2 {
+	filtered := make([]zfs.Snapshot, 0)
+	for _, snap := range allSnaps {
+		if !strings.Contains(snap.Name, j.Prefix) {
 			continue
 		}
-		log.Printf("%+v\n", params)
-		t, err := strconv.ParseInt(params[1], 10, 64)
-		if err != nil {
-			log.Fatalf("invalid time %s", params[1])
-		}
-
-		identifier := strings.Split(params[0], "@") // zroot@foo -> ["zroot", "foo"]
-
-		snaps[i] = zfs.Snapshot{
-			SnapshotName: identifier[1],
-			Dataset:      identifier[0],
-			Name:         params[0],
-			Creation:     time.Unix(t, 0),
-		}
+		filtered = append(filtered, snap)
 	}
 
-	return snaps, nil
+	return filtered, nil
 }
 
 func (j *BackupJob) snapName() string {
-	// ts := time.Now().Format(time.RFC3339)
 	ts := time.Now().UnixMicro()
 	if j.Prefix == "" {
 		return fmt.Sprintf("%d", ts)
@@ -83,51 +70,101 @@ func (j *BackupJob) snapName() string {
 func (j *BackupJob) Snapshot(ctx context.Context) (string, error) {
 	snapName := j.snapName()
 	snap := fmt.Sprintf("%s@%s", j.Source, snapName)
-
-	out, err := zfs.Cmd(
+	err := zfs.CreateSnapshot(
 		ctx,
-		"zfs snapshot %s",
-		snap,
+		j.localExecutor,
+		j.Source,
+		snapName,
 	)
-	if err != nil {
-		log.Printf("zfs-snapshot: error: %s", out)
-		return snapName, nil
-	}
 	return snap, err
 }
 
-func (j *BackupJob) FullSend(ctx context.Context, snap string) {
-	out, err := zfs.Cmd(
+func (j *BackupJob) FullSend(ctx context.Context, snap string) error {
+	out, err := zfs.FullSend(
 		ctx,
-		"zfs send -Lec %s | ssh %s zfs recv -Fu %s",
 		snap,
 		j.TargetHost,
 		j.Target,
 	)
 	if err != nil {
-		log.Fatalf("zfs-send: error: %s", out)
+		return fmt.Errorf("zfs-send: error: %s - %s", err.Error(), out)
 	}
+	return nil
 }
 
-func (j *BackupJob) IncrementalSend(ctx context.Context, prevSnap *zfs.Snapshot, newSnap string) {
-	out, err := zfs.Cmd(
+func (j *BackupJob) IncrementalSend(ctx context.Context, prevSnap *zfs.Snapshot, newSnap string) ([]byte, error) {
+	return zfs.IncrementalSend(
 		ctx,
-		"zfs send -I@%s %s | ssh %s zfs recv %s",
 		prevSnap.SnapshotName,
 		newSnap,
 		j.TargetHost,
 		j.Target,
 	)
-	if err != nil {
-		// FIXME: return an error so we can cleanup
-		log.Fatalf("zfs-send: error: %s", out)
-	}
 }
 
-func (j *BackupJob) Do(ctx context.Context) {
+func (j *BackupJob) Retain(ctx context.Context) error {
+	snaps, err := j.listSnapshots(ctx)
+	if err != nil {
+		return fmt.Errorf("retain: failure listing snapshots: %s", err.Error())
+	}
+
+	forDeletion := make([]zfs.Snapshot, 0, len(snaps))
+	now := time.Now()
+
+	if len(snaps) < j.Keep {
+		return nil
+	}
+
+	deleteCount := len(snaps) - j.Keep
+
+	for i, snap := range snaps {
+		age := now.Sub(snap.Creation)
+		if age >= j.MaxAge {
+			forDeletion[i] = snap
+		}
+
+		if len(forDeletion) == deleteCount {
+			break
+		}
+	}
+
+	return nil
+}
+
+func (j *BackupJob) findCommonAnscestor(ctx context.Context) {
+	localSnapshots, err := j.listSnapshots(ctx)
+}
+
+type HookErr struct {
+	message string
+}
+
+func (e *HookErr) Error() string {
+	return fmt.Sprintf("error while running hook: %s", e.message)
+}
+
+func (j *BackupJob) Do(ctx context.Context) error {
+	err := j.do(ctx)
+	if err != nil {
+		log.Print("job encountered error, running error hook\n")
+		err = j.runHookError(ctx, err)
+		if err != nil {
+			log.Printf("error running hook! %s", err.Error())
+		}
+		return &HookErr{message: err.Error()}
+
+	} else {
+		err = j.runHookCompleted(ctx)
+		return &HookErr{message: err.Error()}
+	}
+
+	return err
+}
+
+func (j *BackupJob) do(ctx context.Context) error {
 	currentSnaps, err := j.listSnapshots(ctx)
 	if err != nil {
-		log.Fatal("failed to list existing snapshots")
+		return fmt.Errorf("failed to list existing snapshots: %s", err.Error())
 	}
 	var prev *zfs.Snapshot
 
@@ -137,14 +174,70 @@ func (j *BackupJob) Do(ctx context.Context) {
 
 	newSnapshot, err := j.Snapshot(ctx)
 	if err != nil {
-		log.Fatalf("error creating new snapshot: %s", err.Error())
+		return fmt.Errorf("error creating new snapshot: %s", err.Error())
 	}
 
+	var sendErr error
 	if prev == nil {
 		// Full Send
-		j.FullSend(ctx, newSnapshot)
+		sendErr = j.FullSend(ctx, newSnapshot)
 	} else {
 		// Inc send
-		j.IncrementalSend(ctx, prev, newSnapshot)
+		_, sendErr = j.IncrementalSend(ctx, prev, newSnapshot)
 	}
+
+	if sendErr != nil {
+		return fmt.Errorf("error sending backup: %s", sendErr.Error())
+	}
+
+	return nil
+}
+
+func (j *BackupJob) runHookCompleted(ctx context.Context) error {
+	cmdDef := j.Hooks.Completed
+	if cmdDef.Command == "" {
+		return nil
+	}
+
+	timeoutCtx, cancel := context.WithTimeout(
+		ctx,
+		time.Second*3,
+	)
+	cmd := exec.CommandContext(timeoutCtx, cmdDef.Command, cmdDef.Args...)
+	defer cancel()
+	cmd.Env = []string{
+		fmt.Sprintf("THANKS_JOB_SOURCE=%s", j.Source),
+		fmt.Sprintf("THANKS_JOB_TARGET=%s", j.Target),
+	}
+	return cmd.Run()
+}
+
+func (j *BackupJob) runHookError(ctx context.Context, err error) error {
+	cmdDef := j.Hooks.Error
+	if cmdDef.Command == "" {
+		return nil
+	}
+
+	timeoutCtx, cancel := context.WithTimeout(
+		ctx,
+		time.Second*3,
+	)
+	cmd := exec.CommandContext(timeoutCtx, cmdDef.Command, cmdDef.Args...)
+	defer cancel()
+	cmd.Env = append(
+		[]string{
+			fmt.Sprintf("THANKS_JOB_SOURCE=%s", j.Source),
+			fmt.Sprintf("THANKS_JOB_TARGET=%s", j.Target),
+			fmt.Sprintf("THANKS_JOB_ERROR=%s", err.Error()),
+			fmt.Sprintf("DISPLAY=%s", os.Getenv("DISPLAY")),
+		},
+		cmdDef.Env...,
+	)
+
+	out, runErr := cmd.CombinedOutput()
+	if runErr != nil {
+		log.Printf("%s", out)
+	}
+
+	return runErr
 }

internal/jobs/jobs_test.go

@@ -0,0 +1,36 @@
+package jobs_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"git.gentoo.party/sam/thanks/internal/jobs"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_JobHookError(t *testing.T) {
+	job := jobs.BackupJob{
+		Target:     "znothing/nothing",
+		TargetHost: "devnull",
+		Source:     "zempty/empty",
+		Keep:       1,
+		MaxAge:     1 * time.Second,
+		Recursive:  false,
+		Hooks: jobs.BackupHooks{
+			Error: jobs.BackupHookCommand{
+				Command: "/usr/bin/notify-send",
+				Args:    []string{"--app-name=thanks", "--urgency=CRITICAL", "Backup Failure", "Backup failed"},
+				Env: []string{
+					"XDG_RUNTIME_DIR=/run/user/1000/",
+				},
+			},
+		},
+	}
+
+	var hookErr *jobs.HookErr
+
+	err := job.Do(context.Background())
+	assert.NotNil(t, err, "expected job to produce an error")
+	assert.NotErrorAsf(t, err, &hookErr, "expected a non-hook error, got %w", err)
+}

internal/runner/runner.go

@@ -1,6 +1,7 @@
 package runner
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"os/exec"
@@ -26,3 +27,35 @@ func (runner *ZRunner) Run(ctx context.Context, command string, formatArgs ...an
 	cmd := exec.Command(runner.Prog, args...)
 	return cmd.CombinedOutput()
 }
+
+// Pipeline accepts a slice of [exec.Cmd] and pipes them together
+// then returns the result of the final command
+func Pipeline(cmds ...*exec.Cmd) ([]byte, error) {
+	for i := 0; i < len(cmds)-1; i++ {
+		stdout, err := cmds[i].StdoutPipe()
+		if err != nil {
+			return nil, err
+		}
+		cmds[i+1].Stdin = stdout
+	}
+
+	out := bytes.Buffer{}
+
+	cmds[len(cmds)-1].Stdout = &out
+
+	for _, cmd := range cmds {
+		err := cmd.Start()
+		if err != nil {
+			return out.Bytes(), err
+		}
+	}
+
+	for _, cmd := range cmds {
+		err := cmd.Wait()
+		if err != nil {
+			return out.Bytes(), err
+		}
+	}
+
+	return out.Bytes(), nil
+}

internal/runner/runner_test.go

@@ -2,9 +2,11 @@ package runner_test
 
 import (
 	"context"
+	"os/exec"
 	"testing"
 
 	"git.gentoo.party/sam/thanks/internal/runner"
+	"github.com/stretchr/testify/assert"
 )
 
 func Test_ZCommand(t *testing.T) {
@@ -19,3 +21,12 @@ func Test_ZCommand(t *testing.T) {
 		t.Errorf("localRunner failed: %s\n\n%s", err.Error(), out)
 	}
 }
+
+func Test_Pipelie(t *testing.T) {
+	c1 := exec.Command("echo", "-n", "foo")
+	c2 := exec.Command("rev")
+	out, err := runner.Pipeline(c1, c2)
+
+	assert.Equal(t, "oof", string(out))
+	assert.Nil(t, err)
+}

internal/zfs/zfs-list_snapshot.txt

@@ -0,0 +1,6 @@
+zroot/home/sam/thanks@thanks-1769803874756987	1769803874	9856814317153087085
+zroot/home/sam/thanks@thanks-1769804009334385	1769804009	10898692868281532431
+zroot/home/sam/thanks@thanks-1769810365460466	1769810365	17939462811459040773
+zroot/home/sam/thanks@thanks-1769810764161053	1769810764	17021914338902266865
+zroot/home/sam/thanks@thanks-1769813076492805	1769813076	17716303459843516357
+zroot/home/sam/thanks@thanks-1769813083041347	1769813083	905910106048059171

internal/zfs/zfs.go

@@ -1,10 +1,18 @@
 package zfs
 
 import (
+	"bufio"
 	"context"
 	"fmt"
+	"io"
+	"log"
 	"os/exec"
+	"strconv"
+	"strings"
 	"time"
+
+	"git.gentoo.party/sam/thanks/internal/executor"
+	"git.gentoo.party/sam/thanks/internal/runner"
 )
 
 type Snapshot struct {
@@ -12,7 +20,7 @@ type Snapshot struct {
 	Name         string // Full dataset@snapname
 	Dataset      string
 	Creation     time.Time
-	GUID         int64
+	GUID         uint64
 }
 
 func Cmd(ctx context.Context, arg string, a ...any) ([]byte, error) {
@@ -29,3 +37,141 @@ func Cmd(ctx context.Context, arg string, a ...any) ([]byte, error) {
 	}
 	return out, err
 }
+
+func CreateSnapshot(ctx context.Context, e executor.Executor, dataset, name string) error {
+	_, err := e.CombinedOutput(
+		ctx,
+		"zfs",
+		"snapshot",
+		fmt.Sprintf("%s@%s", dataset, name),
+	)
+	if err != nil {
+		return fmt.Errorf("zfs-snapshot error: %w", err)
+	}
+
+	return nil
+}
+
+func ParseSnapshots(reader io.Reader) ([]Snapshot, error) {
+	scanner := bufio.NewScanner(reader)
+
+	snaps := make([]Snapshot, 0)
+	for scanner.Scan() {
+		line := scanner.Text()
+		params := strings.Split(line, "\t") // "zroot@foo\tTIME" -> ["zroot@foo", "TIME"]
+		if len(params) != 3 {
+			continue
+		}
+		t, err := strconv.ParseInt(params[1], 10, 64)
+		if err != nil {
+			log.Fatalf("invalid time %s", params[1])
+		}
+
+		identifier := strings.Split(params[0], "@") // zroot@foo -> ["zroot", "foo"]
+
+		GUID, err := strconv.ParseUint(params[2], 10, 64)
+		if err != nil {
+			log.Fatalf("invalid GUID: %s - %s", params[2], err.Error())
+		}
+
+		snaps = append(
+			snaps,
+			Snapshot{
+				SnapshotName: identifier[1],
+				Dataset:      identifier[0],
+				Name:         params[0],
+				Creation:     time.Unix(t, 0),
+				GUID:         GUID,
+			},
+		)
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, fmt.Errorf("error parsing zfs-list output: %s", err.Error())
+	}
+
+	return snaps, nil
+}
+
+func Snapshots(ctx context.Context, e executor.Executor, target string) ([]Snapshot, error) {
+	cmd := e.CommandContext(
+		ctx,
+		"zfs",
+		"list",
+		"-Hp",
+		"-o",
+		"name,creation,guid",
+		"-t",
+		"snapshot",
+		"-s",
+		"creation",
+		target,
+	)
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+	defer stdout.Close()
+
+	err = cmd.Start()
+	if err != nil {
+		return nil, err
+	}
+
+	snaps, err := ParseSnapshots(stdout)
+	if err != nil {
+		return nil, err
+	}
+
+	err = cmd.Wait()
+	if err != nil {
+		return nil, err
+	}
+
+	return snaps, err
+}
+
+func FullSend(ctx context.Context, snap, sshTarget, target string) ([]byte, error) {
+	return runner.Pipeline(
+		exec.CommandContext(
+			ctx,
+			"zfs",
+			"send",
+			"-Lec",
+			snap,
+		),
+		exec.CommandContext(
+			ctx,
+			"ssh",
+			sshTarget,
+			"zfs",
+			"recv",
+			"-Fu",
+			target,
+		),
+	)
+}
+
+func IncrementalSend(ctx context.Context, prevSnapName, newSnapName, sshTarget, target string) ([]byte, error) {
+	return runner.Pipeline(
+		exec.CommandContext(
+			ctx,
+			"zfs",
+			"send",
+			fmt.Sprintf("-I@%s", prevSnapName),
+			newSnapName,
+		),
+		exec.CommandContext(
+			ctx,
+			"ssh",
+			sshTarget,
+			"zfs",
+			"recv",
+			target,
+		),
+	)
+}
+
+func Destroy(ctx context.Context, e executor.Executor, target string) ([]byte, error) {
+	return e.CombinedOutput(ctx, "zfs", target)
+}

internal/zfs/zfs_test.go

@@ -2,8 +2,12 @@ package zfs_test
 
 import (
 	"fmt"
+	"os"
 	"os/exec"
 	"testing"
+
+	"git.gentoo.party/sam/thanks/internal/zfs"
+	"github.com/stretchr/testify/assert"
 )
 
 type zfscmd struct {
@@ -43,3 +47,14 @@ func Test_ZFSCommand(t *testing.T) {
 		t.Errorf("SSH zrunner failed: %s \n\n%s", err.Error(), out)
 	}
 }
+
+func Test_Snapshots(t *testing.T) {
+	f, err := os.Open("./zfs-list_snapshot.txt")
+	assert.Nil(t, err, "failed to open test data")
+	out, err := zfs.ParseSnapshots(f)
+	assert.Nil(t, err, "ParseSnapshots returned non-nil error")
+
+	for _, snap := range out {
+		assert.Contains(t, snap.SnapshotName, "thanks")
+	}
+}

thanks.yaml.example

@@ -0,0 +1,14 @@
+jobs:
+  - source: "zroot/home/sam/thanks"
+    target: "zrust/backup/weller/thanks"
+    targetHost: "backup@woodford.gentoo.party"
+    keep: 30
+    prefix: "thanks-"
+    recursive: false
+    hooks:
+      completed: 
+        cmd: "backup-success.sh" 
+        args: []
+      error:
+        cmd: "backup-error.sh" 
+        args: []