WIP

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{}) {
+func (m *modePush) RunPeriodic(ctx context.Context, trigger *trigger.Manual) {
-	m.snapper.Run(ctx, wakeUpCommon)
+	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 {
+			wakeReplication.Fire()
-			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
-			}
 		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 trigger := <-triggered:
-		case <-wakeup.Wait(ctx):
+			log :=
-			j.mode.ResetConnectBackoff()
+				log.WithField("trigger_id", trigger.ID())
-		case <-periodicDone:
+			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/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 <-triggered:
-		case <-wakeup.Wait(ctx):
-		case <-periodicDone:
 		}
 		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 {
+			snapshotsTaken.Fire()
-			case snapshotsTaken <- struct{}{}:
-			default:
-				if snapshotsTaken != nil {
-					getLogger(ctx).Warn("callback channel is full, discarding snapshot update event")
-				}
-			}
 		}()
 	}
 

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 {
+	a.snapshotsTaken.Fire()
-	case a.snapshotsTaken <- struct{}{}:
-	default:
-		if a.snapshotsTaken != nil {
-			getLogger(a.ctx).Warn("callback channel is full, discarding snapshot update event")
-		}
-	}
 
 	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
 }