diff options
| -rw-r--r-- | internal/praefect/coordinator_test.go | 6 | ||||
| -rw-r--r-- | internal/praefect/nodes/disabled_elector.go | 24 | ||||
| -rw-r--r-- | internal/praefect/nodes/manager.go | 7 |
3 files changed, 28 insertions, 9 deletions
diff --git a/internal/praefect/coordinator_test.go b/internal/praefect/coordinator_test.go index a58413009..ae38bdca9 100644 --- a/internal/praefect/coordinator_test.go +++ b/internal/praefect/coordinator_test.go @@ -909,6 +909,12 @@ func TestStreamDirector_repo_creation(t *testing.T) { ctx, cancel := testhelper.Context() defer cancel() + // NodeManager does not provide a way to get access to the muxed connections easily. The test does + // not set up the database, so GetShard fails. As such, let's just disable the feature flag in this + // test. Once the muxed connections are setup by default, the test will test them and the feature flag + // is removed. + ctx = featureflag.IncomingCtxWithDisabledFeatureFlag(ctx, featureflag.ConnectionMultiplexing) + peeker := &mockPeeker{frame} streamParams, err := coordinator.StreamDirector(correlation.ContextWithCorrelation(ctx, "my-correlation-id"), fullMethod, peeker) require.NoError(t, err) diff --git a/internal/praefect/nodes/disabled_elector.go b/internal/praefect/nodes/disabled_elector.go index 39ce88b3e..cc05ab60a 100644 --- a/internal/praefect/nodes/disabled_elector.go +++ b/internal/praefect/nodes/disabled_elector.go @@ -5,22 +5,28 @@ import ( "sync" "time" + "gitlab.com/gitlab-org/gitaly/internal/metadata/featureflag" "gitlab.com/gitlab-org/gitaly/internal/praefect/metrics" ) // newDisabledElector returns a stub that always returns the same shard where the // primary is the first node from the passed in list. -func newDisabledElector(virtualStorage string, ns []*nodeStatus) *disabledElector { +func newDisabledElector(virtualStorage string, ns []*nodeStatus, muxed muxedNodes) *disabledElector { secondaries := make([]Node, len(ns)-1) for i, node := range ns[1:] { secondaries[i] = node } - return &disabledElector{virtualStorage: virtualStorage, shard: Shard{Primary: ns[0], Secondaries: secondaries}} + return &disabledElector{ + virtualStorage: virtualStorage, + shard: Shard{Primary: ns[0], Secondaries: secondaries}, + muxed: muxed, + } } type disabledElector struct { shard Shard virtualStorage string + muxed muxedNodes } func (de *disabledElector) start(bootstrap, _ time.Duration) { @@ -57,10 +63,22 @@ func (de *disabledElector) checkNodes(ctx context.Context) error { return nil } -func (de *disabledElector) GetShard(context.Context) (Shard, error) { +func (de *disabledElector) GetShard(ctx context.Context) (Shard, error) { if !de.shard.Primary.IsHealthy() { return Shard{}, ErrPrimaryNotHealthy } + if featureflag.IsEnabled(ctx, featureflag.ConnectionMultiplexing) { + secondaries := make([]Node, len(de.shard.Secondaries)) + for i, secondary := range de.shard.Secondaries { + secondaries[i] = de.muxed.getNode(ctx, secondary) + } + + return Shard{ + Primary: de.muxed.getNode(ctx, de.shard.Primary), + Secondaries: secondaries, + }, nil + } + return de.shard, nil } diff --git a/internal/praefect/nodes/manager.go b/internal/praefect/nodes/manager.go index 1c9b4926b..29739d9fb 100644 --- a/internal/praefect/nodes/manager.go +++ b/internal/praefect/nodes/manager.go @@ -198,11 +198,6 @@ func NewManager( cs := newConnectionStatus(*node, conn, log, latencyHistogram, errorTracker) ns = append(ns, cs) - if c.Failover.ElectionStrategy != config.ElectionStrategySQL && - c.Failover.ElectionStrategy != config.ElectionStrategyLocal { - continue - } - muxedConn, err := Dial(ctx, node, registry, errorTracker, handshaker) if err != nil { log.WithError(err).Error("failed to dial Gitaly over a muxed connection") @@ -224,7 +219,7 @@ func NewManager( strategies[virtualStorage.Name] = newLocalElector(virtualStorage.Name, log, ns, vsMuxed) } } else { - strategies[virtualStorage.Name] = newDisabledElector(virtualStorage.Name, ns) + strategies[virtualStorage.Name] = newDisabledElector(virtualStorage.Name, ns, vsMuxed) } } |