decouple reconciler from node manager

Reconciler currently depends on NodeManager to perform health checks.
This commit decouples the reconciler from node manager to allow for
alternative health checking implementations. Additionally, this reduces
the surface of the interface and allows reconciler to only depend on the
health information it needs without having to know about the primary.

This decoupling is necessary to hook up the per repository elector stack.

3 files changed, 20 insertions, 71 deletions

diff --git a/internal/praefect/reconciler/reconciler.go b/internal/praefect/reconciler/reconciler.go
index ec084cc6f..23b081eb3 100644
--- a/internal/praefect/reconciler/reconciler.go
+++ b/internal/praefect/reconciler/reconciler.go
@@ -8,15 +8,15 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/sirupsen/logrus"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore/glsql"
-	"gitlab.com/gitlab-org/gitaly/internal/praefect/nodes"
 )
 
 // Reconciler implements reconciliation logic for repairing outdated repository replicas.
 type Reconciler struct {
 	log                              logrus.FieldLogger
 	db                               glsql.Querier
-	nm                               nodes.Manager
+	hc                               praefect.HealthChecker
 	storages                         map[string][]string
 	reconciliationSchedulingDuration prometheus.Histogram
 	reconciliationJobsTotal          *prometheus.CounterVec
@@ -26,13 +26,13 @@ type Reconciler struct {
 }
 
 // NewReconciler returns a new Reconciler for repairing outdated repositories.
-func NewReconciler(log logrus.FieldLogger, db glsql.Querier, nm nodes.Manager, storages map[string][]string, buckets []float64) *Reconciler {
+func NewReconciler(log logrus.FieldLogger, db glsql.Querier, hc praefect.HealthChecker, storages map[string][]string, buckets []float64) *Reconciler {
 	log = log.WithField("component", "reconciler")
 
 	r := &Reconciler{
 		log:      log,
 		db:       db,
-		nm:       nm,
+		hc:       hc,
 		storages: storages,
 		reconciliationSchedulingDuration: prometheus.NewHistogram(prometheus.HistogramOpts{
 			Name:    "gitaly_praefect_reconciliation_scheduling_seconds",
@@ -110,35 +110,18 @@ func (r *Reconciler) reconcile(ctx context.Context) error {
 
 	var virtualStorages []string
 	var storages []string
-	for virtualStorage := range r.storages {
-		if len(r.storages[virtualStorage]) < 2 {
-			continue
-		}
-
-		shard, err := r.nm.GetShard(virtualStorage)
-		if err != nil {
-			r.log.WithFields(logrus.Fields{
-				"virtual_storage": virtualStorage,
-				"error":           err,
-			}).Error("unable to include virtual storage for reconciliation")
-			continue
-		}
-
-		healthyNodes := shard.GetHealthySecondaries()
-		if shard.Primary.IsHealthy() {
-			healthyNodes = append(healthyNodes, shard.Primary)
-		}
 
-		if len(healthyNodes) < 2 {
+	for virtualStorage, healthyStorages := range r.hc.HealthyNodes() {
+		if len(healthyStorages) < 2 {
 			// minimum two healthy storages within a virtual stoage needed for valid
 			// replication source and target
 			r.log.WithField("virtual_storage", virtualStorage).Info("reconciliation skipped for virtual storage due to not having enough healthy storages")
 			continue
 		}
 
-		for _, node := range healthyNodes {
+		for _, storage := range healthyStorages {
 			virtualStorages = append(virtualStorages, virtualStorage)
-			storages = append(storages, node.GetStorage())
+			storages = append(storages, storage)
 		}
 	}
 
diff --git a/internal/praefect/reconciler/reconciler_benchmark_test.go b/internal/praefect/reconciler/reconciler_benchmark_test.go
index 7ee0b1217..ec266c268 100644
--- a/internal/praefect/reconciler/reconciler_benchmark_test.go
+++ b/internal/praefect/reconciler/reconciler_benchmark_test.go
@@ -8,7 +8,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
-	"gitlab.com/gitlab-org/gitaly/internal/praefect/nodes"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 )
 
@@ -59,23 +59,14 @@ CROSS JOIN (SELECT unnest('{gitaly-1, gitaly-2, gitaly-3}'::text[]) AS storage)
 `, numRepositories, behind)
 	require.NoError(b, err)
 
+	storages := map[string][]string{"virtual-storage-1": {"gitaly-1", "gitaly-2", "gitaly-3"}}
 	for n := 0; n < b.N; n++ {
 		db.Truncate(b, "replication_queue", "replication_queue_lock", "replication_queue_job_lock")
 		r := NewReconciler(
 			testhelper.DiscardTestLogger(b),
 			db,
-			&nodes.MockManager{
-				GetShardFunc: func(virtualStorage string) (nodes.Shard, error) {
-					return nodes.Shard{
-						Primary: &nodes.MockNode{GetStorageMethod: getStorageMethod("gitaly-1"), Healthy: true},
-						Secondaries: []nodes.Node{
-							&nodes.MockNode{GetStorageMethod: getStorageMethod("gitaly-2"), Healthy: true},
-							&nodes.MockNode{GetStorageMethod: getStorageMethod("gitaly-3"), Healthy: true},
-						},
-					}, nil
-				},
-			},
-			map[string][]string{"virtual-storage-1": []string{"gitaly-1", "gitaly-2", "gitaly-3"}},
+			praefect.StaticHealthChecker(storages),
+			storages,
 			prometheus.DefBuckets,
 		)
 
diff --git a/internal/praefect/reconciler/reconciler_test.go b/internal/praefect/reconciler/reconciler_test.go
index 320967ed4..c72e3e153 100644
--- a/internal/praefect/reconciler/reconciler_test.go
+++ b/internal/praefect/reconciler/reconciler_test.go
@@ -12,9 +12,9 @@ import (
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/internal/middleware/metadatahandler"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/datastore/glsql"
-	"gitlab.com/gitlab-org/gitaly/internal/praefect/nodes"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 )
 
@@ -351,37 +351,6 @@ func TestReconciler(t *testing.T) {
 				}
 			}
 
-			nm := &nodes.MockManager{
-				GetShardFunc: func(virtualStorage string) (nodes.Shard, error) {
-					storages, ok := configuredStorages[virtualStorage]
-					if !ok {
-						t.Fatalf("unconfigured virtual storage: %s", virtualStorage)
-					}
-
-					shard := nodes.Shard{}
-					for i, storage := range storages {
-						isHealthy := false
-						for _, healthy := range tc.healthyStorages[virtualStorage] {
-							if healthy == storage {
-								isHealthy = true
-								break
-							}
-						}
-
-						node := &nodes.MockNode{GetStorageMethod: getStorageMethod(storage), Healthy: isHealthy}
-						if i == 0 {
-							// First node is picked to be the primary just to fill the shard object.
-							// Reconciliation logic does not differentiate between primaries and secondaries.
-							shard.Primary = node
-						} else {
-							shard.Secondaries = append(shard.Secondaries, node)
-						}
-					}
-
-					return shard, nil
-				},
-			}
-
 			runCtx, cancelRun := context.WithCancel(ctx)
 			var stopped, resetted bool
 			ticker := helper.NewManualTicker()
@@ -396,7 +365,13 @@ func TestReconciler(t *testing.T) {
 				ticker.Tick()
 			}
 
-			reconciler := NewReconciler(testhelper.DiscardTestLogger(t), db, nm, configuredStorages, prometheus.DefBuckets)
+			reconciler := NewReconciler(
+				testhelper.DiscardTestLogger(t),
+				db,
+				praefect.StaticHealthChecker(tc.healthyStorages),
+				configuredStorages,
+				prometheus.DefBuckets,
+			)
 			reconciler.handleError = func(err error) error { return err }
 
 			err := reconciler.Run(runCtx, ticker)