hook up per repository router and elector

This adds a config option to enable the per repository elector and
router. When the `per_repository` elector is configured, the per
repository primary elector and router are enabled and node manager
is disabled.

The node manager is still created and passed as a dependency in
Praefect's info and server serivces as they haven't been updated to
support repository specific primaries yet.

While the configuration option is added, there are still follow ups
to address before the components are safe to use in production. For
example, creating repositories is not supported yet until a follow
up issue is addressed. The changes introduced here are useful for
continued local development repository specific primaries and features
dependent on them such as the variable replication factor.

1 files changed, 45 insertions, 4 deletions

diff --git a/internal/praefect/node_test.go b/internal/praefect/node_test.go
index 8df700c50..cfd856553 100644
--- a/internal/praefect/node_test.go
+++ b/internal/praefect/node_test.go
@@ -10,8 +10,13 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"gitlab.com/gitlab-org/gitaly/internal/praefect/config"
+	"gitlab.com/gitlab-org/gitaly/internal/praefect/nodes"
 	"gitlab.com/gitlab-org/gitaly/internal/testhelper"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/health"
+	"google.golang.org/grpc/health/grpc_health_v1"
+	"google.golang.org/grpc/status"
 )
 
 func TestDialNodes(t *testing.T) {
@@ -25,16 +30,20 @@ func TestDialNodes(t *testing.T) {
 	type nodeAssertion struct {
 		storage string
 		token   string
+		status  grpc_health_v1.HealthCheckResponse_ServingStatus
+		error   error
 	}
 
 	expectedNodes := []nodeAssertion{
 		{
 			storage: "healthy",
 			token:   "healthy-token",
+			status:  grpc_health_v1.HealthCheckResponse_SERVING,
 		},
 		{
 			storage: "unhealthy",
 			token:   "unhealthy-token",
+			status:  grpc_health_v1.HealthCheckResponse_NOT_SERVING,
 		},
 	}
 
@@ -43,7 +52,10 @@ func TestDialNodes(t *testing.T) {
 		socket := filepath.Join(tmp, n.storage)
 		ln, err := net.Listen("unix", socket)
 		require.NoError(t, err)
+		healthSrv := health.NewServer()
+		healthSrv.SetServingStatus("", n.status)
 		srv := grpc.NewServer()
+		grpc_health_v1.RegisterHealthServer(srv, healthSrv)
 		defer srv.Stop()
 		go srv.Serve(ln)
 
@@ -54,21 +66,50 @@ func TestDialNodes(t *testing.T) {
 		})
 	}
 
+	expectedNodes = append(expectedNodes, nodeAssertion{
+		storage: "invalid",
+		error:   status.Error(codes.Unavailable, `all SubConns are in TransientFailure, latest connection error: connection error: desc = "transport: Error while dialing dial unix non-existent-socket: connect: no such file or directory"`),
+	})
+
 	nodeSet, err := DialNodes(ctx,
-		[]*config.VirtualStorage{{Name: "virtual-storage", Nodes: cfgNodes}}, nil, nil,
+		[]*config.VirtualStorage{{
+			Name: "virtual-storage",
+			Nodes: append(cfgNodes, &config.Node{
+				Storage: "invalid",
+				Address: "unix:non-existent-socket",
+			}),
+		}}, nil, nil,
 	)
 	require.NoError(t, err)
 	defer nodeSet.Close()
 
+	conns := nodeSet.Connections()
+	healthClients := nodeSet.HealthClients()
+
 	var actualNodes []nodeAssertion
-	for _, nodes := range nodeSet {
+	for virtualStorage, nodes := range nodeSet {
 		for _, node := range nodes {
-			actualNodes = append(actualNodes, nodeAssertion{
+			require.NotNil(t, conns[virtualStorage][node.Storage], "connection not found for storage %q", node.Storage)
+			resp, err := healthClients[virtualStorage][node.Storage].Check(ctx, &grpc_health_v1.HealthCheckRequest{})
+
+			assertion := nodeAssertion{
 				storage: node.Storage,
 				token:   node.Token,
-			})
+				error:   err,
+			}
+
+			if resp != nil {
+				assertion.status = resp.Status
+			}
+
+			actualNodes = append(actualNodes, assertion)
+
+			delete(conns[virtualStorage], node.Storage)
+			delete(healthClients[virtualStorage], node.Storage)
 		}
 	}
 
 	require.ElementsMatch(t, expectedNodes, actualNodes)
+	require.Equal(t, Connections{"virtual-storage": {}}, conns, "unexpected connections")
+	require.Equal(t, nodes.HealthClients{"virtual-storage": {}}, healthClients, "unexpected health clients")
 }