From 4ee00091d6ec479ac9ba60e54773486fe36d2959 Mon Sep 17 00:00:00 2001
From: Christian Schwarz <me@cschwarz.com>
Date: Sat, 16 Mar 2019 14:24:05 +0100
Subject: [PATCH] pull job: support manual-only invocation

---
 config/config.go                              | 33 ++++++++++++++-
 .../config_positiveintervalormanual_test.go   | 41 +++++++++++++++++++
 daemon/job/active.go                          | 12 +++---
 docs/configuration/jobs.rst                   |  3 +-
 docs/usage.rst                                |  2 +
 5 files changed, 84 insertions(+), 7 deletions(-)
 create mode 100644 config/config_positiveintervalormanual_test.go

diff --git a/config/config.go b/config/config.go
index 3c2131a..c17240f 100644
--- a/config/config.go
+++ b/config/config.go
@@ -67,7 +67,38 @@ type PushJob struct {
 type PullJob struct {
 	ActiveJob `yaml:",inline"`
 	RootFS    string        `yaml:"root_fs"`
-	Interval  time.Duration `yaml:"interval,positive"`
+	Interval  PositiveDurationOrManual `yaml:"interval"`
+}
+
+type PositiveDurationOrManual struct {
+	Interval time.Duration
+	Manual   bool
+}
+
+var _ yaml.Unmarshaler = (*PositiveDurationOrManual)(nil)
+
+func (i *PositiveDurationOrManual) UnmarshalYAML(u func(interface{}, bool) error) (err error) {
+	var s string
+	if err := u(&s, true); err != nil {
+		return err
+	}
+	switch s {
+	case "manual":
+		i.Manual = true
+		i.Interval = 0
+	case "":
+		return fmt.Errorf("value must not be empty")
+	default:
+		i.Manual = false
+		i.Interval, err = time.ParseDuration(s)
+		if err != nil {
+			return err
+		}
+		if i.Interval <= 0 {
+			return fmt.Errorf("value must be a positive duration, got %q", s)
+		}
+	}
+	return nil
 }
 
 type SinkJob struct {
diff --git a/config/config_positiveintervalormanual_test.go b/config/config_positiveintervalormanual_test.go
new file mode 100644
index 0000000..237813e
--- /dev/null
+++ b/config/config_positiveintervalormanual_test.go
@@ -0,0 +1,41 @@
+package config
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/zrepl/yaml-config"
+)
+
+func TestPositiveDurationOrManual(t *testing.T) {
+	cases := []struct {
+		Comment, Input string
+		Result         *PositiveDurationOrManual
+	}{
+		{"empty is error", "", nil},
+		{"negative is error", "-1s", nil},
+		{"zero seconds is error", "0s", nil},
+		{"zero is error", "0", nil},
+		{"non-manual is error", "something", nil},
+		{"positive seconds works", "1s", &PositiveDurationOrManual{Manual: false, Interval: 1 * time.Second}},
+		{"manual works", "manual", &PositiveDurationOrManual{Manual: true, Interval: 0}},
+	}
+	for _, tc := range cases {
+		t.Run(tc.Comment, func(t *testing.T) {
+			var out struct {
+				FieldName PositiveDurationOrManual `yaml:"fieldname"`
+			}
+			input := fmt.Sprintf("\nfieldname: %s\n", tc.Input)
+			err := yaml.UnmarshalStrict([]byte(input), &out)
+			if tc.Result == nil {
+				assert.Error(t, err)
+				t.Logf("%#v", out)
+			} else {
+				assert.Equal(t, *tc.Result, out.FieldName)
+			}
+		})
+	}
+
+}
diff --git a/daemon/job/active.go b/daemon/job/active.go
index 1341d08..d70184f 100644
--- a/daemon/job/active.go
+++ b/daemon/job/active.go
@@ -153,7 +153,7 @@ type modePull struct {
 	receiver *endpoint.Receiver
 	sender   *rpc.Client
 	rootFS   *zfs.DatasetPath
-	interval time.Duration
+	interval config.PositiveDurationOrManual
 }
 
 func (m *modePull) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
@@ -183,7 +183,12 @@ func (m *modePull) SenderReceiver() (logic.Sender, logic.Receiver) {
 func (*modePull) Type() Type { return TypePull }
 
 func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
-	t := time.NewTicker(m.interval)
+	if m.interval.Manual {
+		GetLogger(ctx).Info("manual pull configured, periodic pull disabled")
+		// "waiting for wakeups" is printed in common ActiveSide.do
+		return
+	}
+	t := time.NewTicker(m.interval.Interval)
 	defer t.Stop()
 	for {
 		select {
@@ -212,9 +217,6 @@ func (m *modePull) ResetConnectBackoff() {
 
 func modePullFromConfig(g *config.Global, in *config.PullJob) (m *modePull, err error) {
 	m = &modePull{}
-	if in.Interval <= 0 {
-		return nil, errors.New("interval must be positive")
-	}
 	m.interval = in.Interval
 
 	m.rootFS, err = zfs.NewDatasetPath(in.RootFS)
diff --git a/docs/configuration/jobs.rst b/docs/configuration/jobs.rst
index 077784f..15911ca 100644
--- a/docs/configuration/jobs.rst
+++ b/docs/configuration/jobs.rst
@@ -225,7 +225,8 @@ Job Type ``pull``
       - ZFS dataset path are received to
         ``$root_fs/$client_identity``
     * - ``interval``
-      - Interval at which to pull from the source job
+      - | Interval at which to pull from the source job (e.g. ``10m``).
+        | ``manual`` disables periodic pulling, replication then only happens on :ref:`wakeup <cli-signal-wakeup>`.
     * - ``pruning``
       - |pruning-spec|
 
diff --git a/docs/usage.rst b/docs/usage.rst
index 9d086a0..4bb2f76 100644
--- a/docs/usage.rst
+++ b/docs/usage.rst
@@ -13,6 +13,8 @@ CLI Overview
     The zrepl binary is self-documenting:
     run ``zrepl help`` for an overview of the available subcommands or ``zrepl SUBCOMMAND --help`` for information on available flags, etc.
 
+.. _cli-signal-wakeup:
+
 .. list-table::
     :widths: 30 70
     :header-rows: 1