Use Yamux configuration in Praefect by default

`praefect_use_yamux_configuration_for_gitaly` feature flag enables
Praefect to read Yamux configuration from the config file. This
commit removes the feature flag and enables the new default settings
that were tested behind it. We're skipping default enabling the flag
as it requires two connections sets and the feature flagged connections
are not used for all of the functionality like elections. To avoid this
complexity, we'll just remove the flag directly and enable the new
defaults.

The new defaults lower the receive buffer size in Yamux leading to less
memory usage but possibly worse throughput in high latency scenarios.
We were previously using custom settings but are now returning to the
Yamux defaults as they are also used in the clients for the sidechannels.
This ensures Praefect doesn't buffer excessively with sidechannels in use.

Changelog: changed

1 files changed, 34 insertions, 63 deletions

diff --git a/cmd/praefect/main.go b/cmd/praefect/main.go
index 1fef8b455..540d7e982 100644
--- a/cmd/praefect/main.go
+++ b/cmd/praefect/main.go
@@ -79,7 +79,6 @@ import (
 	"gitlab.com/gitlab-org/gitaly/v15/internal/gitaly/config/sentry"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/helper"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/log"
-	"gitlab.com/gitlab-org/gitaly/v15/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/praefect"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/praefect/config"
 	"gitlab.com/gitlab-org/gitaly/v15/internal/praefect/datastore"
@@ -311,15 +310,15 @@ func run(
 	transactionManager := transactions.NewManager(conf)
 	sidechannelRegistry := sidechannel.NewRegistry()
 
-	newClientHandshaker := func(config backchannel.Configuration) backchannel.ClientHandshaker {
-		return backchannel.NewClientHandshaker(
-			logger,
-			praefect.NewBackchannelServerFactory(logger, transaction.NewServer(transactionManager), sidechannelRegistry),
-			config,
-		)
-	}
+	backchannelCfg := backchannel.DefaultConfiguration()
+	backchannelCfg.AcceptBacklog = conf.Yamux.AcceptBacklog
+	backchannelCfg.MaximumStreamWindowSizeBytes = conf.Yamux.MaximumStreamWindowSizeBytes
+	clientHandshaker := backchannel.NewClientHandshaker(
+		logger,
+		praefect.NewBackchannelServerFactory(logger, transaction.NewServer(transactionManager), sidechannelRegistry),
+		backchannelCfg,
+	)
 
-	clientHandshaker := newClientHandshaker(backchannel.DefaultConfiguration())
 	assignmentStore := praefect.NewDisabledAssignmentStore(conf.StorageNames())
 	var (
 		nodeManager   nodes.Manager
@@ -329,74 +328,46 @@ func run(
 		primaryGetter praefect.PrimaryGetter
 	)
 	if conf.Failover.ElectionStrategy == config.ElectionStrategyPerRepository {
-		dialNodes := func(clientHandshaker backchannel.ClientHandshaker) (praefect.NodeSet, error) {
-			return praefect.DialNodes(ctx, conf.VirtualStorages, protoregistry.GitalyProtoPreregistered, errTracker, clientHandshaker, sidechannelRegistry)
-		}
-
-		nodeSet, err = dialNodes(clientHandshaker)
+		nodeSet, err = praefect.DialNodes(
+			ctx,
+			conf.VirtualStorages,
+			protoregistry.GitalyProtoPreregistered,
+			errTracker,
+			clientHandshaker,
+			sidechannelRegistry,
+		)
 		if err != nil {
 			return fmt.Errorf("dial nodes: %w", err)
 		}
 		defer nodeSet.Close()
 
-		// Dial a second set of connections to each storage with a different Yamux config
-		// so we can use a feature flag to gradually test the new configuration.
-		nodeSetWithCustomConfig, err := dialNodes(newClientHandshaker(backchannel.Configuration{
-			AcceptBacklog:                conf.Yamux.AcceptBacklog,
-			MaximumStreamWindowSizeBytes: conf.Yamux.MaximumStreamWindowSizeBytes,
-			StreamCloseTimeout:           backchannel.DefaultConfiguration().StreamCloseTimeout,
-		}))
-		if err != nil {
-			return fmt.Errorf("dial nodes reduced window: %w", err)
-		}
-		defer nodeSetWithCustomConfig.Close()
-
-		runHealthManager := func(db glsql.Querier, nodeSet praefect.NodeSet) *nodes.HealthManager {
-			hm := nodes.NewHealthManager(logger, db, nodes.GeneratePraefectName(conf, logger), nodeSet.HealthClients())
-			go func() {
-				if err := hm.Run(ctx, helper.NewTimerTicker(time.Second)); err != nil {
-					logger.WithError(err).Error("health manager exited")
-				}
-			}()
-
-			return hm
-		}
-
-		nodeSetHealthManager := runHealthManager(db, nodeSet)
-		healthChecker = nodeSetHealthManager
+		healthManager := nodes.NewHealthManager(logger, db, nodes.GeneratePraefectName(conf, logger), nodeSet.HealthClients())
+		go func() {
+			if err := healthManager.Run(ctx, helper.NewTimerTicker(time.Second)); err != nil {
+				logger.WithError(err).Error("health manager exited")
+			}
+		}()
 
-		nodeSetWithCustomConfigHealthManager := runHealthManager(nil, nodeSetWithCustomConfig)
+		healthChecker = healthManager
 
 		// Wait for the first health check to complete so the Praefect doesn't start serving RPC
 		// before the router is ready with the health status of the nodes.
-		<-nodeSetHealthManager.Updated()
-		<-nodeSetWithCustomConfigHealthManager.Updated()
+		<-healthManager.Updated()
 
 		elector := nodes.NewPerRepositoryElector(db)
 
 		primaryGetter = elector
 		assignmentStore = datastore.NewAssignmentStore(db, conf.StorageNames())
 
-		newPerRepositoryRouter := func(healthManager *nodes.HealthManager, nodeSet praefect.NodeSet) *praefect.PerRepositoryRouter {
-			return praefect.NewPerRepositoryRouter(
-				nodeSet.Connections(),
-				elector,
-				healthManager,
-				praefect.NewLockedRandom(rand.New(rand.NewSource(time.Now().UnixNano()))),
-				csg,
-				assignmentStore,
-				rs,
-				conf.DefaultReplicationFactors(),
-			)
-		}
-
-		// Configure a router that uses a feature flag to switch between two different sets of connections.
-		// This way we can toggle the feature flag to use a different set of connections for proxied
-		// RPC calls.
-		router = praefect.NewFeatureFlaggedRouter(
-			featureflag.PraefectUseYamuxConfigurationForGitaly,
-			newPerRepositoryRouter(nodeSetWithCustomConfigHealthManager, nodeSetWithCustomConfig),
-			newPerRepositoryRouter(nodeSetHealthManager, nodeSet),
+		router = praefect.NewPerRepositoryRouter(
+			nodeSet.Connections(),
+			elector,
+			healthManager,
+			praefect.NewLockedRandom(rand.New(rand.NewSource(time.Now().UnixNano()))),
+			csg,
+			assignmentStore,
+			rs,
+			conf.DefaultReplicationFactors(),
 		)
 
 		if conf.BackgroundVerification.VerificationInterval > 0 {
@@ -405,7 +376,7 @@ func run(
 				logger,
 				db,
 				nodeSet.Connections(),
-				nodeSetHealthManager,
+				healthManager,
 				conf.BackgroundVerification.VerificationInterval.Duration(),
 				conf.BackgroundVerification.DeleteInvalidRecords,
 			)