Merge #155150 #155154

155150: pgwire: limit the number of connection error logs r=shghasemi a=shghasemi

Rejecting new connections while server is draining printed too many logs, making it difficult to spot other errors during drain. Added a EveryN log to limit the log frequency to every 10 seconds.

Fixes #102339

Release note: None

155154: rpc: disable loopbackTransport via NoLoopbackDialer knob r=tbg a=tbg

Prior to this commit, the NoLoopbackDialer knob would replace the loopback dialer implementation with the regular TCP dialer. However, the loopbackTransport (which is what causes a call to the loopbackDialerFn in the first place) has its own call path and in particular, it sets different gRPC options that might be cheaper than what is commonly used (for example, no snappy compression). It is desirable to remove this discrepancy.

This commit disables use of the loopbackTransport entirely if the knob is set, the desired effect being that when talking to a TestCluster under this knob, it doesn't matter for performance whether the SQL gateway matches the location of the lease; local communication should be exactly as expensive as "remote" communication (all within the same process, but communicating through identical TCP network connections).

Unblocks #153865.

Epic: CRDB-55052

Co-authored-by: Shadi Ghasemitaheri <[email protected]>
Co-authored-by: Tobias Grieger <[email protected]>

Author: 3 people

pkg/rpc/context.go

@@ -55,7 +55,6 @@ import (
 	"google.golang.org/grpc/stats"
 	"storj.io/drpc"
 	"storj.io/drpc/drpcclient"
-	"storj.io/drpc/drpcmigrate"
 )
 
 // NewServer sets up an RPC server. Depending on the ServerOptions, the Server
@@ -608,12 +607,12 @@ func NewContext(ctx context.Context, opts ContextOptions) *Context {
 	if opts.Knobs.NoLoopbackDialer {
 		// The test has decided it doesn't need/want a loopback dialer.
 		// Ensure we still have a working dial function in that case.
+		var errAttemptToLoopbackDial = errors.AssertionFailedf("loopback dialer called but NoLoopbackDialer was set")
 		rpcCtx.loopbackDialFn = func(ctx context.Context) (net.Conn, error) {
-			d := onlyOnceDialer{}
-			return d.dial(ctx, opts.AdvertiseAddr)
+			return nil, errAttemptToLoopbackDial
 		}
 		rpcCtx.loopbackDRPCDialFn = func(ctx context.Context) (net.Conn, error) {
-			return drpcmigrate.DialWithHeader(ctx, "tcp", opts.AdvertiseAddr, drpcmigrate.DRPCHeader)
+			return nil, errAttemptToLoopbackDial
 		}
 	}
 
@@ -1403,13 +1402,17 @@ const (
 	tcpTransport transportType = true
 )
 
+func (rpcCtx *Context) canLoopbackDial() bool {
+	return !rpcCtx.ClientOnly && !rpcCtx.Knobs.NoLoopbackDialer
+}
+
 // GRPCDialOptions returns the minimal `grpc.DialOption`s necessary to connect
 // to a server.
 func (rpcCtx *Context) GRPCDialOptions(
 	ctx context.Context, target string, class rpcbase.ConnectionClass,
 ) ([]grpc.DialOption, error) {
 	transport := tcpTransport
-	if rpcCtx.ContextOptions.AdvertiseAddr == target && !rpcCtx.ClientOnly {
+	if rpcCtx.ContextOptions.AdvertiseAddr == target && rpcCtx.canLoopbackDial() {
 		// See the explanation on loopbackDialFn for an explanation about this.
 		transport = loopbackTransport
 	}
@@ -2042,7 +2045,7 @@ func (rpcCtx *Context) grpcDialRaw(
 	additionalOpts ...grpc.DialOption,
 ) (*grpc.ClientConn, error) {
 	transport := tcpTransport
-	if rpcCtx.ContextOptions.AdvertiseAddr == target && !rpcCtx.ClientOnly {
+	if rpcCtx.ContextOptions.AdvertiseAddr == target && rpcCtx.canLoopbackDial() {
 		// See the explanation on loopbackDialFn for an explanation about this.
 		transport = loopbackTransport
 	}

pkg/rpc/context_testutils.go

@@ -56,11 +56,16 @@ type ContextTestingKnobs struct {
 	// this value if non-nil at construction time.
 	StorageClusterID *uuid.UUID
 
-	// NoLoopbackDialer, when set, indicates that a test does not care
-	// about the special loopback dial semantics.
-	// If this is left unset, the test is responsible for ensuring
-	// SetLoopbackDialer() has been called on the rpc.Context.
-	// (This is done automatically by server.Server/server.SQLServerWrapper.)
+	// NoLoopbackDialer, when true, indicates that a test does not care about the
+	// special loopback dial semantics. In this case, an error results from use of
+	// the loopbackTransport or loopbackDialerFn (which would indicate a
+	// programming error in RPCContext, since NoLoopbackDialer is consulted when
+	// determining whether to use those).
+	//
+	// When false, SetLoopbackDialer() must have been called on the rpc.Context In
+	// productino code, this is done automatically by
+	// server.Server/server.SQLServerWrapper, but some lower-level tests may have
+	// to do it manually.
 	NoLoopbackDialer bool
 }
 

pkg/sql/pgwire/server.go

@@ -309,6 +309,8 @@ type Server struct {
 		// destinations tracks the metrics for each destination.
 		destinations map[string]*destinationMetrics
 	}
+	// limit the number of rejectNewConnection errors
+	connectionErrorLogEveryN log.EveryN
 
 	auth struct {
 		syncutil.RWMutex
@@ -490,6 +492,7 @@ func MakeServer(
 	server.mu.drainCh = make(chan struct{})
 	server.mu.destinations = make(map[string]*destinationMetrics)
 	server.mu.Unlock()
+	server.connectionErrorLogEveryN = log.Every(10 * time.Second)
 	executorConfig.CidrLookup.SetOnChange(server.onCidrChange)
 
 	connAuthConf.SetOnChange(&st.SV, func(ctx context.Context) {
@@ -931,7 +934,9 @@ func (s *Server) ServeConn(
 	st := s.execCfg.Settings
 	// If the server is shutting down, terminate the connection early.
 	if rejectNewConnections {
-		log.Ops.Info(ctx, "rejecting new connection while server is draining")
+		if s.connectionErrorLogEveryN.ShouldLog() {
+			log.Ops.Info(ctx, "rejecting new connection while server is draining")
+		}
 		return s.sendErr(ctx, st, conn, newAdminShutdownErr(ErrDrainingNewConn))
 	}