Detect offline shotover nodes for KafkaSinkCluster

changelog.md

@@ -10,6 +10,10 @@ This assists us in knowing when to make the next release a breaking release and
 * `Transform::transform` now takes `&mut Wrapper` instead of `Wrapper`.
 * `Wrapper` is renamed to ChainState.
 
+### topology.yaml
+
+* A new mandatory configuration `check_shotover_peers_delay_ms` is added for `KafkaSinkCluster`. See [transform.md](docs/src/transforms.md) for details on this configuration.
+
 ## 0.4.0
 
 ### shotover rust API

docs/src/transforms.md

@@ -293,6 +293,11 @@ If SCRAM authentication against the first kafka broker fails, shotover will term
     # When a timeout occurs the connection to the client is immediately closed.
     # read_timeout: 60
 
+    # Delay in milliseconds before the local shotover node checks its next peer's node state.
+    # The local shotover node will try to establish a TCP connection with the next peer in the list.
+    # If the connection cannot be established within connect_timeout_ms, then the peer is considered down.
+    check_shotover_peers_delay_ms: 3000
+
     # When this field is provided TLS is used when connecting to the remote address.
     # Removing this field will disable TLS.
     #tls:

shotover-proxy/benches/windsock/kafka/bench.rs

@@ -97,7 +97,7 @@ impl KafkaBench {
             KafkaTopology::Cluster1 | KafkaTopology::Cluster3 => Box::new(KafkaSinkClusterConfig {
                 connect_timeout_ms: 3000,
                 read_timeout: None,
-                check_shotover_peers_delay_ms: None,
+                check_shotover_peers_delay_ms: 3000,
                 first_contact_points: vec![kafka_address],
                 shotover_nodes: vec![ShotoverNodeConfig {
                     address: host_address.parse().unwrap(),

shotover-proxy/tests/kafka_int_tests/mod.rs

@@ -400,6 +400,64 @@ async fn cluster_2_racks_multi_shotover(#[case] driver: KafkaDriver) {
         .expect("Shotover did not shutdown within 10s");
     }
 }
+#[rstest]
+#[cfg_attr(feature = "kafka-cpp-driver-tests", case::cpp(KafkaDriver::Cpp))]
+#[case::java(KafkaDriver::Java)]
+#[tokio::test(flavor = "multi_thread")] // multi_thread is needed since java driver will block when consuming, causing shotover logs to not appear
+async fn cluster_2_racks_multi_shotover_with_one_shotover_down(#[case] _driver: KafkaDriver) {
+    let _docker_compose =
+        docker_compose("tests/test-configs/kafka/cluster-2-racks/docker-compose.yaml");
+
+    // One shotover instance per rack
+    let mut shotovers = vec![];
+    for i in 1..3 {
+        shotovers.push(
+            shotover_process(&format!(
+                "tests/test-configs/kafka/cluster-2-racks/topology-rack{i}.yaml"
+            ))
+            .with_config(&format!(
+                "tests/test-configs/shotover-config/config{i}.yaml"
+            ))
+            .with_log_name(&format!("shotover{i}"))
+            .start()
+            .await,
+        );
+    }
+
+    // Wait for check_shotover_peers to start
+    tokio::time::sleep(Duration::from_secs(15)).await;
+
+    // Kill one shotover node
+    tokio::time::timeout(
+        Duration::from_secs(10),
+        shotovers.remove(0).shutdown_and_then_consume_events(&[]),
+    )
+    .await
+    .expect("Shotover did not shutdown within 10s");
+
+    // Wait for the other shotover node to detect the down node
+    tokio::time::sleep(Duration::from_secs(5)).await;
+
+    for shotover in shotovers {
+        let events = tokio::time::timeout(
+            Duration::from_secs(10),
+            shotover.shutdown_and_then_consume_events(&[EventMatcher::new()
+                .with_level(Level::Warn)
+                .with_target("shotover::transforms::kafka::sink_cluster::shotover_node")
+                .with_message(r#"Shotover peer localhost:9191 is down"#)
+                .with_count(Count::Any)]), // with count > 0 is not supported
+        )
+        .await
+        .expect("Shotover did not shutdown within 10s");
+
+        events.assert_contains(
+            &EventMatcher::new()
+                .with_level(Level::Warn)
+                .with_target("shotover::transforms::kafka::sink_cluster::shotover_node")
+                .with_message(r#"Shotover peer localhost:9191 is down"#),
+        );
+    }
+}
 
 #[rstest]
 //#[cfg_attr(feature = "kafka-cpp-driver-tests", case::cpp(KafkaDriver::Cpp))] // CPP driver does not support scram

shotover-proxy/tests/test-configs/kafka/cluster-1-rack/topology-single.yaml

@@ -12,3 +12,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-1-rack/topology1.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-1-rack/topology2.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 1
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-1-rack/topology3.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 2
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-2-racks/topology-rack1.yaml

@@ -15,3 +15,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-2-racks/topology-rack2.yaml

@@ -15,3 +15,4 @@ sources:
             local_shotover_broker_id: 1
             first_contact_points: ["172.16.1.5:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-mtls/topology.yaml

@@ -15,6 +15,7 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               certificate_path: "tests/test-configs/kafka/tls/certs/localhost.crt"

shotover-proxy/tests/test-configs/kafka/cluster-sasl-plain/topology-single.yaml

@@ -12,3 +12,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-sasl-plain/topology1.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-sasl-plain/topology2.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 1
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-sasl-plain/topology3.yaml

@@ -18,3 +18,4 @@ sources:
             local_shotover_broker_id: 2
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-sasl-scram-over-mtls/topology-single.yaml

@@ -20,6 +20,7 @@ sources:
                 verify_hostname: true
               delegation_token_lifetime_seconds: 86400 # 1 day
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               certificate_path: "tests/test-configs/kafka/tls/certs/localhost.crt"

shotover-proxy/tests/test-configs/kafka/cluster-sasl-scram-over-mtls/topology1.yaml

@@ -27,6 +27,7 @@ sources:
                 verify_hostname: true
               delegation_token_lifetime_seconds: 15
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               certificate_path: "tests/test-configs/kafka/tls/certs/localhost.crt"

shotover-proxy/tests/test-configs/kafka/cluster-sasl-scram-over-mtls/topology2.yaml

@@ -27,6 +27,7 @@ sources:
                 verify_hostname: true
               delegation_token_lifetime_seconds: 15
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               certificate_path: "tests/test-configs/kafka/tls/certs/localhost.crt"

shotover-proxy/tests/test-configs/kafka/cluster-sasl-scram-over-mtls/topology3.yaml

@@ -27,6 +27,7 @@ sources:
                 verify_hostname: true
               delegation_token_lifetime_seconds: 15
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               certificate_path: "tests/test-configs/kafka/tls/certs/localhost.crt"

shotover-proxy/tests/test-configs/kafka/cluster-sasl-scram/topology-single.yaml

@@ -12,3 +12,4 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000

shotover-proxy/tests/test-configs/kafka/cluster-tls/topology.yaml

@@ -15,6 +15,7 @@ sources:
             local_shotover_broker_id: 0
             first_contact_points: ["172.16.1.2:9092"]
             connect_timeout_ms: 3000
+            check_shotover_peers_delay_ms: 3000
             tls:
               certificate_authority_path: "tests/test-configs/kafka/tls/certs/localhost_CA.crt"
               verify_hostname: true

shotover/Cargo.toml

@@ -125,6 +125,7 @@ dashmap = { version = "6.0.0", optional = true }
 atoi = { version = "2.0.0", optional = true }
 fnv = "1.0.7"
 sasl = { version = "0.5.1", optional = true, default-features = false, features = ["scram"] }
+rand_core = "0.6.4"
 
 [dev-dependencies]
 criterion = { version = "2.6.0", features = ["async_tokio"], package = "codspeed-criterion-compat" }

shotover/src/transforms/kafka/sink_cluster/kafka_node.rs

@@ -17,6 +17,7 @@ use sasl::client::mechanisms::Scram;
 use sasl::client::Mechanism;
 use sasl::common::scram::Sha256;
 use sasl::common::ChannelBinding;
+use std::fmt::Display;
 use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
@@ -278,6 +279,12 @@ impl KafkaAddress {
     }
 }
 
+impl Display for KafkaAddress {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}:{}", self.host.as_str(), self.port)
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct KafkaNode {
     pub broker_id: BrokerId,

shotover/src/transforms/kafka/sink_cluster/mod.rs

@@ -2,6 +2,7 @@ use crate::frame::kafka::{KafkaFrame, RequestBody, ResponseBody};
 use crate::frame::{Frame, MessageType};
 use crate::message::{Message, Messages};
 use crate::tls::{TlsConnector, TlsConnectorConfig};
+use crate::transforms::kafka::sink_cluster::shotover_node::start_shotover_peers_check;
 use crate::transforms::{
     ChainState, DownChainProtocol, Transform, TransformBuilder, TransformContextBuilder,
     UpChainProtocol,
@@ -68,7 +69,7 @@ pub struct KafkaSinkClusterConfig {
     pub local_shotover_broker_id: i32,
     pub connect_timeout_ms: u64,
     pub read_timeout: Option<u64>,
-    pub check_shotover_peers_delay_ms: Option<u64>,
+    pub check_shotover_peers_delay_ms: u64,
     pub tls: Option<TlsConnectorConfig>,
     pub authorize_scram_over_mtls: Option<AuthorizeScramOverMtlsConfig>,
 }
@@ -113,9 +114,11 @@ impl TransformConfig for KafkaSinkClusterConfig {
             first_contact_points?,
             &self.authorize_scram_over_mtls,
             shotover_nodes,
+            self.local_shotover_broker_id,
             rack,
             self.connect_timeout_ms,
             self.read_timeout,
+            self.check_shotover_peers_delay_ms,
             tls,
         )?))
     }
@@ -153,13 +156,26 @@ impl KafkaSinkClusterBuilder {
         first_contact_points: Vec<KafkaAddress>,
         authorize_scram_over_mtls: &Option<AuthorizeScramOverMtlsConfig>,
         shotover_nodes: Vec<ShotoverNode>,
+        local_shotover_broker_id: i32,
         rack: StrBytes,
         connect_timeout_ms: u64,
         timeout: Option<u64>,
+        check_shotover_peers_delay_ms: u64,
         tls: Option<TlsConnector>,
     ) -> Result<KafkaSinkClusterBuilder> {
         let read_timeout = timeout.map(Duration::from_secs);
         let connect_timeout = Duration::from_millis(connect_timeout_ms);
+        let shotover_peers = shotover_nodes
+            .iter()
+            .filter(|x| x.broker_id.0 != local_shotover_broker_id)
+            .cloned()
+            .collect();
+
+        start_shotover_peers_check(
+            shotover_peers,
+            check_shotover_peers_delay_ms,
+            connect_timeout_ms,
+        );
 
         Ok(KafkaSinkClusterBuilder {
             first_contact_points,

shotover/src/transforms/kafka/sink_cluster/shotover_node.rs

@@ -1,9 +1,16 @@
+use crate::tcp::tcp_stream;
 use crate::transforms::kafka::sink_cluster::kafka_node::KafkaAddress;
 use atomic_enum::atomic_enum;
 use kafka_protocol::messages::BrokerId;
 use kafka_protocol::protocol::StrBytes;
+use rand::rngs::StdRng;
+use rand::{Rng, SeedableRng};
+use rand_core::Error;
 use serde::{Deserialize, Serialize};
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
+use std::time::Duration;
+use tokio::time::sleep;
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
@@ -33,9 +40,85 @@ pub(crate) struct ShotoverNode {
     state: Arc<AtomicShotoverNodeState>,
 }
 
+impl ShotoverNode {
+    #![allow(unused)]
+    pub(crate) fn is_up(&self) -> bool {
+        self.state.load(Ordering::Relaxed) == ShotoverNodeState::Up
+    }
+
+    pub(crate) fn set_state(&self, state: ShotoverNodeState) {
+        self.state.store(state, Ordering::Relaxed)
+    }
+}
+
 #[atomic_enum]
 #[derive(PartialEq)]
 pub(crate) enum ShotoverNodeState {
     Up,
     Down,
 }
+
+pub(crate) fn start_shotover_peers_check(
+    shotover_peers: Vec<ShotoverNode>,
+    check_shotover_peers_delay_ms: u64,
+    connect_timeout_ms: u64,
+) {
+    if !shotover_peers.is_empty() {
+        tokio::spawn(async move {
+            // Wait for all shotover nodes to start
+            sleep(Duration::from_secs(10)).await;
+            loop {
+                match check_shotover_peers(
+                    &shotover_peers,
+                    check_shotover_peers_delay_ms,
+                    connect_timeout_ms,
+                )
+                .await
+                {
+                    Ok(_) => {}
+                    Err(err) => {
+                        tracing::error!(
+                            "Restarting the shotover peers check due to error: {err:?}"
+                        );
+                    }
+                };
+            }
+        });
+    }
+}
+
+async fn check_shotover_peers(
+    shotover_peers: &[ShotoverNode],
+    check_shotover_peers_delay_ms: u64,
+    connect_timeout_ms: u64,
+) -> Result<(), Error> {
+    let mut shotover_peers_cycle = shotover_peers.iter().cycle();
+    let mut rng = StdRng::from_rng(rand::thread_rng())?;
+    let check_shotover_peers_delay_ms = check_shotover_peers_delay_ms as i64;
+    loop {
+        if let Some(shotover_peer) = shotover_peers_cycle.next() {
+            let tcp_stream = tcp_stream(
+                Duration::from_millis(connect_timeout_ms),
+                &shotover_peer.address.to_string().as_str(),
+            )
+            .await;
+            match tcp_stream {
+                Ok(_) => {
+                    shotover_peer.set_state(ShotoverNodeState::Up);
+                }
+                Err(_) => {
+                    tracing::warn!(
+                        "Shotover peer {} is down",
+                        shotover_peer.address.to_string()
+                    );
+                    shotover_peer.set_state(ShotoverNodeState::Down);
+                }
+            }
+            let random_delay = (check_shotover_peers_delay_ms
+                + rng.gen_range(
+                    -check_shotover_peers_delay_ms / 10..check_shotover_peers_delay_ms / 10,
+                )) as u64;
+            sleep(Duration::from_millis(random_delay)).await;
+        }
+    }
+}