This is an automated email from the ASF dual-hosted git repository. krishvishal pushed a commit to branch kafka-gateway-failover-tests in repository https://gitbox.apache.org/repos/asf/iggy.git
commit bbe9dda5543d116dededbaeabfe8d24a22e6bc1f Author: Krishna Vishal <[email protected]> AuthorDate: Mon Nov 2 13:02:57 2026 +0530 test(gateways): add failover tests for gateways beside a 3-node cluster Check that no record is lost, written twice or read twice when a gateway or a node fails. --- .config/nextest.toml | 17 + .github/actions/rust/pre-merge/action.yml | 12 +- gateways/kafka/README.md | 7 +- gateways/kafka/docs/AUTHENTICATION.md | 2 +- gateways/kafka/docs/CONSUMER_GROUPS.md | 5 +- gateways/kafka/docs/TEST_SUITE.md | 22 +- gateways/kafka/tests/cluster_failover_tests.rs | 1340 ++++++++++++++++++++++++ gateways/kafka/tests/common/deployment.rs | 417 ++++++++ gateways/kafka/tests/common/docker_clients.rs | 265 +++++ gateways/kafka/tests/common/iggy_server.rs | 60 ++ 10 files changed, 2134 insertions(+), 13 deletions(-) diff --git a/.config/nextest.toml b/.config/nextest.toml index 1c67e3df7..e4d44bb69 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -75,6 +75,23 @@ max-threads = 1 filter = 'binary_id(iggy-gateway-kafka::kafka_client_e2e_tests)' test-group = "kafka_client_e2e" +# Each test in these binaries runs a three-node cluster, so they run one at a time. +[test-groups.kafka_cluster] +max-threads = 1 + +[[profile.default.overrides]] +filter = 'binary_id(iggy-gateway-kafka::coordinator_cluster_tests)' +test-group = "kafka_cluster" + +# The failover tests also run three gateway processes and Kafka clients in Docker for minutes, and +# they time takeovers. Each gets the whole machine, so other tests do not distort that timing, and +# up to ten minutes. +[[profile.default.overrides]] +filter = 'binary_id(iggy-gateway-kafka::cluster_failover_tests)' +test-group = "kafka_cluster" +threads-required = "num-cpus" +slow-timeout = { period = "60s", terminate-after = 10 } + # Each ext-auth test spawns its own iggy-server + mock HTTP server. Each server opens io_uring # rings and listener sockets, so running all 24 at once wastes descriptors for no coverage gain. [test-groups.ext-auth] diff --git a/.github/actions/rust/pre-merge/action.yml b/.github/actions/rust/pre-merge/action.yml index c62d1b30e..b9266ee0c 100644 --- a/.github/actions/rust/pre-merge/action.yml +++ b/.github/actions/rust/pre-merge/action.yml @@ -397,13 +397,15 @@ runs: # naming the feature under test instead of naming the pull. The suite runs its # containers with --pull never, so a failed pull here makes it skip, which the guard # above turns into a job failure that names the missing image. - # Read the tags out of the suite itself; hardcoding them here lets a bump in the test + # Read the tags out of the suites themselves; hardcoding them here lets a bump in a test # desync silently, which pre-pulls the superseded image and puts the real one back - # inside the budget this block exists to avoid. - E2E_IMAGES=$(grep -oP '^const (KCAT|KAFKA)_IMAGE: &str = "\K[^"]+' \ - gateways/kafka/tests/kafka_client_e2e_tests.rs) + # inside the budget this block exists to avoid. The failover suite runs the same + # clients against a cluster. + E2E_IMAGES=$(grep -ohP '^const (KCAT|KAFKA)_IMAGE: &str = "\K[^"]+' \ + gateways/kafka/tests/kafka_client_e2e_tests.rs \ + gateways/kafka/tests/cluster_failover_tests.rs | sort -u) if [[ -z "$E2E_IMAGES" ]]; then - echo "::warning::could not read client image tags from kafka_client_e2e_tests.rs; skipping pre-pull" + echo "::warning::could not read client image tags from the Kafka client suites; skipping pre-pull" fi for image in $E2E_IMAGES; do timeout 300 docker pull -q "$image" \ diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 35f2b5957..c109d0818 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -352,7 +352,7 @@ cargo run -p iggy-gateway-kafka with their members and offsets. A member that keeps its heartbeat goes on without a rebalance. A stop on SIGTERM hands the partitions over without waiting for the session timeout. After a crash, the takeover waits for the Iggy session timeout, 1.2 times `heartbeat.interval`, 36 s by - default (not tested). + default. `tests/cluster_failover_tests.rs` measures about 39 s. - Metadata lists each gateway that owns a log partition, and this gateway. It names this gateway as the leader of every partition. Any gateway can serve any partition, because the Iggy SDK sends each partition request to its primary. A client of a stopped or crashed gateway goes on @@ -386,8 +386,9 @@ cargo run -p iggy-gateway-kafka gateway log names it. - To lower the takeover time after a crash, lower Iggy's `heartbeat.interval`. It must stay above (3 times `consumer_group.heartbeat_interval` + 6.2 s) / 1.2. With - `consumer_group.heartbeat_interval = "1s"`, that is about 7.7 s, so a takeover after about 9 s - (not tested). See `core/server/config.toml`. + `consumer_group.heartbeat_interval = "1s"`, that is about 7.7 s. An 8 s `heartbeat.interval` + gives a session timeout of 9.6 s, and `tests/cluster_failover_tests.rs` measures a takeover + after 10 to 11 s. See `core/server/config.toml`. - Each gateway holds the Iggy sessions that "Concurrency ceiling" counts, up to 256 producer sessions and about 30 others. N gateways times their sessions must stay below Iggy's cluster-wide clients table, 8192 sessions. Three gateways at the default limits hold about 860. diff --git a/gateways/kafka/docs/AUTHENTICATION.md b/gateways/kafka/docs/AUTHENTICATION.md index fcf97cfbe..fa0d6c1c6 100644 --- a/gateways/kafka/docs/AUTHENTICATION.md +++ b/gateways/kafka/docs/AUTHENTICATION.md @@ -67,7 +67,7 @@ connection. The cost rules it out. The Argon2 verify runs on the shard's compio runtime, which has no blocking pool. A burst of Kafka connections therefore stalls that shard's entire request loop, including produce and fetch traffic that has -nothing to do with authentication. The 8192 ceiling (`[metadata] clients_table_max`, `core/consensus/src/impls.rs:211`) is cluster-wide. +nothing to do with authentication. The 8192 ceiling (`[metadata] clients_table_max`, `core/consensus/src/impls.rs:249`) is cluster-wide. So connection churn on the Kafka side turns directly into session-table pressure on the Iggy side. ## Why not a shared service account diff --git a/gateways/kafka/docs/CONSUMER_GROUPS.md b/gateways/kafka/docs/CONSUMER_GROUPS.md index b9762041b..b8d9a988c 100644 --- a/gateways/kafka/docs/CONSUMER_GROUPS.md +++ b/gateways/kafka/docs/CONSUMER_GROUPS.md @@ -312,8 +312,9 @@ a warning. A gateway with `IGGY_KAFKA_COORDINATOR_LOG=true` does not start. The this gateway. The others get `NOT_COORDINATOR` per group. The Java admin client asks every broker and keeps one answer per group. -Not tested: the takeover time after a crash. It waits for the Iggy session timeout, 1.2 times -`heartbeat.interval`, 36 s by default. +After a crash, the takeover waits for the Iggy session timeout, 1.2 times `heartbeat.interval`, +36 s by default. `tests/cluster_failover_tests.rs` measures about 39 s, and 10 to 11 s with an 8 s +`heartbeat.interval` and a 1 s `consumer_group.heartbeat_interval`. With `IGGY_KAFKA_COORDINATOR_LOG=false`, group membership is process memory alone. With `IGGY_KAFKA_BROKERS`, every gateway's `FindCoordinator` then names the gateway with the lowest node diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index 37b99b338..784008091 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -91,10 +91,11 @@ file under `tests/` anymore. | [`metadata_changes_real_bridge_tests.rs`](../tests/metadata_changes_real_bridge_tests.rs) | The bridge following Iggy's metadata change log against a real, spawned `iggy-server`. A group that one gateway stores a first offset for lists on another within about a second, with no scan client, and a deleted topic takes its groups off the list. A bridge user without `manage_servers` still lists groups through the scan. A permission change and a delete of a user reach the bridge, and over SA [...] | [`coordinator_real_bridge_tests.rs`](../tests/coordinator_real_bridge_tests.rs) | Several in-process gateways sharing their groups through the coordinator log on one real, spawned `iggy-server`: every gateway names the same coordinator for each group and the groups spread over all of them, a stopped coordinator's group goes on at another gateway with its member, generation, offsets and commit metadata, a foreign newer claim stops the coordinator until it claims again and reloads, only [...] | [`coordinator_cluster_tests.rs`](../tests/coordinator_cluster_tests.rs) | Three gateways beside a three-node Iggy cluster, one beside each node: the coordinating gateway stops, an Iggy node stops, and each gateway restarts in turn. The member keeps its id and generation through each, and its offsets stay | Needs `iggy-server` | +| [`cluster_failover_tests.rs`](../tests/cluster_failover_tests.rs) | Three compiled gateways beside a three-node Iggy cluster, under load from an idempotent Java producer and a Java consumer group of two members. Each gateway restarts in turn with SIGTERM. Each node stops, the metadata primary first, then the primary of the group's coordinator partition. Each gateway stops with its node. The coordinator crashes with SIGKILL, alone and with its node. kcat finds every acknowledged record [...] | [`cluster_id_real_bridge_tests.rs`](../tests/cluster_id_real_bridge_tests.rs) | The cluster id stored in a real, spawned `iggy-server`: one id for every bridge, the first write winning when several start at once, a persisted topic with no expiry or size cap, the id surviving an `iggy-server` SIGKILL, a stored value that is not an id refused, one id per default stream, and the compiled gateway binary reporting the same id in Metadata and DescribeCluster across a restart ([#4519](https:/ [...] -| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - with `cluster_id_real_bridge_tests.rs`, one of the two suites that spawn the gateway as a process rather than calling `KafkaGateway::run` in-process. A gateway started 16 s before its `iggy-server` waits for it a [...] +| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary, bridged to a real spawned `iggy-server`. With `cluster_id_real_bridge_tests.rs` and `cluster_failover_tests.rs`, it is one of the three suites that spawn the gateway as a process rather than calling `KafkaGateway::run` in-process. A gateway started 16 s [...] -`tests/common/` holds shared helpers (`certs.rs`, `codec.rs`, `fixtures.rs`, `gateway_process.rs`, +`tests/common/` holds shared helpers (`certs.rs`, `codec.rs`, `deployment.rs`, `docker_clients.rs`, `fixtures.rs`, `gateway_process.rs`, `gateways.rs`, `scope.rs`, `server.rs`, `iggy_server.rs`, `tcp.rs`, `wire.rs`), compiled per test binary via `#[path]`, not a test binary itself. `certs.rs` mints a CA, server and client certificates with `rcgen` per test, so nothing in the tree can expire. `codec.rs` is test-only primitive encode/decode scaffolding for hand-building legacy/adversarial wire shapes @@ -130,6 +131,23 @@ shard pools, so the two can run alongside each other. It automates categories S and T of [`MANUAL_TESTING.md`](MANUAL_TESTING.md). Those procedures stay, because they cover cases a test does not assert, but the load-bearing ones now run in CI. +## Failover suite + +`cluster_failover_tests.rs` uses the same two images and `KAFKA_E2E_REQUIRED` as the real-client +suite, and skips in the same way. It needs no JDK. Cargo builds the gateway binary for the test, +but `iggy-server` comes from another package, so build it first: + +```bash +cargo build --bin iggy-server +docker pull edenhill/kcat:1.7.1 +docker pull apache/kafka:3.9.0 +KAFKA_E2E_REQUIRED=1 cargo test -p iggy-gateway-kafka --test cluster_failover_tests +``` + +Each test takes 1 to 3 minutes. It runs in the `kafka_cluster` nextest group, alone on the machine, +so other tests do not change the takeover times it measures. Nextest gives each test up to 10 +minutes. A failed test keeps the logs of its nodes, gateways and clients, and prints where they are. + ## Adding new tests 1. **New API key or version range** — update `SUPPORTED_RANGES` in `api.rs`, `SCOPE.md`, and add diff --git a/gateways/kafka/tests/cluster_failover_tests.rs b/gateways/kafka/tests/cluster_failover_tests.rs new file mode 100644 index 000000000..592f0bf77 --- /dev/null +++ b/gateways/kafka/tests/cluster_failover_tests.rs @@ -0,0 +1,1340 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Three gateways beside three Iggy nodes, under load from real Kafka clients, while gateways and +//! nodes stop, crash and come back. +//! +//! The stack is real end to end. Three `iggy-server` processes form a cluster, the compiled +//! `iggy-gateway-kafka` runs beside each node, and the clients run in Docker. An idempotent Java +//! producer writes numbered records through the three gateways, and a Java consumer group of two +//! members reads them. At the end, kcat reads back what each partition stores. +//! +//! A probe keeps a member of a second group alive with a heartbeat to every gateway, five times a +//! second, and asks every gateway to describe the Java group. Both groups hash to one coordinator +//! partition, so one gateway coordinates both. +//! +//! Each test checks the same things at the end. Every acknowledged record is stored once, at the +//! offset that the producer got, and the group reads it once. The group does not rebalance, and no +//! two gateways answer as coordinator of a group at once. +//! +//! Prerequisites are Docker, the two client images and an `iggy-server` built into the same target +//! directory. Missing any skips, unless `KAFKA_E2E_REQUIRED=1` is set, as in +//! `kafka_client_e2e_tests.rs`. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/deployment.rs"] +mod deployment; +#[path = "common/docker_clients.rs"] +mod docker_clients; +#[path = "common/gateways.rs"] +mod gateways; +#[path = "common/iggy_server.rs"] +mod iggy_server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::collections::{BTreeMap, BTreeSet}; +use std::fs; +use std::future::Future; +use std::net::SocketAddr; +use std::path::Path; +use std::sync::{Arc, Mutex, PoisonError}; +use std::time::Duration; + +use bytes::Bytes; +use futures::future::join_all; +use kafka_protocol::messages::{ + DescribeGroupsRequest, DescribeGroupsResponse, FindCoordinatorRequest, FindCoordinatorResponse, + GroupId, HeartbeatRequest, HeartbeatResponse, MetadataRequest, MetadataResponse, +}; +use kafka_protocol::protocol::{Decodable, StrBytes}; +use serde_json::Value; +use serial_test::serial; +use tempfile::TempDir; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::task::JoinHandle; +use tokio::time::{Instant, MissedTickBehavior}; +use tokio_util::sync::CancellationToken; + +use iggy_gateway_kafka::bridge::{COORDINATOR_LOG_TOPIC, cluster_id_stream}; +use iggy_gateway_kafka::coordinator::{PARTITIONS as COORDINATOR_PARTITIONS, partition_of_group}; +use iggy_gateway_kafka::protocol::api::{ + API_KEY_DESCRIBE_GROUPS, API_KEY_FIND_COORDINATOR, API_KEY_HEARTBEAT, API_KEY_METADATA, + ERROR_FENCED_INSTANCE_ID, ERROR_ILLEGAL_GENERATION, ERROR_NONE, ERROR_REBALANCE_IN_PROGRESS, + ERROR_UNKNOWN_MEMBER_ID, +}; + +use deployment::{Deployment, KAFKA_STREAM, NODES, TAIL_LINES, tail}; +use docker_clients::{Container, missing_prerequisite, run_to_end, skip}; + +const KAFKA_IMAGE: &str = "apache/kafka:3.9.0"; +const KCAT_IMAGE: &str = "edenhill/kcat:1.7.1"; + +/// The topic of the load. Two partitions per node, so a stop of any node moves primaries. +const TOPIC: &str = "failover"; +const PARTITIONS: u32 = 6; + +/// The Java consumer group, two members of `kafka-verifiable-consumer.sh`. +const GROUP: &str = "failover-readers"; +const MEMBERS: usize = 2; + +/// Records per second that the producer writes, with keys that spread them over every partition. +const RATE: u32 = 100; +const KEYS: &str = "60"; + +/// The idempotent Java producer. These are the defaults since Kafka 3.0, written out because +/// `kafka-verifiable-producer.sh` sets `retries=0` before it reads this file. The delivery timeout +/// outlasts every step, so no record expires in the producer. +const PRODUCER_CONFIG: &str = "enable.idempotence=true\n\ + acks=all\n\ + retries=2147483647\n\ + max.in.flight.requests.per.connection=5\n\ + request.timeout.ms=10000\n\ + delivery.timeout.ms=300000\n"; + +/// Acknowledgements, and reads of each member, that a step waits for, so the next step starts +/// under load. +const ACK_PROGRESS: usize = 300; +const READ_PROGRESS: usize = 100; + +/// The Iggy session timeout by default: 1.2 x the 30 s `heartbeat.interval`. +const SESSION_TIMEOUT: Duration = Duration::from_secs(36); + +/// Server settings for the shortest session timeout that a 1 s report interval allows. The server +/// needs 1.2 x `heartbeat.interval` above three report intervals plus 6.2 s, so 8 s is the next +/// whole second, and the timeout is 9.6 s. +const SHORT_SESSION: [(&str, &str); 2] = [ + ("IGGY_HEARTBEAT_INTERVAL", "8s"), + ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"), +]; +const SHORT_SESSION_TIMEOUT: Duration = Duration::from_millis(9_600); + +/// What a takeover after a crash may take beyond the session timeout: the report that ends the +/// session, the lease read of the successor once a second, and its load of the partition. +const TAKEOVER_SLACK: Duration = Duration::from_secs(15); + +/// How long the cluster takes to elect a new primary for a lost one: its `cluster.heartbeat_timeout` +/// of 5 s, with an election. +const VIEW_CHANGE: Duration = Duration::from_secs(10); + +/// How long the groups of a gateway may take to move after its SIGTERM. A successor reads its lease +/// once a second, then loads the partition. A move that waits for the session timeout fails this. +const HANDOVER_LIMIT: Duration = Duration::from_secs(15); + +/// How long the producer may take to write again through another gateway after one crashed. Below +/// the session timeout, so the data path does not wait for it. +const DATA_PATH_LIMIT: Duration = Duration::from_secs(20); + +/// How long a step may take to recover: view changes, reconnects and group moves. +const RECOVERY: Duration = Duration::from_secs(90); + +/// How long a restarted node may take to catch up and get a primary back from the balancer of +/// #4619. It moves one primary at a time, after the cluster is stable for 5 s. +const PRIMARY_MOVE_LIMIT: Duration = Duration::from_secs(120); + +/// How often the probe asks every gateway, and how long one answer may take. +const PROBE_INTERVAL: Duration = Duration::from_millis(200); +const PROBE_TIMEOUT: Duration = Duration::from_secs(2); + +/// Probe rounds that a test must see, so a probe that died early cannot pass the checks. +const MIN_ROUNDS: usize = 50; + +/// The versions of the requests the probe sends. +const HEARTBEAT_VERSION: i16 = 2; +const DESCRIBE_GROUPS_VERSION: i16 = 5; +const FIND_COORDINATOR_VERSION: i16 = 4; +const METADATA_VERSION: i16 = 9; + +/// The answers that make a member join again, so their arrival is a rebalance. +const REJOIN: [i16; 4] = [ + ERROR_ILLEGAL_GENERATION, + ERROR_UNKNOWN_MEMBER_ID, + ERROR_REBALANCE_IN_PROGRESS, + ERROR_FENCED_INSTANCE_ID, +]; + +/// The cap of each client container, which the `timeout` inside it enforces. +const CLIENT_LIFETIME: Duration = Duration::from_mins(10); + +/// How long a client may take to exit after SIGTERM, and kcat to read the topic. +const CLIENT_EXIT_LIMIT: Duration = Duration::from_mins(1); + +/// The events of the clients that the tests count. +const ACKED: &str = r#""name":"producer_send_success""#; +const READ: &str = r#""name":"record_data""#; + +/// A rolling restart. Each gateway gets SIGTERM, as an orchestrator sends it, and comes back +/// before the next one stops. Its groups move to the other gateways at once, with their members, +/// so the Java group reads on without a rebalance. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_load_through_three_gateways_when_each_restarts_in_turn_should_write_and_read_each_record_once() + { + let Some(mut run) = Run::start(&[]).await else { + return; + }; + for node_id in run.rotation().await { + run.note(&format!("SIGTERM to gateway {node_id}")); + let stopped_at = Instant::now(); + run.deployment.terminate_gateway(node_id).await; + let moved = run.wait_served(stopped_at, node_id).await; + run.note(&format!("its groups moved {moved:?} after the SIGTERM")); + assert!( + moved <= HANDOVER_LIMIT, + "the groups of gateway {node_id} moved {moved:?} after its SIGTERM, above \ + {HANDOVER_LIMIT:?}" + ); + run.deployment.start_gateway(node_id).await; + run.wait_listed(node_id).await; + run.wait_for_progress(&format!("the load after gateway {node_id} came back")) + .await; + } + run.finish().await.assert_all(); +} + +/// Each Iggy node stops with SIGTERM and comes back: first the metadata primary, then the node +/// that holds the primary of the group's coordinator partition, then each node not stopped yet. +/// Once a node is back, the balancer of #4619 moves partition primaries to it, under load. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_load_through_three_gateways_when_each_node_stops_in_turn_should_write_and_read_each_record_once() + { + let Some(mut run) = Run::start(&[]).await else { + return; + }; + let mut stopped = Vec::with_capacity(NODES + 1); + while (0..NODES).any(|node| !stopped.contains(&node)) { + let (node, role) = run.next_node_to_stop(&stopped).await; + run.note(&format!("SIGTERM to node {node}, {role}")); + run.deployment.cluster.stop(node); + stopped.push(node); + run.wait_no_primary_on(node).await; + run.wait_for_progress(&format!("the load while node {node} is down")) + .await; + run.deployment.cluster.start(node).await; + run.wait_primary_on(node).await; + run.wait_for_progress(&format!("the load after node {node} came back")) + .await; + } + run.finish().await.assert_all(); +} + +/// Each gateway stops together with its node, with SIGTERM to both, as when their machine shuts +/// down. Both come back before the next pair stops. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_load_through_three_gateways_when_each_gateway_stops_with_its_node_should_write_and_read_each_record_once() + { + let Some(mut run) = Run::start(&[]).await else { + return; + }; + for node_id in run.rotation().await { + let node = Deployment::node_of(node_id); + run.note(&format!("SIGTERM to gateway {node_id} and node {node}")); + let stopped_at = Instant::now(); + run.deployment.terminate_gateway_and_node(node_id).await; + let moved = run.wait_served(stopped_at, node_id).await; + run.note(&format!("its groups moved {moved:?} after the SIGTERM")); + // A coordinator partition with its primary on the stopped node waits for a view change. + assert!( + moved <= HANDOVER_LIMIT, + "the groups of gateway {node_id} moved {moved:?} after its SIGTERM, above \ + {HANDOVER_LIMIT:?}" + ); + run.wait_no_primary_on(node).await; + run.wait_for_progress(&format!("the load while gateway {node_id} is down")) + .await; + run.deployment.cluster.start(node).await; + run.deployment.start_gateway(node_id).await; + run.wait_listed(node_id).await; + run.wait_primary_on(node).await; + run.wait_for_progress(&format!("the load after gateway {node_id} came back")) + .await; + } + run.finish().await.assert_all(); +} + +/// The gateway that coordinates the group crashes, with SIGKILL. The producer writes on through +/// the other gateways at once. The group moves only when Iggy ends the sessions of the dead +/// gateway, after the default session timeout of 36 s, and keeps its members and generation. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_load_through_three_gateways_when_the_coordinator_crashes_should_move_its_groups_after_the_session_timeout() + { + let Some(mut run) = Run::start(&[]).await else { + return; + }; + let takeover = run.crash_coordinator(Crash::Gateway).await; + assert!( + takeover <= SESSION_TIMEOUT + TAKEOVER_SLACK, + "the takeover took {takeover:?}, above the session timeout {SESSION_TIMEOUT:?} and \ + {TAKEOVER_SLACK:?}" + ); + run.finish().await.assert_all(); +} + +/// As the crash above, on nodes with the shortest session timeout that a 1 s +/// `consumer_group.heartbeat_interval` allows: 9.6 s, from an 8 s `heartbeat.interval`. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_a_short_session_timeout_when_the_coordinator_crashes_should_move_its_groups_within_it() + { + let Some(mut run) = Run::start(&SHORT_SESSION).await else { + return; + }; + let takeover = run.crash_coordinator(Crash::Gateway).await; + assert!( + takeover <= SHORT_SESSION_TIMEOUT + TAKEOVER_SLACK, + "the takeover took {takeover:?}, above the session timeout {SHORT_SESSION_TIMEOUT:?} \ + and {TAKEOVER_SLACK:?}" + ); + run.finish().await.assert_all(); +} + +/// The gateway that coordinates the group crashes together with its node, as when their machine +/// loses power. When that node held the metadata primary, a view change comes first, and the new +/// primary starts the session timeout again. The node recovers from its log when it comes back. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn given_a_short_session_timeout_when_the_coordinator_crashes_with_its_node_should_move_its_groups() + { + let Some(mut run) = Run::start(&SHORT_SESSION).await else { + return; + }; + let takeover = run.crash_coordinator(Crash::GatewayAndNode).await; + assert!( + takeover <= SHORT_SESSION_TIMEOUT + VIEW_CHANGE + TAKEOVER_SLACK, + "the takeover took {takeover:?}, above the session timeout {SHORT_SESSION_TIMEOUT:?}, \ + a view change {VIEW_CHANGE:?} and {TAKEOVER_SLACK:?}" + ); + run.finish().await.assert_all(); +} + +/// What crashes with SIGKILL in a crash test. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum Crash { + Gateway, + /// The gateway and the node beside it. + GatewayAndNode, +} + +/// One test's deployment, under load from the clients, watched by the probe. +struct Run { + deployment: Deployment, + producer: Container, + members: Vec<Container>, + probe: Probe, + /// The lines each member printed when the group settled. An assignment after that is a + /// rebalance. + settled: Vec<usize>, + started: Instant, + /// Last, so the processes stop before a failed test reports its logs. + work_dir: WorkDir, +} + +impl Run { + /// Deploys the nodes with `settings` and their gateways, settles the group, then starts the + /// probe and the producer. `None` when the host cannot run the clients. + async fn start(settings: &[(&str, &str)]) -> Option<Self> { + if let Some(reason) = + missing_prerequisite(&[KAFKA_IMAGE, KCAT_IMAGE]).or_else(missing_server) + { + skip(&reason); + return None; + } + let started = Instant::now(); + let work_dir = WorkDir::new(); + let deployment = Deployment::start(work_dir.path(), settings, &[(TOPIC, PARTITIONS)]).await; + note(started, "deployed three nodes and three gateways"); + let bootstrap = deployment.bootstrap(); + let logs = deployment.log_dir().to_path_buf(); + let members: Vec<Container> = (1..=MEMBERS) + .map(|member| start_member(&bootstrap, &logs.join(format!("member-{member}")))) + .collect(); + let settled = settle(&members).await; + note(started, "the group settled"); + let probe = Probe::start(deployment.addrs()).await; + let producer = start_producer(&bootstrap, &logs); + let run = Self { + deployment, + producer, + members, + probe, + settled, + started, + work_dir, + }; + run.wait_for_progress("the first records").await; + run.note("the load runs"); + Some(run) + } + + fn note(&self, what: &str) { + note(self.started, what); + } + + /// The gateways in the order that a rolling restart takes them: the coordinator of the group + /// first, so that the group moves at the first step. + async fn rotation(&self) -> Vec<u16> { + let first = self.coordinator().await; + let mut ids = self.deployment.gateway_ids(); + let at = ids + .iter() + .position(|node_id| *node_id == first) + .expect("the coordinator is a gateway"); + ids.rotate_left(at); + ids + } + + /// The gateway that coordinates the group, once every running gateway names the same one. + async fn coordinator(&self) -> u16 { + let addrs = self.deployment.running_addrs(); + wait_until( + RECOVERY, + PROBE_INTERVAL, + "agreement on the coordinator", + || async { + let mut named = BTreeSet::new(); + for addr in &addrs { + named.insert(named_coordinator(*addr, GROUP).await?); + } + let node_id = *named.first()?; + (named.len() == 1).then_some(u16::try_from(node_id).ok()?) + }, + ) + .await + } + + /// The node to stop next, with its role: first the metadata primary, then the node that holds + /// the primary of the group's coordinator partition, then each node not stopped yet. + async fn next_node_to_stop(&self, stopped: &[usize]) -> (usize, &'static str) { + if stopped.is_empty() { + let primary = wait_until(RECOVERY, PROBE_INTERVAL, "a metadata primary", || { + self.deployment.metadata_primary() + }) + .await; + return (primary, "the metadata primary"); + } + if stopped.len() == 1 { + let stream = cluster_id_stream(KAFKA_STREAM); + let partition = partition_of_group(GROUP); + let primary = wait_until( + RECOVERY, + PROBE_INTERVAL, + "the primary of the coordinator partition", + || async { + let primaries = self + .deployment + .primaries(&stream, COORDINATOR_LOG_TOPIC, [partition]) + .await?; + primaries.first().copied() + }, + ) + .await; + return (primary, "the primary of the coordinator partition"); + } + let node = (0..NODES) + .find(|node| !stopped.contains(node)) + .expect("a node not stopped yet"); + (node, "a node not stopped yet") + } + + /// Kills the gateway that coordinates the group, with its node for [`Crash::GatewayAndNode`], + /// and restarts what it killed once another gateway took the group over. Returns how long + /// after the kill the takeover came. + async fn crash_coordinator(&mut self, crash: Crash) -> Duration { + let node_id = self.coordinator().await; + let node = Deployment::node_of(node_id); + self.note(&format!( + "SIGKILL to gateway {node_id}, the coordinator ({crash:?})" + )); + let crashed_at = Instant::now(); + self.deployment.kill_gateway(node_id); + // A lost metadata primary stalls every partition until the next one is elected, since a + // node refuses partition requests on metadata evidence older than 8 s. + let data_path_limit = match crash { + Crash::Gateway => DATA_PATH_LIMIT, + Crash::GatewayAndNode => { + self.deployment.cluster.crash(node); + RECOVERY + } + }; + self.wait_for_acks("writes through the other gateways", data_path_limit) + .await; + let takeover = self.wait_served(crashed_at, node_id).await; + self.note(&format!("its groups moved {takeover:?} after the crash")); + if crash == Crash::GatewayAndNode { + self.deployment.cluster.start(node).await; + } + self.deployment.start_gateway(node_id).await; + self.wait_listed(node_id).await; + if crash == Crash::GatewayAndNode { + self.wait_primary_on(node).await; + } + self.wait_for_progress("the load after the crash").await; + takeover + } + + /// Waits until the producer gets more acknowledgements and every member reads more records. + async fn wait_for_progress(&self, what: &str) { + self.wait_for_acks(what, RECOVERY).await; + let before: Vec<usize> = self + .members + .iter() + .map(|member| member.count(READ)) + .collect(); + wait_until(RECOVERY, PROBE_INTERVAL, what, || async { + self.members + .iter() + .zip(&before) + .all(|(member, before)| member.count(READ) >= before + READ_PROGRESS) + .then_some(()) + }) + .await; + } + + /// Waits until the producer gets [`ACK_PROGRESS`] more acknowledgements, at most `limit`. + async fn wait_for_acks(&self, what: &str, limit: Duration) { + let before = self.producer.count(ACKED); + wait_until(limit, PROBE_INTERVAL, what, || async { + (self.producer.count(ACKED) >= before + ACK_PROGRESS).then_some(()) + }) + .await; + } + + /// Waits until a gateway other than `gone` answers the probe member as its coordinator, in a + /// round that started at `since` or later. Returns how long after `since` that round started. + async fn wait_served(&self, since: Instant, gone: u16) -> Duration { + let what = format!("a coordinator other than gateway {gone}"); + wait_until(RECOVERY, PROBE_INTERVAL, &what, || async { + self.probe + .first_served(since, usize::from(gone) - 1) + .map(|at| at.saturating_duration_since(since)) + }) + .await + } + + /// Waits until every running gateway lists gateway `node_id` in `Metadata`. A gateway lists + /// another once that one owns coordinator partitions, so a restarted gateway is back then. + async fn wait_listed(&self, node_id: u16) { + let addrs = self.deployment.running_addrs(); + let what = format!("every gateway listing gateway {node_id}"); + wait_until(RECOVERY, PROBE_INTERVAL, &what, || async { + for addr in &addrs { + if !listed_brokers(*addr).await?.contains(&i32::from(node_id)) { + return None; + } + } + Some(()) + }) + .await; + } + + /// Waits until node `node`, which stopped, leads no partition under load: its partitions + /// changed view and have primaries elsewhere. + async fn wait_no_primary_on(&self, node: usize) { + let what = format!("new primaries for the partitions of node {node}"); + wait_until(RECOVERY, Duration::from_secs(1), &what, || async { + let primaries = self.loaded_primaries().await?; + (!primaries.contains(&node)).then_some(()) + }) + .await; + } + + /// Waits until the balancer of #4619 moves a primary of a partition under load to node + /// `node`, which came back with none. The balancer evens out the count of primaries over all + /// partitions, so the move can be of the topic or of the coordinator log. + async fn wait_primary_on(&self, node: usize) { + let what = format!("a primary move to node {node}"); + let primaries = wait_until( + PRIMARY_MOVE_LIMIT, + Duration::from_secs(1), + &what, + || async { + let primaries = self.loaded_primaries().await?; + primaries.contains(&node).then_some(primaries) + }, + ) + .await; + let held: Vec<usize> = (0..NODES) + .map(|replica| { + primaries + .iter() + .filter(|primary| **primary == replica) + .count() + }) + .collect(); + self.note(&format!( + "primaries of the topic and the coordinator log per node: {held:?}" + )); + } + + /// The primary of each partition under load: those of the topic, then those of the + /// coordinator log. + async fn loaded_primaries(&self) -> Option<Vec<usize>> { + let mut primaries = self + .deployment + .primaries(KAFKA_STREAM, TOPIC, 0..PARTITIONS) + .await?; + let coordinator_log = self + .deployment + .primaries( + &cluster_id_stream(KAFKA_STREAM), + COORDINATOR_LOG_TOPIC, + 0..COORDINATOR_PARTITIONS, + ) + .await?; + primaries.extend(coordinator_log); + Some(primaries) + } + + /// Stops the producer, waits until the group read everything it acknowledged, stops the + /// probe and the members, and reads back what the partitions store. + async fn finish(self) -> Outcome { + let Self { + deployment, + mut producer, + mut members, + probe, + settled, + started, + work_dir, + } = self; + note(started, "stopping the producer"); + producer.signal("TERM"); + assert!( + producer.wait_exit(CLIENT_EXIT_LIMIT).await, + "the producer exits after SIGTERM" + ); + let produced = producer.with_stdout(Produced::parse); + wait_until( + RECOVERY, + Duration::from_secs(1), + "the group reading every acknowledged record", + || async { + let read: BTreeSet<u64> = members + .iter() + .flat_map(|member| member.with_stdout(read_values)) + .collect(); + produced + .acked + .keys() + .all(|value| read.contains(value)) + .then_some(()) + }, + ) + .await; + let rounds = probe.stop().await; + let printed: Vec<Vec<String>> = members + .iter() + .map(|member| member.with_stdout(<[String]>::to_vec)) + .collect(); + for member in &members { + member.signal("TERM"); + } + for member in &mut members { + assert!( + member.wait_exit(CLIENT_EXIT_LIMIT).await, + "a member exits after SIGTERM" + ); + } + note(started, "reading the partitions back with kcat"); + let stored = stored_records(&deployment.bootstrap()).await; + let listings = broker_listings(&deployment).await; + note(started, "done"); + Outcome { + produced, + stored, + read: printed + .iter() + .flat_map(|lines| read_values(lines)) + .collect(), + rebalances: rebalances(&printed, &settled), + rounds, + listings, + _work_dir: work_dir, + } + } +} + +/// What a test saw, checked once every client stopped. +struct Outcome { + produced: Produced, + /// Every stored record, as `(partition, offset, value)`. + stored: Vec<(i64, i64, u64)>, + /// The value of each record the group read, once per read. + read: Vec<u64>, + /// The assignment events that came after the group settled. + rebalances: Vec<String>, + rounds: Vec<Round>, + /// What `kcat -L` printed through each gateway, with the line for each gateway it must list. + listings: Vec<(String, Vec<String>)>, + _work_dir: WorkDir, +} + +impl Outcome { + fn assert_all(&self) { + self.assert_each_record_once(); + self.assert_no_rebalance(); + self.assert_one_coordinator_at_a_time(); + self.assert_every_gateway_listed(); + } + + /// Every acknowledged record is stored once, at the partition and offset that the producer + /// got, and the group reads it once. + fn assert_each_record_once(&self) { + let errors = &self.produced.errors; + assert!( + errors.is_empty(), + "{} sends failed: {errors:#?}", + errors.len() + ); + let acked = &self.produced.acked; + assert!(!acked.is_empty(), "the producer must write records"); + let mut stored: BTreeMap<u64, Vec<(i64, i64)>> = BTreeMap::new(); + for &(partition, offset, value) in &self.stored { + stored.entry(value).or_default().push((partition, offset)); + } + let twice: Vec<_> = stored + .iter() + .filter(|(_, at)| at.len() > 1) + .take(20) + .collect(); + assert!( + twice.is_empty(), + "records stored more than once, as value and (partition, offset): {twice:?}" + ); + let misplaced: Vec<_> = acked + .iter() + .filter(|(value, at)| stored.get(*value).is_none_or(|stored| stored != &[**at])) + .map(|(value, at)| (value, at, stored.get(value))) + .take(20) + .collect(); + assert!( + misplaced.is_empty(), + "acknowledged records not stored where the producer was told, as value, acknowledged \ + (partition, offset) and stored ones: {misplaced:?}" + ); + let mut reads: BTreeMap<u64, usize> = BTreeMap::new(); + for value in &self.read { + *reads.entry(*value).or_default() += 1; + } + let read_twice: Vec<_> = reads + .iter() + .filter(|(_, count)| **count > 1) + .take(20) + .collect(); + assert!( + read_twice.is_empty(), + "records the group read more than once, as value and reads: {read_twice:?}" + ); + let unread: Vec<&u64> = acked + .keys() + .filter(|value| !reads.contains_key(value)) + .take(20) + .collect(); + assert!( + unread.is_empty(), + "acknowledged records the group never read: {unread:?}" + ); + eprintln!("{} records acknowledged, stored and read once", acked.len()); + } + + /// The group keeps its members and generation: no member gets a new assignment, and the + /// coordinator describes the group as stable every time. + fn assert_no_rebalance(&self) { + assert!( + self.rebalances.is_empty(), + "the group rebalanced after it settled: {:#?}", + self.rebalances + ); + let unstable: Vec<&(i16, String)> = self + .rounds + .iter() + .flat_map(|round| round.descriptions.iter().flatten()) + .filter(|(code, state)| *code == ERROR_NONE && state != "Stable") + .collect(); + assert!( + unstable.is_empty(), + "the coordinator described the group as not stable: {unstable:?}" + ); + } + + /// No two gateways answer as coordinator of a group in one probe round, and no answer makes + /// the probe member join again. + fn assert_one_coordinator_at_a_time(&self) { + assert!( + self.rounds.len() >= MIN_ROUNDS, + "the probe ran only {} rounds", + self.rounds.len() + ); + let coordinating = |answers: &[Option<i16>]| { + answers + .iter() + .filter(|code| **code == Some(ERROR_NONE)) + .count() + }; + for round in &self.rounds { + let described: Vec<Option<i16>> = round + .descriptions + .iter() + .map(|answer| answer.as_ref().map(|(code, _)| *code)) + .collect(); + assert!( + coordinating(&round.heartbeats) <= 1 && coordinating(&described) <= 1, + "two gateways answered as coordinator at once: {round:?}" + ); + assert!( + !round + .heartbeats + .iter() + .flatten() + .any(|code| REJOIN.contains(code)), + "a gateway told the probe member to join again: {round:?}" + ); + } + } + + /// Each gateway lists all three at the end, as kcat sees it. + fn assert_every_gateway_listed(&self) { + for (listing, expected) in &self.listings { + for line in expected { + assert!( + listing.contains(line.as_str()), + "`kcat -L` must show `{line}`:\n{listing}" + ); + } + } + } +} + +/// What the producer printed. +#[derive(Default)] +struct Produced { + /// Where each acknowledged record went, by value, as `(partition, offset)`. + acked: BTreeMap<u64, (i64, i64)>, + errors: Vec<String>, +} + +impl Produced { + fn parse(lines: &[String]) -> Self { + let mut produced = Self::default(); + for line in lines { + let Some(event) = parse_event(line) else { + continue; + }; + match event["name"].as_str() { + Some("producer_send_success") => { + if let (Some(value), Some(partition), Some(offset)) = ( + event["value"].as_str().and_then(|value| value.parse().ok()), + event["partition"].as_i64(), + event["offset"].as_i64(), + ) { + produced.acked.insert(value, (partition, offset)); + } + } + Some("producer_send_error") => produced.errors.push(line.clone()), + _ => {} + } + } + produced + } +} + +/// The answers of one probe round, one per gateway in node id order. `None` where a gateway did +/// not answer. +#[derive(Debug, Clone)] +struct Round { + at: Instant, + /// The error code of the probe member's heartbeat. + heartbeats: Vec<Option<i16>>, + /// The error code and state of the Java group in `DescribeGroups`. + descriptions: Vec<Option<(i16, String)>>, +} + +/// A member of a second group that heartbeats at every gateway, and asks every gateway to describe +/// the Java group, until it stops. +struct Probe { + rounds: Arc<Mutex<Vec<Round>>>, + stop: CancellationToken, + task: JoinHandle<()>, +} + +impl Probe { + /// Joins a member to a group in the coordinator partition of the Java group, then starts the + /// rounds. + async fn start(addrs: Vec<SocketAddr>) -> Self { + let group = probe_group(); + let coordinator = wait_until( + RECOVERY, + PROBE_INTERVAL, + "the coordinator of the probe group", + || coordinator_addr(&addrs, &group), + ) + .await; + let (member_id, generation) = gateways::join_and_sync(coordinator, &group).await; + let rounds = Arc::new(Mutex::new(Vec::new())); + let stop = CancellationToken::new(); + let member = ProbeMember { + group, + member_id, + generation, + }; + let task = tokio::spawn(probe_rounds( + addrs, + member, + Arc::clone(&rounds), + stop.clone(), + )); + Self { rounds, stop, task } + } + + /// When the first round at `since` or later started in which a gateway other than the one at + /// index `gone` answered the probe member as its coordinator. + fn first_served(&self, since: Instant, gone: usize) -> Option<Instant> { + self.rounds + .lock() + .unwrap_or_else(PoisonError::into_inner) + .iter() + .filter(|round| round.at >= since) + .find(|round| { + round + .heartbeats + .iter() + .enumerate() + .any(|(index, code)| index != gone && *code == Some(ERROR_NONE)) + }) + .map(|round| round.at) + } + + async fn stop(self) -> Vec<Round> { + self.stop.cancel(); + self.task.await.expect("the probe does not panic"); + std::mem::take(&mut *self.rounds.lock().unwrap_or_else(PoisonError::into_inner)) + } +} + +struct ProbeMember { + group: String, + member_id: StrBytes, + generation: i32, +} + +async fn probe_rounds( + addrs: Vec<SocketAddr>, + member: ProbeMember, + rounds: Arc<Mutex<Vec<Round>>>, + stop: CancellationToken, +) { + let mut ticks = tokio::time::interval(PROBE_INTERVAL); + ticks.set_missed_tick_behavior(MissedTickBehavior::Delay); + loop { + tokio::select! { + () = stop.cancelled() => return, + _ = ticks.tick() => {} + } + let at = Instant::now(); + let heartbeats = join_all(addrs.iter().map(|addr| heartbeat(*addr, &member))).await; + let descriptions = join_all(addrs.iter().map(|addr| describe(*addr, GROUP))).await; + rounds + .lock() + .unwrap_or_else(PoisonError::into_inner) + .push(Round { + at, + heartbeats, + descriptions, + }); + } +} + +/// A group in the coordinator partition of the Java group, so one gateway coordinates both. +fn probe_group() -> String { + let partition = partition_of_group(GROUP); + // A thousand names cover the 16 partitions many times over. + (0..1_000) + .map(|n| format!("failover-probe-{n}")) + .find(|group| partition_of_group(group) == partition) + .expect("a group in every partition") +} + +/// The test's directory. A failed test keeps the logs there, drops the node data, and prints where +/// they are with the end of each gateway log. +struct WorkDir(Option<TempDir>); + +impl WorkDir { + fn new() -> Self { + let dir = tempfile::Builder::new() + .prefix("iggy-kafka-failover-") + .tempdir() + .expect("create the work dir"); + Self(Some(dir)) + } + + fn path(&self) -> &Path { + self.0.as_ref().expect("a work dir").path() + } +} + +impl Drop for WorkDir { + fn drop(&mut self) { + if !std::thread::panicking() { + return; + } + let Some(dir) = self.0.take() else { + return; + }; + let path = dir.keep(); + let _ = fs::remove_dir_all(path.join("data")); + let logs = path.join("logs"); + let mut report = vec![format!( + "the failed test kept its logs in {}", + logs.display() + )]; + for node_id in 1..=NODES { + let log = logs.join(format!("gateway-{node_id}.log")); + report.push(format!( + "--- gateway {node_id} ---\n{}", + tail(&log, TAIL_LINES) + )); + } + for client in ["producer", "member-1", "member-2"] { + let log = logs.join(format!("{client}.err")); + report.push(format!("--- {client} stderr ---\n{}", tail(&log, 10))); + } + eprintln!("{}", report.join("\n")); + } +} + +fn note(started: Instant, what: &str) { + eprintln!("[{:>6.1} s] {what}", started.elapsed().as_secs_f64()); +} + +/// Why the cluster cannot start, or `None` when it can: `iggy-server` must be built into the +/// target directory of this test binary. +fn missing_server() -> Option<String> { + let binary = format!("iggy-server{}", std::env::consts::EXE_SUFFIX); + let built = std::env::current_exe() + .ok()? + .ancestors() + .any(|dir| dir.join(&binary).is_file()); + (!built).then(|| "iggy-server is not built; run `cargo build --bin iggy-server` first".into()) +} + +/// Polls `check` every `interval` until it returns a value, and fails the test after `limit`. +async fn wait_until<T, F: Future<Output = Option<T>>>( + limit: Duration, + interval: Duration, + what: &str, + mut check: impl FnMut() -> F, +) -> T { + let deadline = Instant::now() + limit; + loop { + if let Some(value) = check().await { + return value; + } + assert!( + Instant::now() < deadline, + "{what} did not happen within {limit:?}" + ); + tokio::time::sleep(interval).await; + } +} + +/// A member of the group, which commits each batch after it reads it, prints a JSON event for each +/// step and one for each record. +fn start_member(bootstrap: &str, log: &Path) -> Container { + Container::start( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-verifiable-consumer.sh", + "--bootstrap-server", + bootstrap, + "--topic", + TOPIC, + "--group-id", + GROUP, + "--reset-policy", + "earliest", + "--assignment-strategy", + "org.apache.kafka.clients.consumer.RangeAssignor", + "--verbose", + ], + &[], + CLIENT_LIFETIME, + log, + ) +} + +/// The idempotent producer, which writes the numbers from 0 as values until it gets SIGTERM, and +/// prints a JSON event for each record it sent. +fn start_producer(bootstrap: &str, logs: &Path) -> Container { + let config = logs.join("producer.properties"); + fs::write(&config, PRODUCER_CONFIG).expect("write the producer config"); + let rate = RATE.to_string(); + Container::start( + KAFKA_IMAGE, + &[ + "/opt/kafka/bin/kafka-verifiable-producer.sh", + "--bootstrap-server", + bootstrap, + "--topic", + TOPIC, + "--throughput", + &rate, + "--repeating-keys", + KEYS, + "--producer.config", + "/tmp/producer.properties", + ], + &[(&config, "/tmp/producer.properties")], + CLIENT_LIFETIME, + &logs.join("producer"), + ) +} + +/// Waits until each member's latest assignment is its share of the partitions, and returns how +/// many lines each had printed by then. +async fn settle(members: &[Container]) -> Vec<usize> { + wait_until( + RECOVERY, + PROBE_INTERVAL, + "the group's first assignment", + || async { + let mut held = BTreeSet::new(); + let mut printed = Vec::with_capacity(members.len()); + for member in members { + let (lines, assigned) = + member.with_stdout(|lines| (lines.len(), last_assignment(lines))); + for partition in assigned? { + if !held.insert(partition) { + return None; + } + } + printed.push(lines); + } + (held.len() == usize::try_from(PARTITIONS).ok()?).then_some(printed) + }, + ) + .await +} + +/// The partitions of the last `partitions_assigned` event in `lines`. +fn last_assignment(lines: &[String]) -> Option<Vec<i64>> { + let event = lines + .iter() + .rev() + .filter(|line| line.contains(r#""name":"partitions_assigned""#)) + .find_map(|line| parse_event(line))?; + event["partitions"] + .as_array()? + .iter() + .map(|partition| partition["partition"].as_i64()) + .collect() +} + +/// The value of each record a member read, in the order read. +fn read_values(lines: &[String]) -> Vec<u64> { + lines + .iter() + .filter(|line| line.contains(READ)) + .filter_map(|line| parse_event(line)?["value"].as_str()?.parse().ok()) + .collect() +} + +/// The assignment events that each member printed between the settled group and its stop. +fn rebalances(printed: &[Vec<String>], settled: &[usize]) -> Vec<String> { + printed + .iter() + .zip(settled) + .enumerate() + .flat_map(|(member, (lines, settled))| { + lines[*settled..] + .iter() + .filter(|line| { + line.contains(r#""name":"partitions_assigned""#) + || line.contains(r#""name":"partitions_revoked""#) + }) + .map(move |line| format!("member {}: {line}", member + 1)) + }) + .collect() +} + +fn parse_event(line: &str) -> Option<Value> { + line.starts_with('{') + .then(|| serde_json::from_str(line).ok()) + .flatten() +} + +/// Every record that the partitions store, as `(partition, offset, value)`, read with kcat +/// through the gateways. +async fn stored_records(bootstrap: &str) -> Vec<(i64, i64, u64)> { + let args: Vec<String> = [ + "-b", + bootstrap, + "-C", + "-t", + TOPIC, + "-o", + "beginning", + "-e", + "-q", + "-f", + "%p %o %s\n", + ] + .map(String::from) + .to_vec(); + let read = + tokio::task::spawn_blocking(move || run_to_end(KCAT_IMAGE, &args, CLIENT_EXIT_LIMIT)) + .await + .expect("kcat runs"); + assert!(read.succeeded, "kcat reads the topic:\n{}", read.stderr); + read.stdout + .lines() + .filter_map(|line| { + let mut fields = line.split_whitespace(); + Some(( + fields.next()?.parse().ok()?, + fields.next()?.parse().ok()?, + fields.next()?.parse().ok()?, + )) + }) + .collect() +} + +/// What `kcat -L` prints through each gateway, with the line for each gateway it must list. +async fn broker_listings(deployment: &Deployment) -> Vec<(String, Vec<String>)> { + let addrs = deployment.addrs(); + let expected: Vec<String> = deployment + .gateway_ids() + .iter() + .zip(&addrs) + .map(|(node_id, addr)| format!("broker {node_id} at {addr}")) + .collect(); + let mut listings = Vec::with_capacity(addrs.len()); + for addr in addrs { + let args = vec!["-b".to_string(), addr.to_string(), "-L".to_string()]; + let listed = + tokio::task::spawn_blocking(move || run_to_end(KCAT_IMAGE, &args, CLIENT_EXIT_LIMIT)) + .await + .expect("kcat runs"); + assert!( + listed.succeeded, + "kcat -L through {addr}:\n{}", + listed.stderr + ); + listings.push((listed.stdout, expected.clone())); + } + listings +} + +/// The node ids that the gateway at `addr` lists in `Metadata`. +async fn listed_brokers(addr: SocketAddr) -> Option<Vec<i32>> { + let request = MetadataRequest::default().with_topics(Some(Vec::new())); + let body = wire::encode(&request, METADATA_VERSION); + let answer: MetadataResponse = + exchange(addr, API_KEY_METADATA, METADATA_VERSION, &body).await?; + Some( + answer + .brokers + .iter() + .map(|broker| broker.node_id.0) + .collect(), + ) +} + +/// The node id that the gateway at `addr` names as coordinator of `group`. +async fn named_coordinator(addr: SocketAddr, group: &str) -> Option<i32> { + let request = FindCoordinatorRequest::default() + .with_coordinator_keys(vec![StrBytes::from_string(group.to_string())]); + let body = wire::encode(&request, FIND_COORDINATOR_VERSION); + let answer: FindCoordinatorResponse = exchange( + addr, + API_KEY_FIND_COORDINATOR, + FIND_COORDINATOR_VERSION, + &body, + ) + .await?; + let coordinator = answer.coordinators.first()?; + (coordinator.error_code == ERROR_NONE).then_some(coordinator.node_id.0) +} + +/// The address of the gateway that coordinates `group`, as the first gateway that names one says. +async fn coordinator_addr(addrs: &[SocketAddr], group: &str) -> Option<SocketAddr> { + for addr in addrs { + if let Some(node_id) = named_coordinator(*addr, group).await { + let index = usize::try_from(node_id).ok()?.checked_sub(1)?; + return addrs.get(index).copied(); + } + } + None +} + +async fn heartbeat(addr: SocketAddr, member: &ProbeMember) -> Option<i16> { + let request = HeartbeatRequest::default() + .with_group_id(GroupId(StrBytes::from_string(member.group.clone()))) + .with_generation_id(member.generation) + .with_member_id(member.member_id.clone()); + let body = wire::encode(&request, HEARTBEAT_VERSION); + let answer: HeartbeatResponse = + exchange(addr, API_KEY_HEARTBEAT, HEARTBEAT_VERSION, &body).await?; + Some(answer.error_code) +} + +async fn describe(addr: SocketAddr, group: &str) -> Option<(i16, String)> { + let request = DescribeGroupsRequest::default() + .with_groups(vec![GroupId(StrBytes::from_string(group.to_string()))]); + let body = wire::encode(&request, DESCRIBE_GROUPS_VERSION); + let answer: DescribeGroupsResponse = exchange( + addr, + API_KEY_DESCRIBE_GROUPS, + DESCRIBE_GROUPS_VERSION, + &body, + ) + .await?; + let described = answer.groups.first()?; + Some((described.error_code, described.group_state.to_string())) +} + +/// Sends one request on a connection of its own and decodes the answer. `None` when nothing +/// accepts at `addr`, the gateway closes the connection, or no answer comes within +/// [`PROBE_TIMEOUT`]. +async fn exchange<A: Decodable>( + addr: SocketAddr, + key: i16, + version: i16, + body: &[u8], +) -> Option<A> { + let frame = tcp::build_request_frame(key, version, 1, Some("failover-probe"), body); + let answer = async { + let mut stream = TcpStream::connect(addr).await.ok()?; + stream.write_all(&frame).await.ok()?; + let mut length = [0; 4]; + stream.read_exact(&mut length).await.ok()?; + let mut payload = vec![0; usize::try_from(i32::from_be_bytes(length)).ok()?]; + stream.read_exact(&mut payload).await.ok()?; + let (_, mut body) = tcp::parse_response_payload(key, version, Bytes::from(payload)); + A::decode(&mut body, version).ok() + }; + tokio::time::timeout(PROBE_TIMEOUT, answer) + .await + .ok() + .flatten() +} diff --git a/gateways/kafka/tests/common/deployment.rs b/gateways/kafka/tests/common/deployment.rs new file mode 100644 index 000000000..9ba14ed6e --- /dev/null +++ b/gateways/kafka/tests/common/deployment.rs @@ -0,0 +1,417 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Three Iggy nodes and a gateway process beside each, deployed as the gateway documentation +//! describes it. Gateway `n` dials node `n - 1`, and every gateway lists all of them in +//! `IGGY_KAFKA_BROKERS`. The gateways run the compiled binary, so a stop is a SIGTERM and a crash +//! a SIGKILL, as an orchestrator sends them. +//! +//! Callers must also declare the `codec`, `gateways`, `iggy_server`, `tcp` and `wire` modules at +//! their crate root, as for `gateways.rs`. +#![allow(dead_code)] + +use std::fs::{self, OpenOptions}; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command}; +use std::time::Duration; + +use iggy::prelude::{ + AutoLogin, Client, ClusterClient, ClusterNodeRole, Credentials, IggyClient, IggyClientBuilder, + UserClient, +}; +use iggy_binary_protocol::codes::GET_PARTITION_ROUTE_CODE; +use iggy_binary_protocol::requests::messages::GetPartitionRouteRequest; +use iggy_binary_protocol::responses::messages::PartitionRouteResponse; +use iggy_binary_protocol::{WireDecode, WireEncode, WireIdentifier}; +use secrecy::SecretString; +use tokio::time::Instant; + +use super::gateways; +use super::iggy_server::{PortGuard, TestCluster, graceful_kill}; + +/// Iggy nodes, each with a gateway beside it. +pub const NODES: usize = 3; + +/// The stream of the Kafka topics under the default topic mapping. +pub const KAFKA_STREAM: &str = "kafka"; + +/// How long a gateway may take to listen. Its bridge waits up to 60 s for the cluster's metadata +/// primary (`IGGY_STARTUP_LIMIT` in `main.rs`). +const START_LIMIT: Duration = Duration::from_secs(70); + +/// How long a gateway may take to exit after SIGTERM: the hand-over of its groups, the drain of +/// its connections and the sign-out of its Iggy clients. The default drain alone may take 25 s. +const EXIT_LIMIT: Duration = Duration::from_secs(40); + +/// How long one question to the cluster may take. +const QUERY_LIMIT: Duration = Duration::from_secs(10); + +/// The log lines of each gateway that a failure report shows. +pub const TAIL_LINES: usize = 40; + +/// The nodes and their gateways, until the test drops it. +pub struct Deployment { + pub cluster: TestCluster, + gateways: Vec<GatewayProcess>, + log_dir: PathBuf, +} + +impl Deployment { + /// Starts the nodes with `settings`, as `IGGY_*` variables, creates each `(topic, + /// partitions)` of `topics`, then starts a gateway beside each node. The node data goes to + /// `data/` in `work_dir`, and every log to `logs/`. + pub async fn start(work_dir: &Path, settings: &[(&str, &str)], topics: &[(&str, u32)]) -> Self { + let log_dir = work_dir.join("logs"); + fs::create_dir_all(&log_dir).expect("create the log dir"); + let cluster = + TestCluster::spawn_with(&work_dir.join("data"), NODES, settings, Some(&log_dir)).await; + let bridge = gateways::connect(cluster.bridge_config(0)).await; + for &(topic, partitions) in topics { + bridge + .create_kafka_topic(topic, partitions) + .await + .expect("create a topic"); + } + bridge.close().await.expect("close the bridge"); + + let ports: Vec<PortGuard> = (0..NODES).map(|_| PortGuard::acquire()).collect(); + let brokers = (1_u16..) + .zip(&ports) + .map(|(node_id, port)| format!("{node_id}@127.0.0.1:{}", port.port)) + .collect::<Vec<_>>() + .join(","); + let mut gateways: Vec<GatewayProcess> = (1_u16..) + .zip(ports) + .enumerate() + .map(|(node, (node_id, port))| GatewayProcess { + node_id, + addr: SocketAddr::from(([127, 0, 0, 1], port.port)), + iggy_addr: cluster.address(node), + brokers: brokers.clone(), + log: log_dir.join(format!("gateway-{node_id}.log")), + child: None, + _port: port, + }) + .collect(); + for gateway in &mut gateways { + gateway.spawn(); + } + for gateway in &mut gateways { + gateway.wait_listening().await; + } + Self { + cluster, + gateways, + log_dir, + } + } + + pub fn log_dir(&self) -> &Path { + &self.log_dir + } + + /// The node ids of the gateways, in order. + pub fn gateway_ids(&self) -> Vec<u16> { + self.gateways + .iter() + .map(|gateway| gateway.node_id) + .collect() + } + + /// The address of every gateway, stopped ones included, in node id order. + pub fn addrs(&self) -> Vec<SocketAddr> { + self.gateways.iter().map(|gateway| gateway.addr).collect() + } + + /// The address of each running gateway. + pub fn running_addrs(&self) -> Vec<SocketAddr> { + self.gateways + .iter() + .filter(|gateway| gateway.child.is_some()) + .map(|gateway| gateway.addr) + .collect() + } + + /// `bootstrap.servers` for a client: every gateway, as the deployment guide says. + pub fn bootstrap(&self) -> String { + self.addrs() + .iter() + .map(ToString::to_string) + .collect::<Vec<_>>() + .join(",") + } + + /// The node of gateway `node_id`: gateway `n` runs beside node `n - 1`. + pub fn node_of(node_id: u16) -> usize { + usize::from(node_id) - 1 + } + + /// Stops gateway `node_id` with SIGTERM and waits until it exits. Returns how long the exit + /// took. Fails the test when the gateway is still running after [`EXIT_LIMIT`]. + pub async fn terminate_gateway(&mut self, node_id: u16) -> Duration { + let gateway = self.gateway_mut(node_id); + gateway.signal(libc::SIGTERM); + gateway.wait_exit().await + } + + /// Stops gateway `node_id` and its node with SIGTERM at once, as when their machine shuts + /// down, and waits until both exit. + pub async fn terminate_gateway_and_node(&mut self, node_id: u16) { + self.gateway_mut(node_id).signal(libc::SIGTERM); + self.cluster.stop(Self::node_of(node_id)); + self.gateway_mut(node_id).wait_exit().await; + } + + /// Kills gateway `node_id` with SIGKILL, as a crash would. Its connections close, but Iggy + /// keeps its sessions until their timeout. + pub fn kill_gateway(&mut self, node_id: u16) { + let gateway = self.gateway_mut(node_id); + let mut child = gateway.child.take().expect("a running gateway"); + child.kill().expect("SIGKILL the gateway"); + child.wait().expect("reap the gateway"); + } + + /// Starts gateway `node_id` again on its port, and waits until it listens. + pub async fn start_gateway(&mut self, node_id: u16) { + let gateway = self.gateway_mut(node_id); + gateway.spawn(); + gateway.wait_listening().await; + } + + /// The replica id of the metadata primary, as a running node reports it, or `None` when no + /// node answers. + pub async fn metadata_primary(&self) -> Option<usize> { + let client = self.client().await?; + let metadata = tokio::time::timeout(QUERY_LIMIT, client.get_cluster_metadata()).await; + sign_out(&client).await; + let leader = metadata + .ok()? + .ok()? + .nodes + .into_iter() + .find(|node| node.role == ClusterNodeRole::Leader)?; + Some(usize::from(leader.replica_id)) + } + + /// The replica id of the primary of each of `partition_ids` of `topic` in `stream`, as a + /// running node reports them, or `None` when no node answers. + pub async fn primaries( + &self, + stream: &str, + topic: &str, + partition_ids: impl IntoIterator<Item = u32>, + ) -> Option<Vec<usize>> { + let client = self.client().await?; + let routes = async { + let mut primaries = Vec::new(); + for partition_id in partition_ids { + let request = GetPartitionRouteRequest { + stream_id: WireIdentifier::named(stream).ok()?, + topic_id: WireIdentifier::named(topic).ok()?, + partition_id, + } + .to_bytes(); + let reply = tokio::time::timeout( + QUERY_LIMIT, + client.send_binary_request(GET_PARTITION_ROUTE_CODE, request), + ) + .await + .ok()? + .ok()?; + let route = PartitionRouteResponse::decode_from(&reply).ok()?; + primaries.push(usize::from(route.primary_replica)); + } + Some(primaries) + } + .await; + sign_out(&client).await; + routes + } + + /// The last [`TAIL_LINES`] lines of the log of each gateway. + pub fn gateway_log_tails(&self) -> String { + self.gateways + .iter() + .map(|gateway| { + format!( + "--- gateway {} ---\n{}", + gateway.node_id, + tail(&gateway.log, TAIL_LINES) + ) + }) + .collect::<Vec<_>>() + .join("\n") + } + + fn gateway_mut(&mut self, node_id: u16) -> &mut GatewayProcess { + self.gateways + .iter_mut() + .find(|gateway| gateway.node_id == node_id) + .expect("a listed gateway") + } + + /// A client signed in to the first running node that lets it in. + async fn client(&self) -> Option<IggyClient> { + let running = (0..self.cluster.node_count()).filter(|node| self.cluster.is_running(*node)); + for node in running { + let client = IggyClientBuilder::new() + .with_tcp() + .with_server_address(self.cluster.address(node)) + .with_auto_sign_in(AutoLogin::Enabled(Credentials::UsernamePassword( + "iggy".to_string(), + SecretString::from("iggy"), + ))) + .build() + .ok()?; + if matches!( + tokio::time::timeout(QUERY_LIMIT, client.connect()).await, + Ok(Ok(())) + ) { + return Some(client); + } + let _ = tokio::time::timeout(QUERY_LIMIT, client.shutdown()).await; + } + None + } +} + +/// The last `lines` lines of the file at `path`. +pub fn tail(path: &Path, lines: usize) -> String { + let text = fs::read(path).map_or_else( + |error| format!("cannot read {}: {error}", path.display()), + |bytes| String::from_utf8_lossy(&bytes).into_owned(), + ); + let all: Vec<&str> = text.lines().collect(); + all[all.len().saturating_sub(lines)..].join("\n") +} + +/// Ends the session of `client` in Iggy, so the clients table keeps no slot for it. +async fn sign_out(client: &IggyClient) { + let _ = tokio::time::timeout(QUERY_LIMIT, client.logout_user()).await; + let _ = tokio::time::timeout(QUERY_LIMIT, client.shutdown()).await; +} + +/// One `iggy-gateway-kafka` process, started again on the same port after a stop. +struct GatewayProcess { + node_id: u16, + addr: SocketAddr, + iggy_addr: String, + brokers: String, + log: PathBuf, + child: Option<Child>, + _port: PortGuard, +} + +impl GatewayProcess { + fn spawn(&mut self) { + assert!(self.child.is_none(), "gateway {} runs", self.node_id); + let log = OpenOptions::new() + .create(true) + .append(true) + .open(&self.log) + .expect("open the gateway log"); + let mut command = Command::new(env!("CARGO_BIN_EXE_iggy-gateway-kafka")); + // The gateway refuses to start on an unknown inherited IGGY_KAFKA_* variable, and a known + // one would change what the test runs. + command.env_clear(); + if let Some(llvm_profile_file) = std::env::var_os("LLVM_PROFILE_FILE") { + command.env("LLVM_PROFILE_FILE", llvm_profile_file); + } + let child = command + .env("IGGY_KAFKA_BIND_ADDR", self.addr.to_string()) + .env("IGGY_KAFKA_BRIDGE_ENABLED", "true") + .env("IGGY_KAFKA_IGGY_ADDR", &self.iggy_addr) + .env("IGGY_KAFKA_IGGY_USERNAME", "iggy") + .env("IGGY_KAFKA_IGGY_PASSWORD", "iggy") + .env("IGGY_KAFKA_NODE_ID", self.node_id.to_string()) + .env("IGGY_KAFKA_BROKERS", &self.brokers) + .env("IGGY_KAFKA_COORDINATOR_LOG", "true") + .stdout(log.try_clone().expect("share the gateway log")) + .stderr(log) + .spawn() + .expect("spawn iggy-gateway-kafka"); + self.child = Some(child); + } + + /// Waits until the gateway accepts connections. It binds its listener once its bridge signed + /// in, and joins the coordinator lease after that. + async fn wait_listening(&mut self) { + let deadline = Instant::now() + START_LIMIT; + loop { + let child = self.child.as_mut().expect("a started gateway"); + if let Some(status) = child.try_wait().expect("poll the gateway") { + self.child = None; + panic!( + "gateway {} exited during startup with {status}:\n{}", + self.node_id, + tail(&self.log, TAIL_LINES) + ); + } + if tokio::net::TcpStream::connect(self.addr).await.is_ok() { + return; + } + assert!( + Instant::now() < deadline, + "gateway {} did not listen within {START_LIMIT:?}:\n{}", + self.node_id, + tail(&self.log, TAIL_LINES) + ); + tokio::time::sleep(Duration::from_millis(100)).await; + } + } + + fn signal(&self, signal: libc::c_int) { + let child = self.child.as_ref().expect("a running gateway"); + let pid = libc::pid_t::try_from(child.id()).expect("a pid"); + // SAFETY: `pid` is a child of this process that nothing reaped yet, since only `wait_exit` + // and `kill_gateway` reap it, and both take it out of `self.child` first. + let sent = unsafe { libc::kill(pid, signal) }; + assert_eq!(sent, 0, "signal gateway {}", self.node_id); + } + + /// Waits until the gateway exits, and returns how long that took. Fails the test with its log + /// when it still runs after [`EXIT_LIMIT`]. + async fn wait_exit(&mut self) -> Duration { + let mut child = self.child.take().expect("a running gateway"); + let started = Instant::now(); + loop { + if child.try_wait().expect("poll the gateway").is_some() { + return started.elapsed(); + } + if started.elapsed() > EXIT_LIMIT { + let _ = child.kill(); + let _ = child.wait(); + panic!( + "gateway {} still ran {EXIT_LIMIT:?} after SIGTERM:\n{}", + self.node_id, + tail(&self.log, TAIL_LINES) + ); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + } +} + +impl Drop for GatewayProcess { + fn drop(&mut self) { + if let Some(mut child) = self.child.take() { + graceful_kill(&mut child); + let _ = child.wait(); + } + } +} diff --git a/gateways/kafka/tests/common/docker_clients.rs b/gateways/kafka/tests/common/docker_clients.rs new file mode 100644 index 000000000..182de0358 --- /dev/null +++ b/gateways/kafka/tests/common/docker_clients.rs @@ -0,0 +1,265 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Kafka clients in Docker containers, which a test leaves running while it stops and starts +//! gateways and nodes. +//! +//! The containers use the host network, so a client reaches the gateways on the host's loopback. +//! Only Linux does that. They also run with `--pull never`, so a registry download never eats into +//! a test's budget: CI pulls the images before the tests start. +#![allow(dead_code)] + +use std::fs::File; +use std::io::{BufRead, BufReader, Read, Write}; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex, PoisonError}; +use std::thread::JoinHandle; +use std::time::Duration; + +use tokio::time::Instant; + +/// Numbers the containers of one test process, so their names differ. +static NEXT_CONTAINER: AtomicUsize = AtomicUsize::new(0); + +/// Why the clients cannot run on this host, or `None` when they can. +pub fn missing_prerequisite(images: &[&str]) -> Option<String> { + if !cfg!(target_os = "linux") { + return Some("`docker run --network host` reaches the host loopback only on Linux".into()); + } + if !succeeds(Command::new("docker").arg("info")) { + return Some("docker is unavailable".into()); + } + if let Some(image) = images + .iter() + .find(|image| !succeeds(Command::new("docker").args(["image", "inspect", image]))) + { + return Some(format!( + "client image {image} is not pulled; run `docker pull {image}`" + )); + } + if !succeeds(Command::new("timeout").args(["--kill-after=1s", "1s", "true"])) { + return Some("GNU `timeout` with `--kill-after` is unavailable".into()); + } + None +} + +/// Reports why a test skips. Fails it instead when `KAFKA_E2E_REQUIRED=1`, as CI sets it, so a +/// broken prerequisite cannot pass a test over zero assertions. +pub fn skip(reason: &str) { + assert!( + std::env::var("KAFKA_E2E_REQUIRED").as_deref() != Ok("1"), + "KAFKA_E2E_REQUIRED=1 but the suite cannot run: {reason}" + ); + eprintln!("skipping: {reason}"); +} + +fn succeeds(command: &mut Command) -> bool { + command + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()) +} + +/// A client container that runs until it exits or the test drops it. It keeps every line the +/// client prints, and writes them to log files as well. +pub struct Container { + name: String, + process: Child, + stdout: Arc<Mutex<Vec<String>>>, + stderr: Arc<Mutex<Vec<String>>>, + readers: Vec<JoinHandle<()>>, +} + +impl Container { + /// Starts `args` in `image`, with each `(host file, guest path)` of `mounts` read-only. + /// `timeout` inside the container ends it after `lifetime`, also when this process dies before + /// it can remove the container. Its output goes to `<log>.out` and `<log>.err` as well. + pub fn start( + image: &str, + args: &[&str], + mounts: &[(&Path, &str)], + lifetime: Duration, + log: &Path, + ) -> Self { + let name = format!( + "iggy-kafka-failover-{}-{}", + std::process::id(), + NEXT_CONTAINER.fetch_add(1, Ordering::Relaxed) + ); + let mut command = Command::new("docker"); + command.args([ + "run", + "--rm", + "--pull", + "never", + "--network", + "host", + "--name", + &name, + ]); + for (host, guest) in mounts { + command + .arg("-v") + .arg(format!("{}:{guest}:ro", host.display())); + } + let mut process = command + .arg(image) + .args([ + "timeout", + "--kill-after=10s", + &format!("{}s", lifetime.as_secs()), + ]) + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("start a client container"); + let (stdout, stdout_reader) = collect( + process.stdout.take().expect("piped stdout"), + log.with_extension("out"), + ); + let (stderr, stderr_reader) = collect( + process.stderr.take().expect("piped stderr"), + log.with_extension("err"), + ); + Self { + name, + process, + stdout, + stderr, + readers: vec![stdout_reader, stderr_reader], + } + } + + /// Calls `read` with every line that the client printed to stdout so far. + pub fn with_stdout<T>(&self, read: impl FnOnce(&[String]) -> T) -> T { + read(&self.stdout.lock().unwrap_or_else(PoisonError::into_inner)) + } + + /// The lines on stdout so far that contain `needle`. + pub fn count(&self, needle: &str) -> usize { + self.with_stdout(|lines| lines.iter().filter(|line| line.contains(needle)).count()) + } + + /// The last `lines` lines that the client printed to stderr. + pub fn stderr_tail(&self, lines: usize) -> String { + let stderr = self.stderr.lock().unwrap_or_else(PoisonError::into_inner); + stderr[stderr.len().saturating_sub(lines)..].join("\n") + } + + /// Sends `signal` to the client, as `docker kill --signal` does. `timeout`, which runs as the + /// container's first process, passes it on to the client. + pub fn signal(&self, signal: &str) { + let sent = Command::new("docker") + .args(["kill", "--signal", signal, &self.name]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status() + .is_ok_and(|status| status.success()); + assert!(sent, "send {signal} to client container {}", self.name); + } + + /// Waits until the client exits, at most `limit`, and returns whether it did. Every line it + /// printed is read by then. + pub async fn wait_exit(&mut self, limit: Duration) -> bool { + let deadline = Instant::now() + limit; + while self + .process + .try_wait() + .expect("poll the client container") + .is_none() + { + if Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + for reader in self.readers.drain(..) { + let _ = reader.join(); + } + true + } +} + +impl Drop for Container { + fn drop(&mut self) { + // After a clean exit `--rm` already removed it, and this fails harmlessly. + let _ = Command::new("docker") + .args(["rm", "--force", &self.name]) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .status(); + let _ = self.process.kill(); + let _ = self.process.wait(); + } +} + +/// Keeps each line of `pipe`, and writes it to `log`, on a thread of its own until the pipe +/// closes. +fn collect( + pipe: impl Read + Send + 'static, + log: PathBuf, +) -> (Arc<Mutex<Vec<String>>>, JoinHandle<()>) { + let lines = Arc::new(Mutex::new(Vec::new())); + let kept = Arc::clone(&lines); + let reader = std::thread::spawn(move || { + let mut file = File::create(&log).ok(); + for line in BufReader::new(pipe).lines().map_while(Result::ok) { + if let Some(file) = file.as_mut() { + let _ = writeln!(file, "{line}"); + } + kept.lock() + .unwrap_or_else(PoisonError::into_inner) + .push(line); + } + }); + (lines, reader) +} + +/// What a client that ran to its end printed. +pub struct Finished { + pub succeeded: bool, + pub stdout: String, + pub stderr: String, +} + +/// Runs `args` in `image` until the client exits, at most `limit`. Blocks the calling thread. +pub fn run_to_end(image: &str, args: &[String], limit: Duration) -> Finished { + let output = Command::new("timeout") + .args(["--kill-after=10s", &format!("{}s", limit.as_secs())]) + .args([ + "docker", + "run", + "--rm", + "--pull", + "never", + "--network", + "host", + image, + ]) + .args(args) + .output() + .expect("run a client container"); + Finished { + succeeded: output.status.success(), + stdout: String::from_utf8_lossy(&output.stdout).into_owned(), + stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + } +} diff --git a/gateways/kafka/tests/common/iggy_server.rs b/gateways/kafka/tests/common/iggy_server.rs index 1c9ff1c88..b8c75403e 100644 --- a/gateways/kafka/tests/common/iggy_server.rs +++ b/gateways/kafka/tests/common/iggy_server.rs @@ -512,12 +512,28 @@ struct ClusterNode { pub struct TestCluster { data_dir: PathBuf, nodes: Vec<ClusterNode>, + /// Server settings of every node, as `IGGY_*` variables. + settings: Vec<(String, String)>, + /// Where node `n` appends its output, as `node-<n>.log`. Without it, a node writes to the + /// test's own output. + log_dir: Option<PathBuf>, } impl TestCluster { /// Starts `count` nodes and waits until each accepts connections. The cluster can still be /// without a metadata primary then, so a first bridge can need a retry. pub async fn spawn(data_dir: &std::path::Path, count: usize) -> Self { + Self::spawn_with(data_dir, count, &[], None).await + } + + /// [`Self::spawn`] with server settings of its own on every node, as `IGGY_*` variables, and + /// each node's output appended to `node-<n>.log` in `log_dir`. + pub async fn spawn_with( + data_dir: &std::path::Path, + count: usize, + settings: &[(&str, &str)], + log_dir: Option<&std::path::Path>, + ) -> Self { let nodes = (0..count) .map(|_| ClusterNode { child: None, @@ -528,6 +544,11 @@ impl TestCluster { let mut cluster = Self { data_dir: data_dir.to_path_buf(), nodes, + settings: settings + .iter() + .map(|(key, value)| ((*key).to_string(), (*value).to_string())) + .collect(), + log_dir: log_dir.map(std::path::Path::to_path_buf), }; for node in 0..count { cluster.start(node).await; @@ -535,6 +556,21 @@ impl TestCluster { cluster } + pub const fn node_count(&self) -> usize { + self.nodes.len() + } + + pub fn is_running(&self, node: usize) -> bool { + self.nodes[node].child.is_some() + } + + /// The log file of node `node`, when the nodes write to files. + pub fn log_file(&self, node: usize) -> Option<PathBuf> { + self.log_dir + .as_ref() + .map(|dir| dir.join(format!("node-{node}.log"))) + } + /// Starts node `node` on its data dir and ports, and waits until it accepts connections. pub async fn start(&mut self, node: usize) { let mut command = Command::new(iggy_server_binary()); @@ -583,6 +619,21 @@ impl TestCluster { peer_node.replica.port.to_string(), ); } + command.envs( + self.settings + .iter() + .map(|(key, value)| (key.as_str(), value.as_str())), + ); + if let Some(log_file) = self.log_file(node) { + let log = OpenOptions::new() + .create(true) + .append(true) + .open(&log_file) + .expect("open the node log"); + command + .stdout(log.try_clone().expect("share the node log")) + .stderr(log); + } let mut child = command.spawn().expect("spawn an iggy-server node"); wait_for_listener(&mut child, &address, &format!("iggy-server node {node}")).await; self.nodes[node].child = Some(child); @@ -596,6 +647,15 @@ impl TestCluster { } } + /// Kills node `node` with SIGKILL, skipping its shutdown path the way a crash would. Its data + /// dir stays. + pub fn crash(&mut self, node: usize) { + if let Some(mut child) = self.nodes[node].child.take() { + let _ = child.kill(); + let _ = child.wait(); + } + } + pub fn address(&self, node: usize) -> String { format!("127.0.0.1:{}", self.nodes[node].tcp.port) }
