WIP

From https://github.com/zrepl/zrepl/issues/691

The last_n prune rule keeps everything, regardless of if it matches the
regex or not, if there are less than count snapshot. The expectation
would be to never keep non-regex snapshots, regardless of number.

.circleci/config.yml

@@ -58,19 +58,26 @@ commands:
             git config --global user.email "zreplbot@cschwarz.com"
             git config --global user.name "zrepl-github-io-ci"
 
-      # https://circleci.com/docs/2.0/add-ssh-key/#adding-multiple-keys-with-blank-hostnames
-      - run: ssh-add -D
-      # the default circleci ssh config only additional ssh keys for Host !github.com
-      - run:
-          command: |
-            cat > ~/.ssh/config \<<EOF
-            Host *
-              IdentityFile /home/circleci/.ssh/id_rsa_458e62c517f6c480e40452126ce47421
-            EOF
-      - add_ssh_keys:
-          fingerprints:
-            # deploy key for zrepl.github.io
-            - "45:8e:62:c5:17:f6:c4:80:e4:04:52:12:6c:e4:74:21"
+      # if we're pushing, we need to add the deploy key
+      # which is stored as "Additional SSH Keys" in the CircleCI project settings.
+      # We can't use the CircleCI-manage deploy key because we're pushing
+      # to a different repo than the one we're building.
+      - when:
+          condition: << parameters.push >>
+          steps:
+            # https://circleci.com/docs/2.0/add-ssh-key/#adding-multiple-keys-with-blank-hostnames
+            - run: ssh-add -D
+            # the default circleci ssh config only additional ssh keys for Host !github.com
+            - run:
+                command: |
+                  cat > ~/.ssh/config \<<EOF
+                  Host *
+                    IdentityFile /home/circleci/.ssh/id_rsa_458e62c517f6c480e40452126ce47421
+                  EOF
+            - add_ssh_keys:
+                fingerprints:
+                  # deploy key for zrepl.github.io
+                  - "45:8e:62:c5:17:f6:c4:80:e4:04:52:12:6c:e4:74:21"
 
       # caller must install-docdep
       - when:

config/config.go

@@ -121,6 +121,7 @@ type BandwidthLimit struct {
 }
 
 type Replication struct {
+	Triggers    []*ReplicationTriggerEnum
 	Protection  *ReplicationOptionsProtection  `yaml:"protection,optional,fromdefaults"`
 	Concurrency *ReplicationOptionsConcurrency `yaml:"concurrency,optional,fromdefaults"`
 }
@@ -135,6 +136,32 @@ type ReplicationOptionsConcurrency struct {
 	SizeEstimates int `yaml:"size_estimates,optional,default=4"`
 }
 
+type ReplicationTriggerEnum struct {
+	Ret interface{}
+}
+
+func (t *ReplicationTriggerEnum) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
+	t.Ret, err = enumUnmarshal(u, map[string]interface{}{
+		"manual":   &ReplicationTriggerManual{},
+		"periodic": &ReplicationTriggerPeriodic{},
+	})
+	return
+}
+
+type ReplicationTriggerManual struct {
+	Type string `yaml:"type"`
+}
+
+type ReplicationTriggerPeriodic struct {
+	Type     string            `yaml:"type"`
+	Interval *PositiveDuration `yaml:"interval"`
+}
+
+type ReplicationTriggerCron struct {
+	Type string   `yaml:"type"`
+	Cron CronSpec `yaml:"cron"`
+}
+
 type PropertyRecvOptions struct {
 	Inherit  []zfsprop.Property          `yaml:"inherit,optional"`
 	Override map[zfsprop.Property]string `yaml:"override,optional"`
@@ -157,7 +184,6 @@ func (j *PushJob) GetSendOptions() *SendOptions      { return j.Send }
 type PullJob struct {
 	ActiveJob `yaml:",inline"`
 	RootFS    string                   `yaml:"root_fs"`
-	Interval  PositiveDurationOrManual `yaml:"interval"`
 	Recv      *RecvOptions             `yaml:"recv,fromdefaults,optional"`
 }
 

daemon/job/active.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job/reset"
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
@@ -44,6 +45,8 @@ type ActiveSide struct {
 	promReplicationErrors prometheus.Gauge
 	promLastSuccessful    prometheus.Gauge
 
+	triggers *trigger.Triggers
+
 	tasksMtx sync.Mutex
 	tasks    activeSideTasks
 }
@@ -90,7 +93,7 @@ type activeMode interface {
 	SenderReceiver() (logic.Sender, logic.Receiver)
 	Type() Type
 	PlannerPolicy() logic.PlannerPolicy
-	RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{})
+	RunPeriodic(ctx context.Context, wakeReplication *trigger.Manual)
 	SnapperReport() *snapper.Report
 	ResetConnectBackoff()
 }
@@ -132,8 +135,8 @@ func (m *modePush) Type() Type { return TypePush }
 
 func (m *modePush) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
-	m.snapper.Run(ctx, wakeUpCommon)
+func (m *modePush) RunPeriodic(ctx context.Context, trigger *trigger.Manual) {
+	m.snapper.Run(ctx, trigger)
 }
 
 func (m *modePush) SnapperReport() *snapper.Report {
@@ -221,7 +224,7 @@ func (*modePull) Type() Type { return TypePull }
 
 func (m *modePull) PlannerPolicy() logic.PlannerPolicy { return *m.plannerPolicy }
 
-func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
+func (m *modePull) RunPeriodic(ctx context.Context, wakeReplication *trigger.Manual) {
 	if m.interval.Manual {
 		GetLogger(ctx).Info("manual pull configured, periodic pull disabled")
 		// "waiting for wakeups" is printed in common ActiveSide.do
@@ -232,14 +235,7 @@ func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}
 	for {
 		select {
 		case <-t.C:
-			select {
-			case wakeUpCommon <- struct{}{}:
-			default:
-				GetLogger(ctx).
-					WithField("pull_interval", m.interval).
-					Warn("pull job took longer than pull interval")
-				wakeUpCommon <- struct{}{} // block anyways, to queue up the wakeup
-			}
+			wakeReplication.Fire()
 		case <-ctx.Done():
 			return
 		}
@@ -370,6 +366,11 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}, p
 		return nil, errors.Wrap(err, "cannot build replication driver config")
 	}
 
+	j.triggers, err = trigger.FromConfig(in.Replication.Triggers)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build triggers")
+	}
+
 	return j, nil
 }
 
@@ -444,12 +445,17 @@ func (j *ActiveSide) Run(ctx context.Context) {
 
 	defer log.Info("job exiting")
 
-	periodicDone := make(chan struct{})
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	periodicCtx, endTask := trace.WithTask(ctx, "periodic")
+
+	periodCtx, endTask := trace.WithTask(ctx, "periodic")
+	defer endTask()
+	go j.mode.RunPeriodic(periodCtx, periodicTrigger)
+
+	wakeupTrigger := wakeup.Trigger(ctx)
+
+	triggered, endTask := j.triggers.Spawn(ctx, []*trigger.Trigger{periodicTrigger, wakeupTrigger})
 	defer endTask()
-	go j.mode.RunPeriodic(periodicCtx, periodicDone)
 
 	invocationCount := 0
 outer:
@@ -459,10 +465,15 @@ outer:
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context")
 			break outer
-
-		case <-wakeup.Wait(ctx):
-			j.mode.ResetConnectBackoff()
-		case <-periodicDone:
+		case trigger := <-triggered:
+			log :=
+				log.WithField("trigger_id", trigger.ID())
+			log.Info("triggered")
+			switch trigger {
+			case wakeupTrigger:
+				log.Info("trigger is wakeup command, resetting connection backoff")
+				j.mode.ResetConnectBackoff()
+			}
 		}
 		invocationCount++
 		invocationCtx, endSpan := trace.WithSpan(ctx, fmt.Sprintf("invocation-%d", invocationCount))

daemon/job/build_jobs.go

@@ -24,21 +24,6 @@ func JobsFromConfig(c *config.Config, parseFlags config.ParseFlags) ([]Job, erro
 		js[i] = j
 	}
 
-	// receiving-side root filesystems must not overlap
-	{
-		rfss := make([]string, 0, len(js))
-		for _, j := range js {
-			jrfs, ok := j.OwnedDatasetSubtreeRoot()
-			if !ok {
-				continue
-			}
-			rfss = append(rfss, jrfs.ToString())
-		}
-		if err := validateReceivingSidesDoNotOverlap(rfss); err != nil {
-			return nil, err
-		}
-	}
-
 	return js, nil
 }
 

daemon/job/snapjob.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 	"github.com/zrepl/zrepl/daemon/job/wakeup"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
@@ -104,12 +105,18 @@ func (j *SnapJob) Run(ctx context.Context) {
 
 	defer log.Info("job exiting")
 
-	periodicDone := make(chan struct{})
+	wakeupTrigger := wakeup.Trigger(ctx)
+
+	snapshottingTrigger := trigger.New("periodic")
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 	periodicCtx, endTask := trace.WithTask(ctx, "snapshotting")
 	defer endTask()
-	go j.snapper.Run(periodicCtx, periodicDone)
+	go j.snapper.Run(periodicCtx, snapshottingTrigger)
+
+	triggers := trigger.Empty()
+	triggered, endTask := triggers.Spawn(ctx, []trigger.Trigger{snapshottingTrigger, wakeupTrigger})
+	defer endTask()
 
 	invocationCount := 0
 outer:
@@ -119,9 +126,7 @@ outer:
 		case <-ctx.Done():
 			log.WithError(ctx.Err()).Info("context")
 			break outer
-
-		case <-wakeup.Wait(ctx):
-		case <-periodicDone:
+		case <-triggered:
 		}
 		invocationCount++
 

daemon/job/trigger/cron.go

@@ -0,0 +1,23 @@
+package trigger
+
+import (
+	"context"
+
+	"github.com/robfig/cron/v3"
+)
+
+type Cron struct {
+	spec cron.Schedule
+}
+
+var _ Trigger = &Cron{}
+
+func NewCron(spec cron.Schedule) *Cron {
+	return &Cron{spec: spec}
+}
+
+func (t *Cron) ID() string { return "cron" }
+
+func (t *Cron) run(ctx context.Context, signal chan<- struct{}) {
+	panic("unimpl: extract from cron snapper")
+}

daemon/job/trigger/from_config.go

@@ -0,0 +1,30 @@
+package trigger
+
+import (
+	"fmt"
+
+	"github.com/zrepl/zrepl/config"
+)
+
+func FromConfig(in []*config.ReplicationTriggerEnum) (*Triggers, error) {
+	triggers := make([]Trigger, len(in))
+	for i, e := range in {
+		var t Trigger = nil
+		switch te := e.Ret.(type) {
+		case *config.ReplicationTriggerManual:
+			// not a trigger
+			t = NewManual("manual")
+		case *config.ReplicationTriggerPeriodic:
+			t = NewPeriodic(te.Interval.Duration())
+		case *config.ReplicationTriggerCron:
+			t = NewCron(te.Cron.Schedule)
+		default:
+			return nil, fmt.Errorf("unknown trigger type %T", te)
+		}
+		triggers[i] = t
+	}
+	return &Triggers{
+		spawned:  false,
+		triggers: triggers,
+	}, nil
+}

daemon/job/trigger/logger.go

@@ -0,0 +1,12 @@
+package trigger
+
+import (
+	"context"
+
+	"github.com/zrepl/zrepl/logger"
+)
+
+
+func getLogger(ctx context.Context) logger.Logger {
+	panic("unimpl")
+}

daemon/job/trigger/manual.go

@@ -0,0 +1,33 @@
+package trigger
+
+import "context"
+
+type Manual struct {
+	id     string
+	signal chan<- struct{}
+}
+
+var _ Trigger = &Manual{}
+
+func NewManual(id string) *Manual {
+	return &Manual{
+		id:     id,
+		signal: nil,
+	}
+}
+
+func (t *Manual) ID() string {
+	return t.id
+}
+
+func (t *Manual) run(ctx context.Context, signal chan<- struct{}) {
+	if t.signal != nil {
+		panic("run must only be called once")
+	}
+	t.signal = signal
+}
+
+// Panics if called before the trigger has been spanwed as part of a `Triggers`.
+func (t *Manual) Fire() {
+	t.signal <- struct{}{}
+}

daemon/job/trigger/periodic.go

@@ -0,0 +1,33 @@
+package trigger
+
+import (
+	"context"
+	"time"
+)
+
+type Periodic struct {
+	interval time.Duration
+}
+
+var _ Trigger = &Periodic{}
+
+func NewPeriodic(interval time.Duration) *Periodic {
+	return &Periodic{
+		interval: interval,
+	}
+}
+
+func (p *Periodic) ID() string { return "periodic" }
+
+func (p *Periodic) run(ctx context.Context, signal chan<- struct{}) {
+	t := time.NewTicker(p.interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-t.C:
+			signal <- struct{}{}
+		case <-ctx.Done():
+			return
+		}
+	}
+}

daemon/job/trigger/trigger.go

@@ -0,0 +1,91 @@
+package trigger
+
+import (
+	"context"
+	"github.com/zrepl/zrepl/daemon/logging/trace"
+)
+
+type Triggers struct {
+	spawned  bool
+	triggers []Trigger
+}
+
+type Trigger interface {
+	ID() string
+	run(context.Context, chan<- struct{})
+}
+
+func Empty() *Triggers {
+	return &Triggers{
+		spawned:  false,
+		triggers: nil,
+	}
+}
+
+func (t *Triggers) Spawn(ctx context.Context, additionalTriggers []Trigger) (chan Trigger, trace.DoneFunc) {
+	if t.spawned {
+		panic("must only spawn once")
+	}
+	t.spawned = true
+	t.triggers = append(t.triggers, additionalTriggers...)
+	sink := make(chan Trigger)
+	endTask := t.spawn(ctx, sink)
+	return sink, endTask
+}
+
+type triggering struct {
+	trigger Trigger
+	handled chan struct{}
+}
+
+func (t *Triggers) spawn(ctx context.Context, sink chan Trigger) trace.DoneFunc {
+	ctx, endTask := trace.WithTask(ctx, "triggers")
+	ctx, add, wait := trace.WithTaskGroup(ctx, "trigger-tasks")
+	triggered := make(chan triggering, len(t.triggers))
+	for _, t := range t.triggers {
+		t := t
+		signal := make(chan struct{})
+		go add(func(ctx context.Context) {
+			t.run(ctx, signal)
+		})
+		go func() {
+			for {
+				select {
+				case <-ctx.Done():
+					return
+				case <-signal:
+					handled := make(chan struct{})
+					select {
+					case triggered <- triggering{trigger: t, handled: handled}:
+					default:
+						panic("this funtion ensures that there's always room in the channel")
+					}
+					select {
+					case <-handled:
+					case <-ctx.Done():
+						return
+					}
+				}
+			}
+		}()
+	}
+	go func() {
+		defer wait()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case triggering := <-triggered:
+				select {
+				case sink <- triggering.trigger:
+				default:
+					getLogger(ctx).
+						WithField("trigger_id", triggering.trigger.ID()).
+						Warn("dropping triggering because job is busy")
+				}
+				close(triggering.handled)
+			}
+		}
+	}()
+	return endTask
+}

daemon/job/wakeup/wakeup.go

@@ -3,6 +3,8 @@ package wakeup
 import (
 	"context"
 	"errors"
+
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 )
 
 type contextKey int
@@ -17,6 +19,10 @@ func Wait(ctx context.Context) <-chan struct{} {
 	return wc
 }
 
+func Trigger(ctx context.Context) trigger.Trigger {
+	panic("unimpl")
+}
+
 type Func func() error
 
 var AlreadyWokenUp = errors.New("already woken up")

daemon/logging/build_logging.go

@@ -63,6 +63,7 @@ type Subsystem string
 const (
 	SubsysMeta         Subsystem = "meta"
 	SubsysJob          Subsystem = "job"
+	SubsysTrigger      Subsystem = "trigger"
 	SubsysReplication  Subsystem = "repl"
 	SubsysEndpoint     Subsystem = "endpoint"
 	SubsysPruning      Subsystem = "pruning"

daemon/snapper/cron.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/hooks"
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 	"github.com/zrepl/zrepl/util/suspendresumesafetimer"
 	"github.com/zrepl/zrepl/zfs"
 )
@@ -42,7 +43,7 @@ type Cron struct {
 	wakeupWhileRunningCount int
 }
 
-func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
+func (s *Cron) Run(ctx context.Context, snapshotsTaken *trigger.Manual) {
 
 	for {
 		now := time.Now()
@@ -75,13 +76,7 @@ func (s *Cron) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
 			s.running = false
 			s.mtx.Unlock()
 
-			select {
-			case snapshotsTaken <- struct{}{}:
-			default:
-				if snapshotsTaken != nil {
-					getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
-				}
-			}
+			snapshotsTaken.Fire()
 		}()
 	}
 

daemon/snapper/manual.go

@@ -2,11 +2,13 @@ package snapper
 
 import (
 	"context"
+
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 )
 
 type manual struct{}
 
-func (s *manual) Run(ctx context.Context, wakeUpCommon chan<- struct{}) {
+func (s *manual) Run(ctx context.Context, snapshotsTaken *trigger.Manual) {
 	// nothing to do
 }
 

daemon/snapper/periodic.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/pkg/errors"
 
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 	"github.com/zrepl/zrepl/daemon/logging/trace"
 
 	"github.com/zrepl/zrepl/config"
@@ -51,7 +52,7 @@ type periodicArgs struct {
 	interval       time.Duration
 	fsf            zfs.DatasetFilter
 	planArgs       planArgs
-	snapshotsTaken chan<- struct{}
+	snapshotsTaken *trigger.Manual
 	dryRun         bool
 }
 
@@ -103,7 +104,7 @@ func (s State) sf() state {
 type updater func(u func(*Periodic)) State
 type state func(a periodicArgs, u updater) state
 
-func (s *Periodic) Run(ctx context.Context, snapshotsTaken chan<- struct{}) {
+func (s *Periodic) Run(ctx context.Context, snapshotsTaken *trigger.Manual) {
 	defer trace.WithSpanFromStackUpdateCtx(&ctx)()
 	getLogger(ctx).Debug("start")
 	defer getLogger(ctx).Debug("stop")
@@ -207,13 +208,7 @@ func periodicStateSnapshot(a periodicArgs, u updater) state {
 
 	ok := plan.execute(a.ctx, false)
 
-	select {
-	case a.snapshotsTaken <- struct{}{}:
-	default:
-		if a.snapshotsTaken != nil {
-			getLogger(a.ctx).Warn("callback channel is full, discarding snapshot update event")
-		}
-	}
+	a.snapshotsTaken.Fire()
 
 	return u(func(snapper *Periodic) {
 		if !ok {

daemon/snapper/snapper.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/job/trigger"
 	"github.com/zrepl/zrepl/zfs"
 )
 
@@ -17,7 +18,7 @@ const (
 )
 
 type Snapper interface {
-	Run(ctx context.Context, snapshotsTaken chan<- struct{})
+	Run(ctx context.Context, snapshotsTaken *trigger.Manual)
 	Report() Report
 }
 

docs/configuration/overview.rst

@@ -130,7 +130,7 @@ The following high-level steps take place during replication and can be monitore
   * Move the **replication cursor** bookmark on the sending side (see below).
   * Move the **last-received-hold** on the receiving side (see below).
   * Release the send-side step-holds.
-   
+
 The idea behind the execution order of replication steps is that if the sender snapshots all filesystems simultaneously at fixed intervals, the receiver will have all filesystems snapshotted at time ``T1`` before the first snapshot at ``T2 = T1 + $interval`` is replicated.
 
 ZFS Background Knowledge
@@ -258,6 +258,9 @@ On your setup, ensure that
 
 * all ``filesystems`` filter specifications are disjoint
 * no ``root_fs`` is a prefix or equal to another ``root_fs``
+
+  * For ``sink`` jobs, consider all possible ``root_fs/${client_identity}``.
+
 * no ``filesystems`` filter matches any ``root_fs``
 
 **Exceptions to the rule**:

pruning/keep_last_n.go

@@ -33,11 +33,6 @@ func NewKeepLastN(n int, regex string) (*KeepLastN, error) {
 }
 
 func (k KeepLastN) KeepRule(snaps []Snapshot) (destroyList []Snapshot) {
-
-	if k.n > len(snaps) {
-		return []Snapshot{}
-	}
-
 	matching, notMatching := partitionSnapList(snaps, func(snapshot Snapshot) bool {
 		return k.re.MatchString(snapshot.Name())
 	})

pruning/keep_last_n_test.go

@@ -90,7 +90,7 @@ func TestKeepLastN(t *testing.T) {
 				stubSnap{"a2", false, o(12)},
 			},
 			rules: []KeepRule{
-				MustKeepLastN(3, "a"),
+				MustKeepLastN(4, "a"),
 			},
 			expDestroy: map[string]bool{
 				"b1": true,