Make connect timeout configurable

docs/src/examples/cassandra-cluster-shotover-sidecar.md

@@ -132,6 +132,7 @@ chain_config:
  port: 9043
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9042"
+ connect_timeout_ms: 3000
 named_topics:
  testtopic: 5
 source_to_chain_mapping:

docs/src/transforms.md

@@ -106,6 +106,11 @@ While `system.peers`/`system.peers_v2` will be rewritten to list the configured
  # * which shotover_nodes entry is included in system.local and excluded from system.peers
  local_shotover_host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
 
+ # Number of milliseconds to wait for a connection to be created to a destination cassandra instance.
+ # If the timeout is exceeded then connection to another node is attempted
+ # If all known nodes have resulted in connection timeouts an error will be returned to the client.
+ connect_timeout_ms: 3000
+
  # When this field is provided TLS is used when connecting to the remote address.
  # Removing this field will disable TLS.
  #tls:
@@ -149,6 +154,10 @@ No cluster discovery or routing occurs with this transform.
  # The IP address and port of the upstream Cassandra node/service.
  remote_address: "127.0.0.1:9042"
 
+ # Number of milliseconds to wait for a connection to be created to the destination cassandra instance.
+ # If the timeout is exceeded then an error is returned to the client.
+ connect_timeout_ms: 3000
+
  # When this field is provided TLS is used when connecting to the remote address.
  # Removing this field will disable TLS.
  #tls:
@@ -290,6 +299,7 @@ If we have a parallelism of 3 then we would have 3 instances of the chain: C1, C
  name: "DR chain"
  - RedisSinkSingle:
  remote_address: "127.0.0.1:6379"
+ connect_timeout_ms: 3000
 ```
 
 ### Protect
@@ -390,6 +400,8 @@ This transform will attempt to cache values for a given primary key in a Redis h
  - RedisSinkSingle:
  # The IP address and port of the upstream redis node/service.
  remote_address: "127.0.0.1:6379"
+ connect_timeout_ms: 3000
+
 ```
 
 ### RedisClusterPortsRewrite
@@ -422,6 +434,11 @@ This transform is a full featured Redis driver that will connect to a Redis clus
  # When this field is not provided connection_count defaults to 1.
  connection_count: 1
 
+ # Number of milliseconds to wait for a connection to be created to a destination redis instance.
+ # If the timeout is exceeded then connection to another node is attempted
+ # If all known nodes have resulted in connection timeouts an error will be returned to the client.
+ connect_timeout_ms: 3000
+
  # When this field is provided TLS is used when connecting to the remote address.
  # Removing this field will disable TLS.
  #tls:
@@ -461,6 +478,10 @@ This transform will take a query, serialise it into a RESP2 compatible format an
  # The IP address and port of the upstream redis node/service.
  remote_address: "127.0.0.1:6379"
 
+ # Number of milliseconds to wait for a connection to be created to the destination redis instance.
+ # If the timeout is exceeded then an error is returned to the client.
+ connect_timeout_ms: 3000
+
  # When this field is provided TLS is used when connecting to the remote address.
  # Removing this field will disable TLS.
  #tls:

docs/src/user-guide/configuration.md

@@ -14,7 +14,7 @@ The configuration file is used to change general behavior of Shotover. Currently
 
 This is a single string that you can use to configure logging with Shotover. It supports [env_filter](https://docs.rs/env_logger/0.7.1/env_logger/) style configuration and filtering syntax. Log levels and filters can be dynamically changed while Shotover is still running.
 
-### observability_interface 
+### observability_interface
 
 Shotover has an observability interface for you to collect Prometheus data from. This value will define the address and port for Shotover's observability interface. It is configured as a string in the format of `127.0.0.1:8080` for IPV4 addresses or `[2001:db8::1]:8080` for IPV6 addresses. More information is on the [observability page](./observability.md).
 
@@ -120,11 +120,13 @@ chain_config:
  name: "DR chain"
  - RedisSinkCluster:
  first_contact_points: [ "127.0.0.1:2120", "127.0.0.1:2121", "127.0.0.1:2122", "127.0.0.1:2123", "127.0.0.1:2124", "127.0.0.1:2125" ]
+ connect_timeout_ms: 3000
  #The rest of the chain, these transforms are blocking
  - QueryCounter:
  name: "Main chain"
  - RedisSinkCluster:
  first_contact_points: [ "127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225" ]
+ connect_timeout_ms: 3000
 ```
 
 ### `source_to_chain_mapping` Chain Mapping

shotover-proxy/example-configs-docker/cassandra-peers-rewrite/topology.yaml

@@ -9,5 +9,6 @@ chain_config:
  port: 9043
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9042"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs-docker/redis-cluster-ports-rewrite/topology.yaml

@@ -9,5 +9,6 @@ chain_config:
  new_port: 6380
  - RedisSinkSingle:
  remote_address: "0.0.0.0:6379"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  redis_prod: redis_chain

shotover-proxy/example-configs/cassandra-cluster-multi-rack/topology_rack1.yaml

@@ -21,5 +21,6 @@ chain_config:
  data_center: "dc1"
  rack: "rack3"
  host_id: "fa74d7ec-1223-472b-97de-04a32ccdb70b"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster-multi-rack/topology_rack2.yaml

@@ -21,5 +21,6 @@ chain_config:
  data_center: "dc1"
  rack: "rack3"
  host_id: "fa74d7ec-1223-472b-97de-04a32ccdb70b"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster-multi-rack/topology_rack3.yaml

@@ -21,5 +21,6 @@ chain_config:
  data_center: "dc1"
  rack: "rack3"
  host_id: "fa74d7ec-1223-472b-97de-04a32ccdb70b"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster-tls/topology.yaml

@@ -18,6 +18,7 @@ chain_config:
  data_center: "dc1"
  rack: "rack1"
  host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/docker-images/cassandra-tls-4.0.6/certs/localhost_CA.crt"
  certificate_path: "example-configs/docker-images/cassandra-tls-4.0.6/certs/localhost.crt"

shotover-proxy/example-configs/cassandra-cluster/topology-dummy-peers-v3.yaml

@@ -25,5 +25,6 @@ chain_config:
  data_center: "dc1"
  rack: "rack1"
  host_id: "fa74d7ec-1223-472b-97de-04a32ccdb70b"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster/topology-dummy-peers-v4.yaml

@@ -25,5 +25,6 @@ chain_config:
  data_center: "dc1"
  rack: "rack1"
  host_id: "fa74d7ec-1223-472b-97de-04a32ccdb70b"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster/topology-v3.yaml

@@ -13,6 +13,7 @@ chain_config:
  data_center: "dc1"
  rack: "rack1"
  host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
+ connect_timeout_ms: 3000
 
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-cluster/topology-v4.yaml

@@ -13,6 +13,7 @@ chain_config:
  data_center: "dc1"
  rack: "rack1"
  host_id: "2dd022d6-2937-4754-89d6-02d2933a8f7a"
+ connect_timeout_ms: 3000
 
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-passthrough/topology-encode.yaml

@@ -10,5 +10,6 @@ chain_config:
  encode_responses: true
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-passthrough/topology.yaml

@@ -7,5 +7,6 @@ chain_config:
  main_chain:
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-protect-aws/topology.yaml

@@ -18,5 +18,6 @@ chain_config:
  - col1
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-protect-local/topology.yaml

@@ -16,5 +16,6 @@ chain_config:
  - col1
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-redis-cache/topology.yaml

@@ -16,7 +16,9 @@ chain_config:
  chain:
  - RedisSinkSingle:
  remote_address: "127.0.0.1:6379"
+ connect_timeout_ms: 3000
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-request-throttling/topology.yaml

@@ -9,5 +9,6 @@ chain_config:
  max_requests_per_second: 100000
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9043"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  cassandra_prod: main_chain

shotover-proxy/example-configs/cassandra-tls/topology-with-key.yaml

@@ -11,6 +11,7 @@ chain_config:
  main_chain:
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9042"
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/docker-images/cassandra-tls-4.0.6/certs/localhost_CA.crt"
  certificate_path: "example-configs/docker-images/cassandra-tls-4.0.6/certs/localhost.crt"

shotover-proxy/example-configs/cassandra-tls/topology.yaml

@@ -11,6 +11,7 @@ chain_config:
  main_chain:
  - CassandraSinkSingle:
  remote_address: "127.0.0.1:9042"
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/docker-images/cassandra-tls-4.0.6/certs/localhost_CA.crt"
  verify_hostname: false

shotover-proxy/example-configs/redis-cluster-dr/topology.yaml

@@ -21,9 +21,11 @@ chain_config:
  name: "DR chain"
  - RedisSinkCluster:
  first_contact_points: [ "127.0.0.1:2120", "127.0.0.1:2121", "127.0.0.1:2122", "127.0.0.1:2123", "127.0.0.1:2124", "127.0.0.1:2125" ]
+ connect_timeout_ms: 3000
  - QueryCounter:
  name: "Main chain"
  - RedisSinkCluster:
  first_contact_points: [ "127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225" ]
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  redis_prod: redis_chain

shotover-proxy/example-configs/redis-cluster-handling/topology.yaml

@@ -10,6 +10,7 @@ chain_config:
  - RedisSinkCluster:
  first_contact_points: ["127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225"]
  direct_destination: "127.0.0.1:2220"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  # connect our Redis source to our Redis cluster sink (transform).
  redis_prod: redis_chain

shotover-proxy/example-configs/redis-cluster-hiding/topology.yaml

@@ -9,6 +9,7 @@ chain_config:
  # configure Shotover to connect to the Redis cluster via our defined contact points
  - RedisSinkCluster:
  first_contact_points: ["127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225"]
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  # connect our Redis source to our Redis cluster sink (transform).
  redis_prod: redis_chain

shotover-proxy/example-configs/redis-cluster-tls/topology-with-key.yaml

@@ -7,6 +7,7 @@ chain_config:
  redis_chain:
  - RedisSinkCluster:
  first_contact_points: ["127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225"]
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/redis-tls/certs/ca.crt"
  certificate_path: "example-configs/redis-tls/certs/redis.crt"

shotover-proxy/example-configs/redis-cluster-tls/topology.yaml

@@ -7,6 +7,7 @@ chain_config:
  redis_chain:
  - RedisSinkCluster:
  first_contact_points: ["127.0.0.1:2220", "127.0.0.1:2221", "127.0.0.1:2222", "127.0.0.1:2223", "127.0.0.1:2224", "127.0.0.1:2225"]
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/redis-tls/certs/ca.crt"
  verify_hostname: true

shotover-proxy/example-configs/redis-multi/topology.yaml

@@ -13,13 +13,16 @@ chain_config:
  - RedisTimestampTagger
  - RedisSinkSingle:
  remote_address: "127.0.0.1:3331"
+ connect_timeout_ms: 3000
  two:
  - RedisTimestampTagger
  - RedisSinkSingle:
  remote_address: "127.0.0.1:3332"
+ connect_timeout_ms: 3000
  three:
  - RedisTimestampTagger
  - RedisSinkSingle:
  remote_address: "127.0.0.1:3333"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  redis_prod: redis_chain

shotover-proxy/example-configs/redis-passthrough/topology.yaml

@@ -7,5 +7,6 @@ chain_config:
  redis_chain:
  - RedisSinkSingle:
  remote_address: "127.0.0.1:1111"
+ connect_timeout_ms: 3000
 source_to_chain_mapping:
  redis_prod: redis_chain

shotover-proxy/example-configs/redis-tls/topology.yaml

@@ -14,6 +14,7 @@ chain_config:
  redis_chain_tls:
  - RedisSinkSingle:
  remote_address: "127.0.0.1:1111"
+ connect_timeout_ms: 3000
  tls:
  certificate_authority_path: "example-configs/redis-tls/certs/ca.crt"
  certificate_path: "example-configs/redis-tls/certs/redis.crt"

shotover-proxy/src/tcp.rs

@@ -5,12 +5,15 @@ use tokio::{
  time::timeout,
 };
 
-pub async fn tcp_stream<A: ToSocketAddrs + std::fmt::Debug>(destination: A) -> Result<TcpStream> {
- timeout(Duration::from_secs(3), TcpStream::connect(&destination))
+pub async fn tcp_stream<A: ToSocketAddrs + std::fmt::Debug>(
+ connect_timeout: Duration,
+ destination: A,
+) -> Result<TcpStream> {
+ timeout(connect_timeout, TcpStream::connect(&destination))
  .await
  .map_err(|_| {
  anyhow!(
- "destination {destination:?} did not respond to connection attempt within 3 seconds"
+ "destination {destination:?} did not respond to connection attempt within {connect_timeout:?}"
  )
  })?
  .map_err(|e| {

shotover-proxy/src/transforms/cassandra/connection.rs

@@ -36,12 +36,13 @@ pub struct CassandraConnection {
 
 impl CassandraConnection {
  pub async fn new<A: ToSocketAddrs + std::fmt::Debug>(
+ connect_timeout: Duration,
  host: A,
  codec: CassandraCodec,
  mut tls: Option<TlsConnector>,
  pushed_messages_tx: Option<mpsc::UnboundedSender<Messages>>,
  ) -> Result<Self> {
- let tcp_stream = tcp::tcp_stream(host).await?;
+ let tcp_stream = tcp::tcp_stream(connect_timeout, host).await?;
 
  let (out_tx, out_rx) = mpsc::unbounded_channel::<Request>();
  let (return_tx, return_rx) = mpsc::unbounded_channel::<Request>();

shotover-proxy/src/transforms/cassandra/sink_cluster/mod.rs

@@ -51,6 +51,7 @@ pub struct CassandraSinkClusterConfig {
  pub local_shotover_host_id: Uuid,
  pub shotover_nodes: Vec<ShotoverNode>,
  pub tls: Option<TlsConnectorConfig>,
+ pub connect_timeout_ms: u64,
  pub read_timeout: Option<u64>,
 }
 
@@ -77,6 +78,7 @@ impl CassandraSinkClusterConfig {
  chain_name,
  local_node,
  tls,
+ self.connect_timeout_ms,
  self.read_timeout,
  ),
  )))
@@ -157,10 +159,12 @@ impl CassandraSinkCluster {
  chain_name: String,
  local_shotover_node: ShotoverNode,
  tls: Option<TlsConnector>,
+ connect_timeout_ms: u64,
  timeout: Option<u64>,
  ) -> Self {
  let failed_requests = register_counter!("failed_requests", "chain" => chain_name.clone(), "transform" => "CassandraSinkCluster");
  let receive_timeout = timeout.map(Duration::from_secs);
+ let connect_timeout = Duration::from_millis(connect_timeout_ms);
 
  let (local_nodes_tx, local_nodes_rx) = watch::channel(vec![]);
  let (task_handshake_tx, task_handshake_rx) = mpsc::channel(1);
@@ -173,7 +177,7 @@ impl CassandraSinkCluster {
 
  Self {
  contact_points,
- connection_factory: ConnectionFactory::new(tls),
+ connection_factory: ConnectionFactory::new(connect_timeout, tls),
  shotover_peers,
  control_connection: None,
  control_connection_address: None,