diff --git a/.config/nextest.toml b/.config/nextest.toml index e56cf79de7..243a8f8d10 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -59,6 +59,7 @@ filter = ''' + binary_id(iggy-gateway-kafka::metadata_real_bridge_tests) + binary_id(iggy-gateway-kafka::offsets_real_bridge_tests) + binary_id(iggy-gateway-kafka::phase1_e2e_tests) + + binary_id(iggy-gateway-kafka::sasl_real_server_tests) ''' test-group = "kafka_bridge" diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md index 692aa6a2cb..1e2d4ea181 100644 --- a/gateways/kafka/README.md +++ b/gateways/kafka/README.md @@ -77,7 +77,7 @@ Default bind: a `PLAINTEXT` listener on `127.0.0.1:9093`. Environment variables: | `IGGY_KAFKA_WRITE_TIMEOUT_SECS` | `10` | Seconds allowed to write a response frame | | `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown waits for in-flight connections before abandoning them. The bridge then signs its Iggy clients out, 5 s at most | | `IGGY_KAFKA_INSTANCE_ID` | `0` | This gateway's number among the gateways fronting one Iggy cluster. It is the high half of every producer id `InitProducerId` hands out, and Kafka requires those to be cluster-unique, so give every gateway its own value. A single gateway can leave it at `0`. | -| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. A failed connection is fatal, not a downgrade to stubs. | +| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. If Iggy does not answer, startup tries again 1 s after each failed connect, for up to 60 s, and then the gateway stops. A wrong password or a missing permission stops it at once. It never falls back to stubs. | | `IGGY_KAFKA_SASL_ENABLED` | `false` | Single-listener form: require SASL/PLAIN before serving any other API (`true` or `false`, nothing else). With `IGGY_KAFKA_LISTENERS`, a `SASL_PLAINTEXT` or `SASL_SSL` listener says the same | | `IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS` | `15` | Seconds an unauthenticated connection may sit between frames. Waiting for an authentication slot and the verification itself each get this budget, the verification's starting once it holds a slot. Separate from the 10-minute idle timeout that applies once authenticated | | `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` | `4` | Credential verifications the gateway runs at once, across all connections. Each costs a password hash on an Iggy shard thread. This bounds the gateway's side only: a check that times out frees its slot while its hash keeps running inside Iggy. Size it below the Iggy node's shard count | @@ -316,6 +316,10 @@ IGGY_KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9093,SSL://0.0.0.0:9094,SASL_SSL://0. Each security protocol can be bound once, since the protocol is also the listener's name. +No `accept()` error stops a listener. An error of one pending connection, such as a reset, retries +at once. Any other error, such as no free file descriptor, waits 10 ms before the next accept. +The wait doubles up to 1 s while the error continues. + TLS is rustls. One certificate serves every `SSL` and `SASL_SSL` listener, and the files are re-read every `IGGY_KAFKA_TLS_RELOAD_INTERVAL_SECS`: a rotated certificate takes effect for new connections without a restart, a key that does not match leaves the previous material serving and @@ -422,8 +426,10 @@ Four things to know before switching it on: - **Enabling it breaks every existing client at once.** The two SASL keys appear in the `ApiVersions` advertisement only while it is on, and unauthenticated clients are refused. - **Every connection costs a login**, meaning one password hash on an Iggy shard thread and one - replicated registration. Verification is deliberately not cached, since caching it per username - would let a second connection present any password. Connection churn is therefore server load. + replicated registration. The check then signs out, so its Iggy session ends at once. Without + the sign out, Iggy keeps the session for 36 s, and a full session table refuses new logins. + Verification is deliberately not cached, since caching it per username would let a second + connection present any password. Connection churn is therefore server load. `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` bounds the checks the gateway runs at once, and a peer whose login was rejected is refused for a delay that doubles per rejection, from 0.5s up to 30s. - **Authentication and an ACL view, not enforcement.** The gateway verifies the credentials and @@ -596,7 +602,9 @@ The initial connect retries a fixed, bounded number of times (`RECONNECTION_RETR Iggy SDK client's own default of unlimited retries, one dial per second, forever), and the whole attempt - retries included - is capped at `REQUEST_TIMEOUT` (15s) wall-clock, so `IggyBridge::connect` fails in bounded time whether the address refuses the connection or silently drops it, instead of -blocking the calling task indefinitely. Every other bridge call (`ensure_stream_and_topic`, +blocking the calling task indefinitely. At startup, the gateway retries a connect that a retry can +fix (6 or 7). It waits 1 s between tries, for up to 60 s. The cluster id load after it does the same. +So a gateway that starts before its Iggy node waits for it. Every other bridge call (`ensure_stream_and_topic`, `high_watermark(s)`, `probe`, `close`) carries the same `REQUEST_TIMEOUT` for the same reason: the SDK reconnects internally, mid-call, on a transport error, through the same undead-lined dial path - a bridge call made well after the initial connect can still hit this if diff --git a/gateways/kafka/docs/AUTHENTICATION.md b/gateways/kafka/docs/AUTHENTICATION.md index 8e9045a517..a5bfee2437 100644 --- a/gateways/kafka/docs/AUTHENTICATION.md +++ b/gateways/kafka/docs/AUTHENTICATION.md @@ -25,7 +25,7 @@ verified on every connection, against Iggy, and that verification is deliberatel **Authorization is meant to stay in Iggy.** The gateway will not implement its own ACL model. The intent is to carry the authenticated Iggy identity onto the data plane and let the server's existing permission checks -decide what it can do. That is later work: today the verified client is shut down as soon as the credential +decide what it can do. That is later work: today the verified client signs out and shuts down as soon as the credential clears, and nothing carries the principal past authentication. **A client certificate is an identity too, with the same reach.** On an `SSL` listener that verifies client @@ -63,12 +63,12 @@ connection. The cost rules it out. | Argon2id verify, run inline on the shard thread | 19 MiB, t=2, p=1 | | Replicated VSR `Register` | one consensus commit | | Leader settlement round trip, plus a second login if it redirects | one or two round trips | -| Session slots, cluster-wide across all transports | 8192, oldest committed evicted past it | +| Session slots, cluster-wide across all transports | 8192, and a full table refuses new logins | 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 (`core/consensus/src/impls.rs:178`) is cluster-wide, so -connection churn on the Kafka side turns directly into session-table pressure and eviction on the Iggy side. +nothing to do with authentication. The 8192 ceiling (`[metadata] clients_table_max`, `core/consensus/src/impls.rs:211`) 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 @@ -103,7 +103,7 @@ Two consequences worth naming rather than discovering: traffic on that shard. Measured on debug builds, a single login takes about 10 ms and 32 concurrent logins take about 75 ms each, with throughput flattening near 410 per second. The shape is the point, latency climbing while throughput plateaus, not the absolute numbers. `MANUAL_TESTING.md` has the table. -- **Nothing holds the verified session.** The connection that proved a credential drops it at once. +- **Nothing holds the verified session.** The connection that proved a credential signs out at once. Produce and Fetch act as the bridge's Iggy user for every client. A data client per principal needs pooling, a separate question from how credentials are verified. Pooling is gateway-side code over the public SDK and needs no SDK change. diff --git a/gateways/kafka/docs/TEST_SUITE.md b/gateways/kafka/docs/TEST_SUITE.md index d0ca7de545..fe9c028885 100644 --- a/gateways/kafka/docs/TEST_SUITE.md +++ b/gateways/kafka/docs/TEST_SUITE.md @@ -20,7 +20,7 @@ suite goes through `tests/common/fixtures.rs::load_fixture_body_or_skip`, which regeneration hint when a fixture is missing, and panics instead when `KAFKA_FIXTURES_REQUIRED=1` is set (CI sets this) so a broken generation step can't leave a suite green with zero assertions. -### `iggy-server` binary (required for `bridge_iggy_integration_tests` and every `*_real_bridge_tests` suite) +### `iggy-server` binary (required for `bridge_iggy_integration_tests`, `sasl_real_server_tests` and every `*_real_bridge_tests` suite) No fixtures needed, but `iggy-server` has to be built *first* - these suites spawn it directly and do not build it for you: @@ -80,6 +80,7 @@ file under `tests/` anymore. | [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No | | [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, the handshake budget, certificate reload without a restart, and the startup refusals | No | | [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. `kafka-cluster cluster-id` prints the same id across a gateway restart. Lookups by time resolve through `kafka-get-offsets` (a time, 0, `-3` and a future time), `kafka-consumer-groups --reset-offsets --to-datetime`, and the consumer's `offsetsForTimes` in a probe compiled with the host `javac`. librdkafka and the Java client connect with `security.protocol=SSL` and `SASL_SSL`, and a client certificate admits each as the Iggy user it names. With auto topic creation on, the Java producer sends to a topic nobody created. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker, a built `iggy-server` and a host JDK (`javac`) | +| [`sasl_real_server_tests.rs`](../tests/sasl_real_server_tests.rs) | `IggyAuthenticator` against a real, spawned `iggy-server`: twice as many logins as a small session table holds all succeed, because each check signs out | No (needs the `iggy-server` binary - see Prerequisites) | | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) | | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) | | [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) | @@ -88,7 +89,7 @@ file under `tests/` anymore. | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) | | [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16), DescribeGroups (key 15), DeleteGroups (key 42) and OffsetDelete (key 47) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and only the live groups once a scan failed with Iggy down. DeleteGroups refuses a group with members, and after they leave deletes it on every topic. Through a relay that holds an Iggy call past its deadline, a commit and a group deletion each keep the other out until the call ends, and past the SDK's 30 s deadline until a replay proves what the call did. A delete that an earlier delete in doubt kept from Iggy lets go of its group. A deletion that ends after its request answered 14 answers 0 to the retry. OffsetDelete keeps the offsets of a topic that a member subscribes to, deletes the others, and finds a group that another gateway committed for since the last scan | No (needs the `iggy-server` binary - see Prerequisites) | | [`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://github.com/apache/iggy/issues/4519)) | No (needs both binaries - see Prerequisites) | -| [`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 | No (needs both binaries - see Prerequisites) | +| [`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 and then serves | No (needs both binaries - see Prerequisites) | `tests/common/` holds shared helpers (`certs.rs`, `codec.rs`, `fixtures.rs`, `gateway_process.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` diff --git a/gateways/kafka/src/auth.rs b/gateways/kafka/src/auth.rs index 42e5a740d2..7c0098a305 100644 --- a/gateways/kafka/src/auth.rs +++ b/gateways/kafka/src/auth.rs @@ -29,7 +29,7 @@ use std::time::{Duration, Instant}; use async_trait::async_trait; use iggy::prelude::{ AutoLogin, Client, Credentials, Identifier, IggyError, Permissions, TcpClient, - TcpClientConfigBuilder, UserStatus, + TcpClientConfigBuilder, UserClient, UserStatus, }; use tokio::sync::SemaphorePermit; use tracing::{debug, warn}; @@ -64,12 +64,12 @@ const VERIFY_TIMEOUT: Duration = Duration::from_secs(10); /// would only hide the failure underneath one the client cannot see. const VERIFY_RECONNECTION_RETRIES: u32 = 0; -/// Budget for tearing the verification client down again. +/// Budget for signing the verification client out and tearing it down again. /// /// Deliberately far shorter than [`VERIFY_TIMEOUT`]. The authentication slot is already released by /// the time teardown runs, but every second here still comes out of the caller's own -/// pre-authentication budget. Nothing is lost by cutting it short: dropping the client closes the -/// socket regardless. +/// pre-authentication budget. Little is lost by cutting it short: dropping the client closes the +/// socket regardless, and Iggy ends a session that did not sign out when its lease runs out. const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1); /// Delay imposed on a peer after its first rejected login, doubled on every further rejection. @@ -391,6 +391,7 @@ impl SaslAuthenticator for IggyAuthenticator { // The password verify the slot bounds is done. Holding it across the permission read and // teardown would keep a slot busy for two more round trips per login. drop(slot); + let signed_in = matches!(connected, Ok(Ok(()))); let outcome = match connected { Err(_elapsed) => Err(AuthError::Unavailable), Ok(Err(error)) => Err(classify(&error)), @@ -402,20 +403,28 @@ impl SaslAuthenticator for IggyAuthenticator { Ok(Ok(())) => fetch_permissions(&client, &credentials.username).await, }; - // Shut down rather than drop, so the socket closes cleanly and Iggy releases the session - // now instead of noticing the dead peer later. + // Sign out first: a shutdown only closes the socket, and Iggy keeps the session in its + // clients table until the lease runs out, 36 s by default. A full table refuses every new + // login, so a burst of Kafka logins would lock out the cluster's other clients. Then shut + // down rather than drop, so the socket closes cleanly. // // Both failure shapes are reported, and separately: the outer `Elapsed` and the inner // `IggyError` mean different things, and folding them together would log "timed out" for a // shutdown that failed immediately. Neither leaks the client, since dropping it closes the // socket regardless, but an invisible failure here would hide a real one. - match tokio::time::timeout(TEARDOWN_TIMEOUT, client.shutdown()).await { + let teardown = async { + if signed_in && let Err(error) = client.logout_user().await { + debug!(%error, "a credential-verification client did not sign out"); + } + client.shutdown().await + }; + match tokio::time::timeout(TEARDOWN_TIMEOUT, teardown).await { Ok(Ok(())) => {} Ok(Err(error)) => { warn!(%error, "failed to shut down a credential-verification client"); } Err(_elapsed) => { - warn!("timed out shutting down a credential-verification client"); + warn!("timed out signing out and shutting down a credential-verification client"); } } diff --git a/gateways/kafka/src/bridge/error.rs b/gateways/kafka/src/bridge/error.rs index f35bd68846..d348098aff 100644 --- a/gateways/kafka/src/bridge/error.rs +++ b/gateways/kafka/src/bridge/error.rs @@ -196,6 +196,19 @@ impl BridgeError { } } + /// True when the same call can succeed later: Kafka clients retry its code, 6 or 7. A wrong + /// password, a missing permission or a bad configuration is not. + #[must_use] + pub fn is_retriable(&self) -> bool { + match self { + Self::ClusterIdCall { source, .. } => source.is_retriable(), + _ => matches!( + self.to_kafka_error_code(), + ERROR_NOT_LEADER_OR_FOLLOWER | ERROR_REQUEST_TIMED_OUT + ), + } + } + /// True when Iggy rejected the bridge's own login. The Kafka client only sees -1. #[must_use] pub const fn is_bridge_login_rejected(&self) -> bool { @@ -658,4 +671,43 @@ mod tests { assert!(!BridgeError::Iggy(IggyError::Unauthorized).is_bridge_login_rejected()); assert!(!BridgeError::Timeout.is_bridge_login_rejected()); } + + #[test] + fn given_an_iggy_that_is_not_up_yet_should_be_retriable() { + for error in [ + BridgeError::Iggy(IggyError::CannotEstablishConnection), + BridgeError::Iggy(IggyError::Disconnected), + BridgeError::Timeout, + ] { + assert!(error.is_retriable(), "{error}"); + } + let call = BridgeError::ClusterIdCall { + call: ClusterIdCall::Poll, + stream: "kafka".to_string(), + topic: "topic", + source: Box::new(BridgeError::Iggy(IggyError::NotConnected)), + }; + assert!( + call.is_retriable(), + "a cluster id call answers for its cause" + ); + } + + #[test] + fn given_a_wrong_setting_should_not_be_retriable() { + for error in [ + BridgeError::Iggy(IggyError::InvalidCredentials), + BridgeError::Iggy(IggyError::Unauthorized), + BridgeError::InvalidConfig("empty address".to_string()), + ] { + assert!(!error.is_retriable(), "{error}"); + } + let call = BridgeError::ClusterIdCall { + call: ClusterIdCall::CreateStream, + stream: "kafka".to_string(), + topic: "topic", + source: Box::new(BridgeError::Iggy(IggyError::Unauthorized)), + }; + assert!(!call.is_retriable()); + } } diff --git a/gateways/kafka/src/main.rs b/gateways/kafka/src/main.rs index 9c1f824024..6be8c6ce58 100644 --- a/gateways/kafka/src/main.rs +++ b/gateways/kafka/src/main.rs @@ -22,11 +22,13 @@ use std::time::Duration; use tokio::signal; use tokio::sync::{Semaphore, broadcast}; +use tokio::time::Instant; use tracing::{info, warn}; use iggy_gateway_kafka::auth::IggyAuthenticator; use iggy_gateway_kafka::bridge::{ - DEFAULT_TOPIC_PARTITIONS, IggyBridge, IggyBridgeConfig, IggyTls, MAX_TOPIC_PARTITIONS, + BridgeError, DEFAULT_TOPIC_PARTITIONS, IggyBridge, IggyBridgeConfig, IggyTls, + MAX_TOPIC_PARTITIONS, }; use iggy_gateway_kafka::env::parse_bool; use iggy_gateway_kafka::listener::{ @@ -50,10 +52,17 @@ async fn main() -> Result<(), Box> { .listeners .iter() .any(|listener| listener.protocol == SecurityProtocol::SaslPlaintext); - let bridge = connect_bridge(sasl_enabled).await?; + let give_up = Instant::now() + IGGY_STARTUP_LIMIT; + let bridge = connect_bridge(sasl_enabled, give_up).await?; // Loaded before the listeners bind, so no client waits in the backlog behind Iggy. let cluster_id = match &bridge { - Some(bridge) => Some(bridge.load_cluster_id().await.map_err(|e| e.to_string())?), + Some(bridge) => Some( + retry_until("load the Kafka cluster id", give_up, || { + bridge.load_cluster_id() + }) + .await + .map_err(|e| e.to_string())?, + ), None => None, }; @@ -100,6 +109,37 @@ async fn main() -> Result<(), Box> { Ok(outcome??) } +/// How long startup waits for Iggy. A gateway can start before its Iggy node, and one connect of +/// the bridge gives up within 15 s. +const IGGY_STARTUP_LIMIT: Duration = Duration::from_secs(60); + +/// The pause between two startup attempts to reach Iggy. +const IGGY_STARTUP_RETRY_INTERVAL: Duration = Duration::from_secs(1); + +/// Runs `attempt` until it succeeds. An error that a retry can fix starts a new attempt, until +/// `give_up`. +async fn retry_until( + what: &str, + give_up: Instant, + mut attempt: impl FnMut() -> Attempt, +) -> Result +where + Attempt: Future>, +{ + loop { + match attempt().await { + Err(error) + if error.is_retriable() + && Instant::now() + IGGY_STARTUP_RETRY_INTERVAL < give_up => + { + warn!("cannot {what} yet, retrying in {IGGY_STARTUP_RETRY_INTERVAL:?}: {error}"); + tokio::time::sleep(IGGY_STARTUP_RETRY_INTERVAL).await; + } + outcome => return outcome, + } + } +} + /// How long shutdown waits for the Iggy clients to sign out. The connection drain before it can /// take 25 s of a 30 s grace period. const BRIDGE_CLOSE_TIMEOUT: Duration = Duration::from_secs(5); @@ -200,10 +240,13 @@ fn reject_unknown_kafka_env_vars() -> Result<(), String> { /// Connects the Iggy bridge when `IGGY_KAFKA_BRIDGE_ENABLED` is true, otherwise returns `None` /// and every handler keeps answering with its stub. /// -/// Off by default, so wiring one API at a time does not change what an operator sees. A failed -/// connect is fatal: an operator who turned the bridge on must learn it is unreachable at -/// startup, not from `NOT_LEADER_OR_FOLLOWER` on every produce. -async fn connect_bridge(sasl_enabled: bool) -> Result>, String> { +/// Off by default, so wiring one API at a time does not change what an operator sees. A connect +/// that still fails at `give_up` is fatal: an operator who turned the bridge on must learn it is +/// unreachable at startup, not from `NOT_LEADER_OR_FOLLOWER` on every produce. +async fn connect_bridge( + sasl_enabled: bool, + give_up: Instant, +) -> Result>, String> { if !bridge_enabled()? { let set: Vec<&str> = IggyBridgeConfig::KNOWN_ENV_VARS .iter() @@ -234,9 +277,11 @@ async fn connect_bridge(sasl_enabled: bool) -> Result>, S in the clear; set IGGY_KAFKA_IGGY_TLS_ENABLED=true to encrypt it" ); } - let bridge = IggyBridge::connect(config) - .await - .map_err(|e| format!("failed to connect the Iggy bridge: {e}"))?; + let bridge = retry_until("connect the Iggy bridge", give_up, || { + IggyBridge::connect(config.clone()) + }) + .await + .map_err(|e| format!("failed to connect the Iggy bridge: {e}"))?; info!("Iggy bridge connected"); Ok(Some(Arc::new(bridge))) } diff --git a/gateways/kafka/src/server.rs b/gateways/kafka/src/server.rs index 5cec88e3c1..ba0f429f5e 100644 --- a/gateways/kafka/src/server.rs +++ b/gateways/kafka/src/server.rs @@ -440,9 +440,10 @@ impl KafkaGateway { /// /// # Errors /// - /// Returns an error on invalid config, unusable TLS material, or a non-transient `accept()` - /// error on any listener. With a bridge and no [`Self::with_cluster_id`], also returns the - /// bridge's error when it cannot load the cluster id from Iggy. + /// Returns an error on invalid config, unusable TLS material, or a panic in the accept loop of + /// a listener. An `accept()` error ends nothing: the loop retries it. With a bridge and no + /// [`Self::with_cluster_id`], also returns the bridge's error when it cannot load the cluster + /// id from Iggy. pub async fn run( self, listeners: Vec, @@ -542,10 +543,10 @@ impl KafkaGateway { } Ok(()) } - // An accept loop ends early only on a non-transient accept error, since shutdown is - // what the other branch is for. + // An accept loop ends early only by a panic, since shutdown is what the other branch + // is for. Some(finished) = accept_loops.join_next() => match finished { - Ok(result) => result, + Ok(()) => Ok(()), Err(join_error) => Err(io::Error::other(join_error).into()), }, }; @@ -684,18 +685,22 @@ struct AcceptLoop { tls: Option>, } -/// Accepts on one listener until the cancel token fires or `accept()` fails for good. -async fn accept_loop(this: AcceptLoop) -> Result<()> { +/// Accepts on one listener until the cancel token fires. No `accept()` error ends it, because the +/// listener cannot come back without a restart of the gateway, which ends every connection and +/// all group state. +async fn accept_loop(this: AcceptLoop) { + let mut backoff = AcceptBackoff::default(); loop { let accepted = tokio::select! { - () = this.cancel.cancelled() => return Ok(()), + () = this.cancel.cancelled() => return, accepted = this.socket.accept() => accepted, }; if this.cancel.is_cancelled() { - return Ok(()); + return; } match accepted { Ok((stream, peer)) => { + backoff.reset(); let Ok(permit) = Arc::clone(&this.conn_limiter).try_acquire_owned() else { warn!(%peer, protocol = %this.listener.protocol, max_connections = this.config.max_connections, "listener connection limit reached, rejecting"); continue; @@ -720,18 +725,54 @@ async fn accept_loop(this: AcceptLoop) -> Result<()> { serve_connection(stream, connection).await; }); } - Err(e) if is_transient_accept_error(&e) => { - // Brief backoff on fd exhaustion to avoid busy-spinning. - if matches!(e.raw_os_error(), Some(23 | 24)) { - tokio::time::sleep(Duration::from_millis(10)).await; + Err(e) => { + let Some(delay) = backoff.after(&e) else { + debug!(%e, protocol = %this.listener.protocol, "accept failed for one connection, continuing"); + continue; + }; + warn!(%e, protocol = %this.listener.protocol, ?delay, "accept failed, retrying"); + tokio::select! { + () = this.cancel.cancelled() => return, + () = tokio::time::sleep(delay) => {} } - warn!(%e, "transient accept error, continuing"); } - Err(e) => return Err(e.into()), } } } +/// The first wait of [`AcceptBackoff`]. +const ACCEPT_BACKOFF_MIN: Duration = Duration::from_millis(10); +/// The longest wait of [`AcceptBackoff`], so the listener takes connections again soon after the +/// error ends. +const ACCEPT_BACKOFF_MAX: Duration = Duration::from_secs(1); + +/// The wait before the next `accept()` after an error. +/// +/// An error of one pending connection retries at once. Any other error, such as no free file +/// descriptor, waits, and the wait doubles while the errors continue. So an error that lasts +/// neither spins a core nor floods the log. +#[derive(Debug, Default)] +struct AcceptBackoff { + /// `None` until an error waits, and again after an accept succeeds. + next: Option, +} + +impl AcceptBackoff { + /// How long to wait after `error`. `None` to accept again at once. + fn after(&mut self, error: &io::Error) -> Option { + if is_connection_accept_error(error) { + return None; + } + let delay = self.next.unwrap_or(ACCEPT_BACKOFF_MIN); + self.next = Some((delay * 2).min(ACCEPT_BACKOFF_MAX)); + Some(delay) + } + + const fn reset(&mut self) { + self.next = None; + } +} + /// What an accepted socket carries into its own task. struct Connection { peer: SocketAddr, @@ -823,14 +864,23 @@ async fn drain(tracker: &TaskTracker, cancel: &CancellationToken, deadline: Dura } } -fn is_transient_accept_error(err: &std::io::Error) -> bool { +/// Whether an `accept()` error belongs to one pending connection, not to the listener. Linux +/// `accept(2)` returns the network errors of a pending connection, and its manual says to retry +/// them like `EAGAIN`. +fn is_connection_accept_error(error: &io::Error) -> bool { matches!( - err.kind(), - io::ErrorKind::Interrupted | io::ErrorKind::ConnectionAborted | io::ErrorKind::WouldBlock + error.kind(), + io::ErrorKind::Interrupted + | io::ErrorKind::WouldBlock + | io::ErrorKind::ConnectionAborted + | io::ErrorKind::ConnectionReset + | io::ErrorKind::ConnectionRefused + | io::ErrorKind::NetworkDown + | io::ErrorKind::NetworkUnreachable + | io::ErrorKind::HostUnreachable ) || matches!( - err.raw_os_error(), - // EMFILE / ENFILE are common across Unix platforms when fd limits are hit. - Some(23 | 24) + error.raw_os_error(), + Some(libc::EPROTO | libc::ENOPROTOOPT | libc::EHOSTDOWN | libc::EOPNOTSUPP) ) } @@ -1790,24 +1840,41 @@ mod tests { } #[test] - fn transient_accept_error_classification_covers_all_branches() { - for kind in [ - io::ErrorKind::Interrupted, - io::ErrorKind::ConnectionAborted, - io::ErrorKind::WouldBlock, + fn given_an_error_of_one_connection_when_accepting_should_retry_at_once() { + let mut backoff = AcceptBackoff::default(); + for error in [ + io::Error::from(io::ErrorKind::Interrupted), + io::Error::from(io::ErrorKind::ConnectionAborted), + io::Error::from(io::ErrorKind::ConnectionReset), + io::Error::from_raw_os_error(libc::ENETUNREACH), + io::Error::from_raw_os_error(libc::EHOSTUNREACH), + io::Error::from_raw_os_error(libc::EPROTO), + io::Error::from_raw_os_error(libc::EHOSTDOWN), ] { - assert!(is_transient_accept_error(&io::Error::from(kind))); + assert_eq!(backoff.after(&error), None, "{error}"); } + } - #[cfg(unix)] - { - assert!(is_transient_accept_error(&io::Error::from_raw_os_error(23))); - assert!(is_transient_accept_error(&io::Error::from_raw_os_error(24))); + #[test] + fn given_errors_of_the_listener_when_accepting_should_retry_after_a_growing_wait() { + let mut backoff = AcceptBackoff::default(); + let waits: Vec<_> = [libc::EMFILE, libc::ENFILE, libc::ENOBUFS, libc::ENOMEM] + .into_iter() + .map(|code| backoff.after(&io::Error::from_raw_os_error(code))) + .collect(); + assert_eq!( + waits, + [10, 20, 40, 80].map(|millis| Some(Duration::from_millis(millis))) + ); + + let unknown = io::Error::other("an error no list names"); + for _ in 0..10 { + backoff.after(&unknown); } + assert_eq!(backoff.after(&unknown), Some(ACCEPT_BACKOFF_MAX)); - assert!(!is_transient_accept_error(&io::Error::from( - io::ErrorKind::ConnectionRefused, - ))); + backoff.reset(); + assert_eq!(backoff.after(&unknown), Some(ACCEPT_BACKOFF_MIN)); } #[tokio::test] diff --git a/gateways/kafka/tests/common/gateway_process.rs b/gateways/kafka/tests/common/gateway_process.rs index b7e3f70e6f..d29c0ce156 100644 --- a/gateways/kafka/tests/common/gateway_process.rs +++ b/gateways/kafka/tests/common/gateway_process.rs @@ -26,6 +26,7 @@ use std::path::PathBuf; use std::process::{Child, Command}; use crate::iggy_server::{PortGuard, TestServer, graceful_kill, wait_for_listener}; +use iggy_gateway_kafka::bridge::IggyBridgeConfig; use std::env; /// Locates the already-built `iggy-gateway-kafka` binary. @@ -50,9 +51,16 @@ impl TestGateway { /// Spawns `iggy-gateway-kafka` with the bridge enabled against `server`, then blocks until /// its listener is ready or the startup budget is exhausted. pub async fn spawn(server: &TestServer) -> Self { + let mut gateway = Self::start(&server.test_config()); + gateway.wait_ready().await; + gateway + } + + /// Starts `iggy-gateway-kafka` against the Iggy server that `config` names, which need not run + /// yet. [`Self::wait_ready`] waits for its listener. + pub fn start(config: &IggyBridgeConfig) -> Self { let port_guard = PortGuard::acquire(); let address = format!("127.0.0.1:{}", port_guard.port); - let config = server.test_config(); let mut command = Command::new(iggy_gateway_binary()); // env_clear(): without it the child inherits this test process's full environment, and @@ -76,20 +84,26 @@ impl TestGateway { ); let child = command.spawn().expect("spawn iggy-gateway-kafka"); - let mut gateway = Self { + Self { child, address, _port_guard: port_guard, - }; - gateway.wait_ready().await; - gateway + } + } + + /// Whether the process still runs. + pub fn is_running(&mut self) -> bool { + self.child + .try_wait() + .expect("poll iggy-gateway-kafka") + .is_none() } /// Bare TCP connect poll, same rationale as `TestServer::wait_ready`: a full Kafka client /// handshake would pay retry overhead on every failed attempt. A connect slightly ahead of /// the gateway's own accept loop can't happen here - `main.rs` connects the bridge before /// binding the listener, so a successful connect already implies the bridge is up too. - async fn wait_ready(&mut self) { + pub async fn wait_ready(&mut self) { wait_for_listener(&mut self.child, &self.address, "iggy-gateway-kafka").await; } } diff --git a/gateways/kafka/tests/common/iggy_server.rs b/gateways/kafka/tests/common/iggy_server.rs index 1ce8a45f41..be4013d977 100644 --- a/gateways/kafka/tests/common/iggy_server.rs +++ b/gateways/kafka/tests/common/iggy_server.rs @@ -262,12 +262,12 @@ impl TestServer { /// its assertions, orphaning a process that still holds its `PortGuard` slot, which then /// fails the next test that draws that slot to bind. pub async fn spawn_with_password(data_dir: &std::path::Path, password: &str) -> Self { - Self::spawn_configured(data_dir, password, None, &[]).await + Self::spawn_configured(data_dir, password, None, &[], PortGuard::acquire()).await } /// [`Self::spawn`] with server settings of its own, as `IGGY_*` variables. pub async fn spawn_with_env(data_dir: &std::path::Path, env: &[(&str, &str)]) -> Self { - Self::spawn_configured(data_dir, "iggy", None, env).await + Self::spawn_configured(data_dir, "iggy", None, env, PortGuard::acquire()).await } /// Spawns `iggy-server` whose TCP listener speaks only TLS, with the given PEM pair. @@ -279,7 +279,19 @@ impl TestServer { cert_file: &std::path::Path, key_file: &std::path::Path, ) -> Self { - Self::spawn_configured(data_dir, "iggy", Some((cert_file, key_file)), &[]).await + Self::spawn_configured( + data_dir, + "iggy", + Some((cert_file, key_file)), + &[], + PortGuard::acquire(), + ) + .await + } + + /// [`Self::spawn`] on the port that [`Self::stop`] kept, as a node that comes back. + pub async fn spawn_on(data_dir: &std::path::Path, port_guard: PortGuard) -> Self { + Self::spawn_configured(data_dir, "iggy", None, &[], port_guard).await } async fn spawn_configured( @@ -287,8 +299,8 @@ impl TestServer { password: &str, tls: Option<(&std::path::Path, &std::path::Path)>, env: &[(&str, &str)], + port_guard: PortGuard, ) -> Self { - let port_guard = PortGuard::acquire(); let address = format!("127.0.0.1:{}", port_guard.port); let mut command = Command::new(iggy_server_binary()); diff --git a/gateways/kafka/tests/phase1_e2e_tests.rs b/gateways/kafka/tests/phase1_e2e_tests.rs index a03aa5f642..73c5e0f35c 100644 --- a/gateways/kafka/tests/phase1_e2e_tests.rs +++ b/gateways/kafka/tests/phase1_e2e_tests.rs @@ -36,6 +36,7 @@ //! follow-up rather than bundled into this PR's scope. use std::net::SocketAddr; +use std::time::Duration; use bytes::{Bytes, BytesMut}; use iggy::prelude::{Consumer, Identifier, MessageClient, PollingStrategy}; @@ -339,3 +340,33 @@ async fn phase1_produce_flow_through_real_gateway_process_and_real_iggy_server() assert_eq!(polled[0].payload.as_ref(), b"phase1-hello"); assert_eq!(polled[1].payload.as_ref(), b"phase1-world"); } + +/// A gateway started before its Iggy node waits for it, instead of exiting. +#[tokio::test] +async fn given_iggy_that_starts_late_when_the_gateway_starts_should_wait_for_it() { + // Past the longest connect of the bridge, 15 s, after which the gateway used to exit. + const IGGY_DELAY: Duration = Duration::from_secs(16); + + let data_dir = tempfile::tempdir().expect("tempdir"); + let server = TestServer::spawn(data_dir.path()).await; + let config = server.test_config(); + let port = server.stop(); + + let mut gateway = TestGateway::start(&config); + tokio::time::sleep(IGGY_DELAY).await; + assert!(gateway.is_running(), "the gateway waits for Iggy"); + let _server = TestServer::spawn_on(data_dir.path(), port).await; + gateway.wait_ready().await; + + let addr: SocketAddr = gateway.address.parse().expect("gateway address"); + let (_, body) = round_trip( + addr, + API_KEY_CREATE_TOPICS, + CREATE_TOPICS_VERSION, + 1, + &build_create_topics_request(TOPIC, 1), + ) + .await; + let (error_code, _, _) = decode_create_topics_response(body); + assert_eq!(error_code, ERROR_NONE, "the bridge reaches Iggy"); +} diff --git a/gateways/kafka/tests/sasl_real_server_tests.rs b/gateways/kafka/tests/sasl_real_server_tests.rs new file mode 100644 index 0000000000..86aac107e1 --- /dev/null +++ b/gateways/kafka/tests/sasl_real_server_tests.rs @@ -0,0 +1,67 @@ +// 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. + +//! The SASL credential check against a real, spawned `iggy-server`. + +use secrecy::SecretString; +use serial_test::serial; +use tokio::sync::Semaphore; + +use iggy_gateway_kafka::auth::{IggyAuthenticator, SaslAuthenticator}; +use iggy_gateway_kafka::protocol::sasl::PlainCredentials; + +#[path = "common/iggy_server.rs"] +mod iggy_server; + +use iggy_server::TestServer; + +/// Iggy keeps a session that did not sign out until its lease runs out, 36 s later, and a full +/// clients table refuses every new login. +#[tokio::test] +#[serial] +async fn given_more_logins_than_the_clients_table_holds_should_admit_each() { + const CLIENTS_TABLE_MAX: usize = 4; + + let data_dir = tempfile::tempdir().expect("tempdir"); + let table_max = CLIENTS_TABLE_MAX.to_string(); + // Iggy frees the slot of a session that signed out at its next liveness pass, one per + // heartbeat interval, 5 s by default. A login waits for a slot only 10 s. + let server = TestServer::spawn_with_env( + data_dir.path(), + &[ + ("IGGY_METADATA_CLIENTS_TABLE_MAX", &table_max), + ("IGGY_CONSUMER_GROUP_HEARTBEAT_INTERVAL", "1s"), + ], + ) + .await; + let authenticator = IggyAuthenticator::new(server.test_config().address); + let credentials = PlainCredentials { + username: "iggy".to_string(), + password: SecretString::from("iggy"), + }; + let slots = Semaphore::new(1); + + for login in 0..2 * CLIENTS_TABLE_MAX { + let principal = authenticator + .authenticate( + &credentials, + slots.acquire().await.expect("an open semaphore"), + ) + .await; + assert!(principal.is_ok(), "login {login}: {principal:?}"); + } +}