ryerraguntla commented on code in PR #4043: URL: https://github.com/apache/iggy/pull/4043#discussion_r4020517195
########## gateways/kafka/tests/bridge_iggy_integration_tests.rs: ########## @@ -0,0 +1,918 @@ +// 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. + +//! Integration tests for `IggyBridge` against a real `iggy-server` process - not the +//! `KafkaGateway` under test elsewhere in this suite. `#3533` acceptance criteria this file +//! exercises directly: `ensure_stream_and_topic` idempotent on repeated calls, and the bridge +//! module invoked from a real (non-unit) test rather than only compiled. + +use std::collections::HashMap; +use std::fs::{File, OpenOptions}; +use std::path::PathBuf; +use std::process::{Child, Command}; +use std::sync::OnceLock; +use std::time::Duration; + +use iggy::prelude::{ + AutoLogin, Client, Credentials, Identifier, IggyClient, IggyClientBuilder, IggyMessage, + MessageClient, Partitioning, StreamClient, TopicClient, +}; +use secrecy::SecretString; +use serial_test::serial; + +use iggy_gateway_kafka::bridge::{ + BridgeError, IggyBridge, IggyBridgeConfig, TopicMapping, TopicOverride, +}; + +/// Desired slot count - the actual count `port_band()` computes may be smaller if the machine's +/// real ephemeral range leaves little room, but this crate never needs more than a handful live +/// at once (this binary's own tests already run one at a time - see `.config/nextest.toml`'s +/// `kafka_bridge` test-group). +const DESIRED_SLOTS: u16 = 200; + +/// Lowest port `port_band()` will ever place its band at - clear of the well-known/privileged +/// range (0-1023). +const MIN_CANDIDATE_PORT: u16 = 10000; + +/// Ephemeral range assumed when the kernel's own can't be read (no `/proc` - e.g. macOS), +/// matching the Linux default. Mirrors `core/integration`'s own `port_reserver.rs`. +const DEFAULT_EPHEMERAL_RANGE: (u16, u16) = (32768, 60999); + +const IP_LOCAL_PORT_RANGE: &str = "/proc/sys/net/ipv4/ip_local_port_range"; + +fn parse_ephemeral_range(range: &str) -> Option<(u16, u16)> { + let mut bounds = range.split_whitespace(); + let floor: u16 = bounds.next()?.parse().ok()?; + let ceiling: u16 = bounds.next()?.parse().ok()?; + Some((floor, ceiling)) +} + +/// The range the kernel picks `bind(0)` ports from, read once. +fn ephemeral_range() -> (u16, u16) { + static RANGE: OnceLock<(u16, u16)> = OnceLock::new(); + *RANGE.get_or_init(|| { + std::fs::read_to_string(IP_LOCAL_PORT_RANGE) + .ok() + .as_deref() + .and_then(parse_ephemeral_range) + .unwrap_or(DEFAULT_EPHEMERAL_RANGE) + }) +} + +/// First port and slot count of a band clear of the kernel's ephemeral range: below the floor by +/// preference (matches `core/integration`'s own `port_reserver.rs`), above the ceiling when the +/// floor leaves no room there. +/// +/// A *hardcoded* band, chosen once without reading the real range (this file's own prior version: +/// `15000..15199`, on the unverified assumption that every real deployment's floor sits above +/// it), is wrong on any box tuned wider - `net.ipv4.ip_local_port_range = "1024 65535"` swallows +/// that whole band. `flock` alone doesn't save it either: it is advisory, so it only excludes +/// another `PortGuard`-based process, never an unrelated `bind(0)` elsewhere on the box landing on +/// the same number from the kernel's own ephemeral pool. +fn port_band() -> (u16, u16) { + static BAND: OnceLock<(u16, u16)> = OnceLock::new(); + *BAND.get_or_init(|| { + let (floor, ceiling) = ephemeral_range(); + band_for(floor, ceiling) + }) +} + +/// Pure band-selection logic, split out from [`port_band`] so it's testable against synthetic +/// ranges without needing to fake `/proc` contents. +fn band_for(floor: u16, ceiling: u16) -> (u16, u16) { + if floor > MIN_CANDIDATE_PORT { + let room = floor - MIN_CANDIDATE_PORT; + return (MIN_CANDIDATE_PORT, room.min(DESIRED_SLOTS)); + } + let Some(start) = ceiling.checked_add(1) else { + panic!( + "kernel ephemeral range [{floor}, {ceiling}] leaves no room for a test port band \ + below {MIN_CANDIDATE_PORT} nor above {ceiling} - narrow the range, e.g. \ + sysctl -w net.ipv4.ip_local_port_range='32768 60999'" + ); + }; + // start >= 1 (ceiling + 1), so this never overflows u16. + let room = u16::MAX - start + 1; + (start, room.min(DESIRED_SLOTS)) +} + +#[test] +fn given_a_readable_range_when_parsed_should_take_both_bounds() { + assert_eq!( + parse_ephemeral_range("32768\t60999\n"), + Some((32768, 60999)) + ); + assert_eq!(parse_ephemeral_range("1024 65535"), Some((1024, 65535))); + assert_eq!(parse_ephemeral_range(""), None); + assert_eq!(parse_ephemeral_range("32768"), None); + assert_eq!(parse_ephemeral_range("garbage 60999"), None); +} + +#[test] +fn given_room_below_the_floor_when_choosing_a_band_should_take_it() { + let (start, slots) = band_for(32768, 60999); + assert_eq!(start, MIN_CANDIDATE_PORT); + assert!( + start + slots <= 32768, + "band [{start}, {}] runs into the ephemeral floor", + start + slots - 1 + ); +} + +/// A box tuned `net.ipv4.ip_local_port_range = "1024 60999"` is exactly the case the prior +/// hardcoded `15000..15199` band silently broke under: room below the floor down to +/// `MIN_CANDIDATE_PORT` is gone, but the ceiling still leaves room above it. +#[test] +fn given_no_room_below_the_floor_when_choosing_a_band_should_fall_back_above_the_ceiling() { + let (start, slots) = band_for(1024, 60999); + assert!(slots > 0, "must find room above the ceiling, not just fail"); + assert!( + start > 60999, + "band must start above the ephemeral ceiling, got {start}" + ); +} + +#[test] +#[should_panic(expected = "leaves no room")] +fn given_no_room_on_either_side_when_choosing_a_band_should_fail_loudly() { + band_for(1024, u16::MAX); +} + +/// Exclusive claim on one port, released (and the port freed for reuse) when dropped - including +/// on an unclean process exit, since the OS drops the `flock` with the file descriptor. Unlike +/// bind-then-drop, this process itself never binds the port before `iggy-server` does; see +/// `port_band`'s doc comment for the limits of that guarantee against *other* processes. +struct PortGuard { + port: u16, + _lock: File, +} + +impl PortGuard { + fn acquire() -> Self { + let (band_start, band_slots) = port_band(); + let lock_dir = std::env::temp_dir().join("iggy-kafka-gateway-test-port-locks"); + std::fs::create_dir_all(&lock_dir).expect("create port lock dir"); + for offset in 0..band_slots { + let port = band_start + offset; + let path = lock_dir.join(format!("{port}.lock")); + let Ok(file) = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + else { + continue; + }; + if file.try_lock().is_ok() { + return Self { port, _lock: file }; + } + } + panic!( + "no free port slot in [{band_start}, {}]", + band_start + band_slots - 1 + ); + } +} + +/// Locates the already-built `iggy-server` binary. Does not build it - see this crate's +/// `docs/TEST_SUITE.md` for the prerequisite, the same one `core/integration`'s own +/// server-spawning tests already carry. +/// +/// `assert_cmd::cargo::cargo_bin`, matching `core/integration`'s own +/// `harness::handle::server::start` - not a from-scratch `cargo build` invocation. An earlier +/// version of this function drove `cargo build --package server --bin iggy-server` directly, +/// reasoning that `Command::cargo_bin` "only resolves `CARGO_BIN_EXE_*` for binaries owned by +/// *this* package" - true of the `CARGO_BIN_EXE_*` env-var lookup alone, but `cargo_bin` falls +/// back to `legacy_cargo_bin` when that's unset, which derives the target directory from +/// `env::current_exe()` (this test binary's own path) and looks for a same-named file there - +/// correct under `CARGO_TARGET_DIR`, `--release`, or a `--target` triple subdirectory precisely +/// because it reads back from where cargo actually placed *this* binary, not a guessed path. That +/// version also re-ran a `cargo build` from every one of this file's server-spawning tests +/// (nextest runs each as its own process, so the `OnceLock` memoized nothing across them), +/// serialized by `.config/nextest.toml`'s `kafka_bridge` test-group but still real, avoidable +/// per-test overhead this version has none of. +fn iggy_server_binary() -> PathBuf { + assert_cmd::cargo::cargo_bin("iggy-server") +} + +struct TestServer { + child: Child, + address: String, + password: String, + _port_guard: PortGuard, +} + +impl TestServer { + /// Spawns `iggy-server` with the default `iggy`/`iggy` root credentials. See + /// [`Self::spawn_with_password`] for the general form. + async fn spawn(data_dir: &std::path::Path) -> Self { + Self::spawn_with_password(data_dir, "iggy").await + } + + /// Spawns `iggy-server` with an isolated temp data dir, a locked TCP port, and the given root + /// password, then blocks until its listener is ready or the startup budget is exhausted. + /// + /// A dedicated password parameter (not just the `spawn()` default everywhere) lets a + /// password-shaped regression test (special characters, say) reuse this harness's + /// `PortGuard`/graceful-`Drop`/`wait_ready` machinery instead of hand-rolling a second, + /// `Drop`-less spawn: a `Drop`-less copy has no guard to run `graceful_kill` on a panic before + /// its assertions, orphaning a process that still holds its `PortGuard` slot, which then + /// fails the next test that draws that slot to bind. + async fn spawn_with_password(data_dir: &std::path::Path, password: &str) -> Self { + let port_guard = PortGuard::acquire(); + let address = format!("127.0.0.1:{}", port_guard.port); + + let mut command = Command::new(iggy_server_binary()); + command + .env("IGGY_PATH", data_dir.display().to_string()) + .env("IGGY_TCP_ADDRESS", &address) + .env("IGGY_HTTP_ENABLED", "false") + .env("IGGY_QUIC_ENABLED", "false") + // WebSocket defaults to enabled on a fixed 127.0.0.1:8092 (config.toml), unlike TCP + // which reads a per-test port from PortGuard - every spawned server here would fight + // over that one port otherwise, and a bind failure aborts boot. + .env("IGGY_WEBSOCKET_ENABLED", "false") + // Matches core/integration's own spawned-server harness (harness/handle/server.rs): + // pinned shards (config.toml default: pin_cores = true) of concurrently running + // servers pile onto the same cores and starve each other. This crate's own tests are + // serialized against each other (.config/nextest.toml's kafka_bridge test-group - + // #[serial] alone does not survive nextest, which runs each test as its own process), + // but a spawned server here can still land alongside a pinned server from a different + // package's test in the same nextest run. + .env("IGGY_SHARDING_PIN_CORES", "false") Review Comment: Added IGGY_SHARDING_CPU_ALLOCATION=0..4. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
