replication: simplify parallel replication variables & expose them in config

closes #140

daemon/job/active.go

@@ -11,6 +11,7 @@ import (
 	"github.com/prometheus/common/log"
 
 	"github.com/zrepl/zrepl/daemon/logging/trace"
+	"github.com/zrepl/zrepl/util/envconst"
 
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job/reset"
@@ -33,6 +34,8 @@ type ActiveSide struct {
 	name      endpoint.JobID
 	connecter transport.Connecter
 
+	replicationDriverConfig driver.Config
+
 	prunerFactory *pruner.PrunerFactory
 
 	promRepStateSecs      *prometheus.HistogramVec // labels: state
@@ -159,8 +162,12 @@ func modePushFromConfig(g *config.Global, in *config.PushJob, jobID endpoint.Job
 	}
 
 	m.plannerPolicy = &logic.PlannerPolicy{
-		EncryptedSend:     logic.TriFromBool(in.Send.Encrypted),
-		ReplicationConfig: replicationConfig,
+		EncryptedSend:             logic.TriFromBool(in.Send.Encrypted),
+		ReplicationConfig:         replicationConfig,
+		SizeEstimationConcurrency: in.Replication.Concurrency.SizeEstimates,
+	}
+	if err := m.plannerPolicy.Validate(); err != nil {
+		return nil, errors.Wrap(err, "cannot build planner policy")
 	}
 
 	if m.snapper, err = snapper.FromConfig(g, m.senderConfig.FSF, in.Snapshotting); err != nil {
@@ -254,8 +261,12 @@ func modePullFromConfig(g *config.Global, in *config.PullJob, jobID endpoint.Job
 	}
 
 	m.plannerPolicy = &logic.PlannerPolicy{
-		EncryptedSend:     logic.DontCare,
-		ReplicationConfig: replicationConfig,
+		EncryptedSend:             logic.DontCare,
+		ReplicationConfig:         replicationConfig,
+		SizeEstimationConcurrency: in.Replication.Concurrency.SizeEstimates,
+	}
+	if err := m.plannerPolicy.Validate(); err != nil {
+		return nil, errors.Wrap(err, "cannot build planner policy")
 	}
 
 	m.receiverConfig, err = buildReceiverConfig(in, jobID)
@@ -266,6 +277,16 @@ func modePullFromConfig(g *config.Global, in *config.PullJob, jobID endpoint.Job
 	return m, nil
 }
 
+func replicationDriverConfigFromConfig(in *config.Replication) (c driver.Config, err error) {
+	c = driver.Config{
+		StepQueueConcurrency:     in.Concurrency.Steps,
+		MaxAttempts:              envconst.Int("ZREPL_REPLICATION_MAX_ATTEMPTS", 3),
+		ReconnectHardFailTimeout: envconst.Duration("ZREPL_REPLICATION_RECONNECT_HARD_FAIL_TIMEOUT", 10*time.Minute),
+	}
+	err = c.Validate()
+	return c, err
+}
+
 func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (j *ActiveSide, err error) {
 
 	j = &ActiveSide{}
@@ -326,6 +347,11 @@ func activeSide(g *config.Global, in *config.ActiveJob, configJob interface{}) (
 		return nil, err
 	}
 
+	j.replicationDriverConfig, err = replicationDriverConfigFromConfig(in.Replication)
+	if err != nil {
+		return nil, errors.Wrap(err, "cannot build replication driver config")
+	}
+
 	return j, nil
 }
 
@@ -459,7 +485,7 @@ func (j *ActiveSide) do(ctx context.Context) {
 			*tasks = activeSideTasks{}
 			tasks.replicationCancel = func() { repCancel(); endSpan() }
 			tasks.replicationReport, repWait = replication.Do(
-				ctx, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
+				ctx, j.replicationDriverConfig, logic.NewPlanner(j.promRepStateSecs, j.promBytesReplicated, sender, receiver, j.mode.PlannerPolicy()),
 			)
 			tasks.state = ActiveSideReplicating
 		})

daemon/job/build_jobs_test.go

@@ -144,3 +144,113 @@ func TestSampleConfigsAreBuiltWithoutErrors(t *testing.T) {
 	}
 
 }
+
+func TestReplicationOptions(t *testing.T) {
+	tmpl := `
+jobs:
+- name: foo
+  type: push
+  connect:
+    type: local
+    listener_name: foo
+    client_identity: bar
+  filesystems: {"<": true}
+  %s
+  snapshotting:
+    type: manual
+  pruning:
+    keep_sender:
+    - type: last_n
+      count: 10
+    keep_receiver:
+    - type: last_n
+      count: 10
+`
+
+	type Test struct {
+		name        string
+		input       string
+		expectOk    func(t *testing.T, a *ActiveSide, m *modePush)
+		expectError bool
+	}
+
+	tests := []Test{
+		{
+			name: "defaults",
+			input: `
+  replication: {}
+`,
+			expectOk: func(t *testing.T, a *ActiveSide, m *modePush) {},
+		},
+		{
+			name: "steps_zero",
+			input: `
+  replication:
+    concurrency:
+      steps: 0
+`,
+			expectError: true,
+		},
+		{
+			name: "size_estimates_zero",
+			input: `
+  replication:
+    concurrency:
+      size_estimates: 0
+`,
+			expectError: true,
+		},
+		{
+			name: "custom_values",
+			input: `
+  replication:
+    concurrency:
+      steps: 23
+      size_estimates: 42
+`,
+			expectOk: func(t *testing.T, a *ActiveSide, m *modePush) {
+				assert.Equal(t, 23, a.replicationDriverConfig.StepQueueConcurrency)
+				assert.Equal(t, 42, m.plannerPolicy.SizeEstimationConcurrency)
+			},
+		},
+		{
+			name: "negative_values_forbidden",
+			input: `
+  replication:
+    concurrency:
+      steps: -23
+      size_estimates: -42
+`,
+			expectError: true,
+		},
+	}
+
+	fill := func(s string) string { return fmt.Sprintf(tmpl, s) }
+
+	for _, ts := range tests {
+		t.Run(ts.name, func(t *testing.T) {
+			assert.True(t, (ts.expectError) != (ts.expectOk != nil))
+
+			cstr := fill(ts.input)
+			t.Logf("testing config:\n%s", cstr)
+			c, err := config.ParseConfigBytes([]byte(cstr))
+			require.NoError(t, err)
+			jobs, err := JobsFromConfig(c)
+			if ts.expectOk != nil {
+				require.NoError(t, err)
+				require.NotNil(t, c)
+				require.NoError(t, err)
+				require.Len(t, jobs, 1)
+				a := jobs[0].(*ActiveSide)
+				m := a.mode.(*modePush)
+				ts.expectOk(t, a, m)
+			} else if ts.expectError {
+				require.Error(t, err)
+			} else {
+				t.Fatalf("test must define expectOk or expectError")
+			}
+
+		})
+	}
+
+}