server: configurable RPC dial timeout (#27862)

Author: lukepalmer

.changelog/27862.txt

@@ -0,0 +1,3 @@
+```release-note:improvement
+server: RPC dial timeout is configurable
+```

client/client.go

@@ -386,7 +386,7 @@ func NewClient(cfg *config.Config, consulCatalog consul.CatalogAPI, consulProxie
 		consulProxiesFunc:    consulProxiesFunc,
 		consulServices:       consulServices,
 		start:                time.Now(),
-		connPool:             pool.NewPool(logger, clientRPCCache, clientMaxStreams, tlsWrap, cfg.RPCSessionConfig),
+		connPool:             pool.NewPool(logger, clientRPCCache, clientMaxStreams, tlsWrap, cfg.RPCSessionConfig, cfg.RPCDialTimeout),
 		tlsWrap:              tlsWrap,
 		streamingRpcs:        structs.NewStreamingRpcRegistry(),
 		logger:               logger,

client/config/config.go

@@ -285,6 +285,10 @@ type Config struct {
 	// RPCSessionConfig configures yamux multiplex
 	RPCSessionConfig *yamux.Config
 
+	// RPCDialTimeout is the timeout used when establishing new outbound RPC
+	// connections to servers. Defaults to 10s.
+	RPCDialTimeout time.Duration
+
 	// PluginLoader is used to load plugins.
 	PluginLoader loader.PluginCatalog
 
@@ -925,6 +929,7 @@ func DefaultConfig() *Config {
 		TemplateConfig:          DefaultTemplateConfig(),
 		RPCHoldTimeout:          5 * time.Second,
 		RPCSessionConfig:        yamux.DefaultConfig(),
+		RPCDialTimeout:          10 * time.Second,
 		CNIPath:                 DefaultCNIPath,
 		CNIConfigDir:            "/opt/cni/config",
 		CNIInterfacePrefix:      "eth",

client/rpc.go

@@ -221,7 +221,7 @@ func bridgedStreamingRpcHandler(sideA io.ReadWriteCloser) structs.StreamingRpcHa
 // streaming RPC.
 func (c *Client) streamingRpcConn(server *servers.Server, method string) (net.Conn, error) {
 	// Dial the server
-	conn, err := net.DialTimeout("tcp", server.Addr.String(), 10*time.Second)
+	conn, err := net.DialTimeout("tcp", server.Addr.String(), c.config.RPCDialTimeout)
 	if err != nil {
 		return nil, err
 	}

client/testing.go

@@ -120,7 +120,7 @@ func TestRPCOnlyClient(t testing.TB, cb func(c *config.Config), srvAddr net.Addr
 	}
 	client.heartbeatStop = newHeartbeatStop(
 		client.getAllocRunner, time.Second, client.logger, client.shutdownCh)
-	client.connPool = pool.NewPool(testlog.HCLogger(t), 10*time.Second, 10, nil, yamux.DefaultConfig())
+	client.connPool = pool.NewPool(testlog.HCLogger(t), 10*time.Second, 10, nil, yamux.DefaultConfig(), 10*time.Second)
 	client.init()
 
 	cancelFunc := func() {

command/agent/agent.go

@@ -573,6 +573,9 @@ func convertServerConfig(agentConfig *Config) (*nomad.Config, error) {
 		if agentConfig.RPC.StreamOpenTimeout > 0 {
 			conf.RPCSessionConfig.StreamOpenTimeout = agentConfig.RPC.StreamOpenTimeout
 		}
+		if agentConfig.RPC.DialTimeout > 0 {
+			conf.RPCDialTimeout = agentConfig.RPC.DialTimeout
+		}
 	}
 
 	// Set the TLS config
@@ -946,6 +949,9 @@ func convertClientConfig(agentConfig *Config) (*clientconfig.Config, error) {
 		if agentConfig.RPC.StreamOpenTimeout > 0 {
 			conf.RPCSessionConfig.StreamOpenTimeout = agentConfig.RPC.StreamOpenTimeout
 		}
+		if agentConfig.RPC.DialTimeout > 0 {
+			conf.RPCDialTimeout = agentConfig.RPC.DialTimeout
+		}
 	}
 
 	conf.PreferredAddressFamily = agentConfig.Client.PreferredAddressFamily

command/agent/agent_test.go

@@ -817,6 +817,21 @@ func TestConvertClientConfig(t *testing.T) {
 				must.True(t, cc.DisableAllocationHookMetrics)
 			},
 		},
+		{
+			name: "rpc dial timeout default",
+			assert: func(t *testing.T, cc *clientconfig.Config) {
+				must.Eq(t, 10*time.Second, cc.RPCDialTimeout)
+			},
+		},
+		{
+			name: "rpc dial timeout override",
+			modConfig: func(c *Config) {
+				c.RPC.DialTimeout = 25 * time.Second
+			},
+			assert: func(t *testing.T, cc *clientconfig.Config) {
+				must.Eq(t, 25*time.Second, cc.RPCDialTimeout)
+			},
+		},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -842,6 +857,46 @@ func TestConvertClientConfig(t *testing.T) {
 	}
 }
 
+// TestAgent_ServerConfig_RPCDialTimeout verifies that rpc.dial_timeout is
+// plumbed through to the nomad server config, falling back to the default
+// when unset.
+func TestAgent_ServerConfig_RPCDialTimeout(t *testing.T) {
+	ci.Parallel(t)
+
+	cases := []struct {
+		name       string
+		modConfig  func(*Config)
+		expectedTO time.Duration
+	}{
+		{
+			name:       "default",
+			expectedTO: 10 * time.Second,
+		},
+		{
+			name: "override",
+			modConfig: func(c *Config) {
+				c.RPC.DialTimeout = 25 * time.Second
+			},
+			expectedTO: 25 * time.Second,
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			conf := DevConfig(nil)
+			require.NoError(t, conf.normalizeAddrs())
+			if tc.modConfig != nil {
+				tc.modConfig(conf)
+			}
+
+			nc, err := convertServerConfig(conf)
+			must.NoError(t, err)
+			must.NotNil(t, nc)
+			must.Eq(t, tc.expectedTO, nc.RPCDialTimeout)
+		})
+	}
+}
+
 func TestAgent_ClientConfig_discovery(t *testing.T) {
 	ci.Parallel(t)
 	conf := DefaultConfig()

command/agent/config.go

@@ -872,6 +872,11 @@ type RPCConfig struct {
 	// and send a RST to the remote side.
 	StreamCloseTimeout    time.Duration
 	StreamCloseTimeoutHCL string `hcl:"stream_close_timeout,optional"`
+
+	// DialTimeout is the timeout used when establishing new outbound RPC
+	// connections to peers. A zero value uses the default.
+	DialTimeout    time.Duration
+	DialTimeoutHCL string `hcl:"dial_timeout,optional"`
 }
 
 func (r *RPCConfig) Copy() *RPCConfig {
@@ -921,6 +926,12 @@ func (r *RPCConfig) Merge(rpc *RPCConfig) *RPCConfig {
 	if rpc.StreamCloseTimeout > 0 {
 		result.StreamCloseTimeout = rpc.StreamCloseTimeout
 	}
+	if rpc.DialTimeoutHCL != "" {
+		result.DialTimeoutHCL = rpc.DialTimeoutHCL
+	}
+	if rpc.DialTimeout > 0 {
+		result.DialTimeout = rpc.DialTimeout
+	}
 	return &result
 }
 
@@ -941,6 +952,9 @@ func (r *RPCConfig) Validate() error {
 		if r.StreamOpenTimeout < 0 {
 			return errors.New("rcp.stream_open_timeout must be greater than zero")
 		}
+		if r.DialTimeout < 0 {
+			return errors.New("rpc.dial_timeout must be greater than or equal to zero")
+		}
 	}
 
 	return nil
@@ -1819,6 +1833,8 @@ func DefaultConfig() *Config {
 			StreamOpenTimeoutHCL:      "75s",
 			StreamCloseTimeout:        5 * time.Minute,
 			StreamCloseTimeoutHCL:     "5m",
+			DialTimeout:               10 * time.Second,
+			DialTimeoutHCL:            "10s",
 		},
 		Client: &ClientConfig{
 			Enabled:               false,

command/agent/config_parse.go

@@ -191,6 +191,7 @@ func ParseConfigFile(path string) (*Config, error) {
 		{"rpc.connection_write_timeout", &c.RPC.ConnectionWriteTimeout, &c.RPC.ConnectionWriteTimeoutHCL, nil},
 		{"rpc.stream_open_timeout", &c.RPC.StreamOpenTimeout, &c.RPC.StreamOpenTimeoutHCL, nil},
 		{"rpc.stream_close_timeout", &c.RPC.StreamCloseTimeout, &c.RPC.StreamCloseTimeoutHCL, nil},
+		{"rpc.dial_timeout", &c.RPC.DialTimeout, &c.RPC.DialTimeoutHCL, nil},
 		{
 			"server.client_introduction.default_identity_ttl",
 			&c.Server.ClientIntroduction.DefaultIdentityTTL,

command/agent/config_parse_test.go

@@ -868,6 +868,8 @@ var sample1 = &Config{
 		StreamOpenTimeoutHCL:      "75s",
 		StreamCloseTimeout:        5 * time.Minute,
 		StreamCloseTimeoutHCL:     "5m",
+		DialTimeout:               15 * time.Second,
+		DialTimeoutHCL:            "15s",
 	},
 	Audit: &config.AuditConfig{
 		Enabled: pointer.Of(true),