snapshotting: support 'periodic' and 'manual' mode

1. Change config format to support multiple types
   of snapshotting modes.
2. Implement a hacky way to support periodic or completely
   manual snaphots.

In manual mode, the user has to trigger replication using the wakeup
mechanism after they took snapshots using their own tooling.

As indicated by the comment, a more general solution would be desirable,
but we want to get the release out and 'manual' mode is a feature that
some people requested...

config/config.go

@@ -38,7 +38,7 @@ type PassiveJob struct {
 
 type PushJob struct {
 	ActiveJob `yaml:",inline"`
-	Snapshotting Snapshotting          `yaml:"snapshotting"`
+	Snapshotting SnapshottingEnum          `yaml:"snapshotting"`
 	Filesystems FilesystemsFilter `yaml:"filesystems"`
 }
 
@@ -55,17 +55,26 @@ type SinkJob struct {
 
 type SourceJob struct {
 	PassiveJob `yaml:",inline"`
-	Snapshotting Snapshotting      `yaml:"snapshotting"`
+	Snapshotting SnapshottingEnum      `yaml:"snapshotting"`
 	Filesystems FilesystemsFilter `yaml:"filesystems"`
 }
 
 type FilesystemsFilter map[string]bool
 
-type Snapshotting struct {
+type SnapshottingEnum struct {
-	SnapshotPrefix string        `yaml:"snapshot_prefix"`
+	Ret interface{}
+}
+
+type SnapshottingPeriodic struct {
+	Type string		`yaml:"type"`
+	Prefix string	`yaml:"prefix"`
 	Interval time.Duration `yaml:"interval,positive"`
 }
 
+type SnapshottingManual struct {
+	Type string `yaml:"type"`
+}
+
 type PruningSenderReceiver struct {
 	KeepSender   []PruningEnum `yaml:"keep_sender"`
 	KeepReceiver []PruningEnum `yaml:"keep_receiver"`
@@ -346,6 +355,14 @@ func (t *PruningEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error)
 	return
 }
 
+func (t *SnapshottingEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
+	t.Ret, err = enumUnmarshal(u, map[string]interface{}{
+		"periodic": &SnapshottingPeriodic{},
+		"manual": &SnapshottingManual{},
+	})
+	return
+}
+
 func (t *LoggingOutletEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
 	t.Ret, err = enumUnmarshal(u, map[string]interface{}{
 		"stdout": &StdoutLoggingOutlet{},

config/config_minimal_test.go

@@ -28,8 +28,7 @@ jobs:
     "pool1/poudriere/ports<": false #don't backup the ports trees
   }
   snapshotting:
-    snapshot_prefix: zrepl_
+    type: manual
-    interval: 10m
   pruning:
     keep_sender:
     - type: not_replicated

config/config_snapshotting_test.go

@@ -0,0 +1,57 @@
+package config
+
+import (
+	"fmt"
+	"github.com/stretchr/testify/assert"
+	"testing"
+	"time"
+)
+
+func TestSnapshotting(t *testing.T) {
+	tmpl := `
+jobs:
+- name: foo
+  type: push
+  connect:
+    type: local
+    listener_name: foo
+    client_identity: bar
+  filesystems: {"<": true}
+  %s
+  pruning:
+    keep_sender:
+    - type: last_n
+      count: 10
+    keep_receiver:
+    - type: last_n
+      count: 10
+`
+	manual := `
+  snapshotting:
+    type: manual
+`
+	periodic := `
+  snapshotting:
+    type: periodic
+    prefix: zrepl_
+    interval: 10m
+`
+
+	fillSnapshotting := func(s string) string {return fmt.Sprintf(tmpl, s)}
+	var c *Config
+
+	t.Run("manual", func(t *testing.T) {
+		c = testValidConfig(t, fillSnapshotting(manual))
+		snm := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingManual)
+		assert.Equal(t, "manual", snm.Type)
+	})
+
+	t.Run("periodic", func(t *testing.T) {
+		c = testValidConfig(t, fillSnapshotting(periodic))
+		snp := c.Jobs[0].Ret.(*PushJob).Snapshotting.Ret.(*SnapshottingPeriodic)
+		assert.Equal(t, "periodic", snp.Type)
+		assert.Equal(t, 10*time.Minute, snp.Interval)
+		assert.Equal(t, "zrepl_" , snp.Prefix)
+	})
+
+}

config/samples/local.yml

@@ -17,8 +17,9 @@ jobs:
       "system<": true,
     }
     snapshotting:
-      snapshot_prefix: zrepl_
+      type: periodic
       interval: 10m
+      prefix: zrepl_
     pruning:
       keep_sender:
       - type: not_replicated

config/samples/push.yml

@@ -9,8 +9,7 @@ jobs:
       type: tcp
       address: "backup-server.foo.bar:8888"
     snapshotting:
-      snapshot_prefix: zrepl_
+      type: manual
-      interval: 10m
     pruning:
       keep_sender:
         - type: not_replicated

config/samples/source.yml

@@ -12,5 +12,6 @@ jobs:
     "secret": false
   }
   snapshotting:
-    snapshot_prefix: zrepl_
+    type: periodic
     interval: 10m
+    prefix: zrepl_

config/samples/source_ssh.yml

@@ -11,5 +11,7 @@ jobs:
     "secret": false
   }
   snapshotting:
-    snapshot_prefix: zrepl_
+    type: periodic
     interval: 10m
+    prefix: zrepl_
+

daemon/job/active.go

@@ -62,7 +62,7 @@ type activeMode interface {
 
 type modePush struct {
 	fsfilter         endpoint.FSFilter
-	snapper *snapper.Snapper
+	snapper *snapper.PeriodicOrManual
 }
 
 func (m *modePush) SenderReceiver(client *streamrpc.Client) (replication.Sender, replication.Receiver, error) {
@@ -86,7 +86,7 @@ func modePushFromConfig(g *config.Global, in *config.PushJob) (*modePush, error)
 	}
 	m.fsfilter = fsf
 
-	if m.snapper, err = snapper.FromConfig(g, fsf, &in.Snapshotting); err != nil {
+	if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil {
 		return nil, errors.Wrap(err, "cannot build snapper")
 	}
 

daemon/job/passive.go

@@ -72,7 +72,7 @@ func modeSinkFromConfig(g *config.Global, in *config.SinkJob) (m *modeSink, err
 
 type modeSource struct {
 	fsfilter zfs.DatasetFilter
-	snapper *snapper.Snapper
+	snapper *snapper.PeriodicOrManual
 }
 
 func modeSourceFromConfig(g *config.Global, in *config.SourceJob) (m *modeSource, err error) {
@@ -84,7 +84,7 @@ func modeSourceFromConfig(g *config.Global, in *config.SourceJob) (m *modeSource
 	}
 	m.fsfilter = fsf
 
-	if m.snapper, err = snapper.FromConfig(g, fsf, &in.Snapshotting); err != nil {
+	if m.snapper, err = snapper.FromConfig(g, fsf, in.Snapshotting); err != nil {
 		return nil, errors.Wrap(err, "cannot build snapper")
 	}
 

daemon/snapper/snapper.go

@@ -112,8 +112,8 @@ func getLogger(ctx context.Context) Logger {
 	return logger.NewNullLogger()
 }
 
-func FromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *config.Snapshotting) (*Snapper, error) {
+func PeriodicFromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *config.SnapshottingPeriodic) (*Snapper, error) {
-	if in.SnapshotPrefix == "" {
+	if in.Prefix == "" {
 		return nil, errors.New("prefix must not be empty")
 	}
 	if in.Interval <= 0 {
@@ -121,7 +121,7 @@ func FromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in *config.Snap
 	}
 
 	args := args{
-		prefix: in.SnapshotPrefix,
+		prefix: in.Prefix,
 		interval: in.Interval,
 		fsf: fsf,
 		// ctx and log is set in Run()

daemon/snapper/snapper_all.go

@@ -0,0 +1,39 @@
+package snapper
+
+import (
+	"context"
+	"fmt"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+)
+
+// FIXME: properly abstract snapshotting:
+//   - split up things that trigger snapshotting from the mechanism
+//     - timer-based trigger (periodic)
+//     - call from control socket (manual)
+//     - mixed modes?
+//   - support a `zrepl snapshot JOBNAME` subcommand for config.SnapshottingManual
+type PeriodicOrManual struct {
+	s *Snapper
+}
+
+func (s *PeriodicOrManual) Run(ctx context.Context, wakeUpCommon chan <- struct{}) {
+	if s.s != nil {
+		s.s.Run(ctx, wakeUpCommon)
+	}
+}
+
+func FromConfig(g *config.Global, fsf *filters.DatasetMapFilter, in config.SnapshottingEnum) (*PeriodicOrManual, error) {
+	switch v := in.Ret.(type) {
+	case *config.SnapshottingPeriodic:
+		snapper, err := PeriodicFromConfig(g, fsf, v)
+		if err != nil {
+			return nil, err
+		}
+		return &PeriodicOrManual{snapper}, nil
+	case *config.SnapshottingManual:
+		return &PeriodicOrManual{}, nil
+	default:
+		return nil, fmt.Errorf("unknown snapshotting type %T", v)
+	}
+}