diff --git a/linux/Cargo.lock b/linux/Cargo.lock index 87474f3f95..5ba98a4706 100644 --- a/linux/Cargo.lock +++ b/linux/Cargo.lock @@ -2056,16 +2056,13 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ - "base64 0.22.1", "bytes", "futures-channel", "futures-util", "http", "http-body", "hyper", - "ipnet", "libc", - "percent-encoding", "pin-project-lite", "socket2", "tokio", @@ -2321,12 +2318,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "ipnet" -version = "2.12.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" - [[package]] name = "itertools" version = "0.13.0" @@ -3467,41 +3458,6 @@ dependencies = [ "syn 2.0.119", ] -[[package]] -name = "reqwest" -version = "0.13.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "16a1cfa75cc186dd73d5818e510e042e40927bccc9c236b061cea97e1eb08029" -dependencies = [ - "base64 0.23.1", - "bytes", - "encoding_rs", - "futures-core", - "h2", - "http", - "http-body", - "http-body-util", - "hyper", - "hyper-rustls", - "hyper-util", - "js-sys", - "log", - "mime", - "percent-encoding", - "pin-project-lite", - "serde", - "serde_json", - "sync_wrapper", - "tokio", - "tower", - "tower-http", - "tower-service", - "url", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "ring" version = "0.17.14" @@ -4379,9 +4335,6 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" -dependencies = [ - "futures-core", -] [[package]] name = "synstructure" @@ -4768,7 +4721,6 @@ dependencies = [ "memchr", "parse-display", "pin-project-lite", - "reqwest", "serde", "serde_json", "serde_with", @@ -5146,24 +5098,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "tower-http" -version = "0.6.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" -dependencies = [ - "bitflags 2.13.2", - "bytes", - "futures-util", - "http", - "http-body", - "pin-project-lite", - "tower", - "tower-layer", - "tower-service", - "url", -] - [[package]] name = "tower-layer" version = "0.3.3" @@ -5493,16 +5427,6 @@ dependencies = [ "wasm-bindgen-shared", ] -[[package]] -name = "wasm-bindgen-futures" -version = "0.4.78" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6ef4c5d3d2cdf5c54f4231181768f5510842e350db025faf1f7163b1030ed928" -dependencies = [ - "js-sys", - "wasm-bindgen", -] - [[package]] name = "wasm-bindgen-macro" version = "0.2.128" diff --git a/linux/crates/test-fixtures/Cargo.toml b/linux/crates/test-fixtures/Cargo.toml index 93964e79a2..b24add33b5 100644 --- a/linux/crates/test-fixtures/Cargo.toml +++ b/linux/crates/test-fixtures/Cargo.toml @@ -12,7 +12,7 @@ path = "src/lib.rs" [dependencies] rcgen.workspace = true tempfile.workspace = true -testcontainers = { workspace = true, features = ["http_wait_plain"] } +testcontainers.workspace = true testcontainers-modules = { workspace = true, features = ["postgres", "mssql_server"] } thiserror.workspace = true tokio = { workspace = true, features = ["fs", "io-util", "process"] } diff --git a/linux/crates/test-fixtures/src/clickhouse_fixture.rs b/linux/crates/test-fixtures/src/clickhouse_fixture.rs index c512dc9485..9e42f5445b 100644 --- a/linux/crates/test-fixtures/src/clickhouse_fixture.rs +++ b/linux/crates/test-fixtures/src/clickhouse_fixture.rs @@ -1,9 +1,13 @@ use std::time::Duration; -use testcontainers::core::wait::HttpWaitStrategy; -use testcontainers::core::{IntoContainerPort, WaitFor}; +use testcontainers::bollard::query_parameters::InspectContainerOptions; +use testcontainers::core::IntoContainerPort; +use testcontainers::core::client::docker_client_instance; use testcontainers::runners::AsyncRunner; -use testcontainers::{ContainerAsync, GenericImage, ImageExt}; +use testcontainers::{ContainerAsync, ContainerRequest, GenericImage, ImageExt, TestcontainersError}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpStream; +use tokio::time::Instant; use crate::container_file::with_files; use crate::{ContainerFile, ExecOutput, FixtureCredentials, FixtureError, FixtureImage, PkiFile, TestNetwork, TestPki}; @@ -11,10 +15,13 @@ use crate::{ContainerFile, ExecOutput, FixtureCredentials, FixtureError, Fixture const HTTP_PORT: u16 = ClickHouseFixture::HTTP_PORT; const HTTPS_PORT: u16 = ClickHouseFixture::HTTPS_PORT; const PASSWORD_USER: &str = "tablepro"; -/// The image starts the server once to create CLICKHOUSE_DB, stops it and -/// starts it again before /ping answers; on a busy CI runner that has -/// taken longer than testcontainers' default of one minute. -const STARTUP_TIMEOUT: Duration = Duration::from_secs(180); +const DATABASE: &str = "tablepro"; +const SERVICE: &str = "clickhouse"; +const READY_TIMEOUT: Duration = Duration::from_secs(60); +// A connection that is accepted and never answered must not use up the deadline. +const READY_ATTEMPT: Duration = Duration::from_secs(5); +const READY_RETRY: Duration = Duration::from_millis(250); +const LOG_TAIL_LINES: usize = 40; pub struct ClickHouseFixture { container: ContainerAsync, @@ -33,35 +40,42 @@ impl ClickHouseFixture { pub const HTTPS_PORT: u16 = 8443; pub async fn start() -> Result { - Self::launch(None).await + Self::launch(None, Vec::new(), READY_TIMEOUT).await } pub async fn start_in(network: &TestNetwork) -> Result { - Self::launch(Some(network)).await + Self::launch(Some(network), Vec::new(), READY_TIMEOUT).await } - async fn launch(network: Option<&TestNetwork>) -> Result { + // No CLICKHOUSE_DB: the image would start, stop and restart the server + // to create it. One HTTP CREATE DATABASE is both that and the readiness check. + async fn launch( + network: Option<&TestNetwork>, + extra_files: Vec, + ready_timeout: Duration, + ) -> Result { let pki = TestPki::generate()?; let password = FixtureCredentials::generate_password(); - let ready = HttpWaitStrategy::new("/ping") - .with_port(HTTP_PORT.tcp()) - .with_expected_status_code(200_u16); - let mut request = FixtureImage::CLICKHOUSE + let mut request: ContainerRequest = FixtureImage::CLICKHOUSE .generic() .with_exposed_port(HTTP_PORT.tcp()) .with_exposed_port(HTTPS_PORT.tcp()) - .with_wait_for(WaitFor::http(ready)) - .with_startup_timeout(STARTUP_TIMEOUT) - .with_env_var("CLICKHOUSE_DB", "tablepro"); + .into(); let network_alias = network.map(|network| network.container_name(Self::NETWORK_ROLE)); if let (Some(network), Some(alias)) = (network, &network_alias) { request = request.with_network(network.name()).with_container_name(alias); } - let container = with_files(request, Self::files(&pki, &password)?).start().await?; - let host = container.get_host().await?.to_string(); - let http_port = container.get_host_port_ipv4(HTTP_PORT).await?; - let https_port = container.get_host_port_ipv4(HTTPS_PORT).await?; - Ok(Self { + let mut files = Self::files(&pki, &password)?; + files.extend(extra_files); + let container = with_files(request, files).start().await?; + let (host, http_port, https_port) = match published(&container).await { + Ok(published) => published, + Err(error) => { + let reason = format!("its ports are not published: {error}"); + return Err(not_ready(&container, reason, "none".to_owned()).await); + } + }; + let fixture = Self { container, pki, password, @@ -69,7 +83,9 @@ impl ClickHouseFixture { http_port, https_port, network_alias, - }) + }; + fixture.create_database(ready_timeout).await?; + Ok(fixture) } pub fn files(pki: &TestPki, password: &str) -> Result, FixtureError> { @@ -101,8 +117,14 @@ impl ClickHouseFixture { \n\ \n" ); + // Otherwise the server logs only to files, and a failed start leaves no log to report. + let console = "\n\ + 1\n\ + information\n\ + \n"; Ok(vec![ ContainerFile::readable("/etc/clickhouse-server/config.d/tls.xml", server), + ContainerFile::readable("/etc/clickhouse-server/config.d/console.xml", console), ContainerFile::readable("/etc/clickhouse-server/users.d/tablepro.xml", users), ContainerFile::readable(format!("{tls}/ca.pem"), pki.read(PkiFile::Ca)?), ContainerFile::readable(format!("{tls}/server.pem"), pki.read(PkiFile::ServerCertificate)?), @@ -145,4 +167,250 @@ impl ClickHouseFixture { pub async fn exec(&self, command: &[&str]) -> Result { ExecOutput::run(&self.container, command).await } + + async fn create_database(&self, ready_timeout: Duration) -> Result<(), FixtureError> { + let request = query_request( + &self.host, + &self.password_credentials(), + &format!("CREATE DATABASE IF NOT EXISTS {DATABASE}"), + ); + let deadline = Instant::now() + ready_timeout; + loop { + let attempt = exchange(&self.host, self.http_port, &request); + let last_reply = match tokio::time::timeout(READY_ATTEMPT, attempt).await { + Ok(Ok(reply)) if status_code(&reply) == Some(200) => return Ok(()), + Ok(Ok(reply)) => reply_summary(&reply), + Ok(Err(error)) => error.to_string(), + Err(_) => format!("no reply within {} seconds", READY_ATTEMPT.as_secs()), + }; + if !self.container.is_running().await? { + return Err(not_ready(&self.container, "the container exited".to_owned(), last_reply).await); + } + if Instant::now() >= deadline { + let reason = format!("no answer within {} seconds", ready_timeout.as_secs()); + return Err(not_ready(&self.container, reason, last_reply).await); + } + tokio::time::sleep(READY_RETRY).await; + } + } +} + +async fn published(container: &ContainerAsync) -> Result<(String, u16, u16), TestcontainersError> { + let host = container.get_host().await?.to_string(); + let http_port = container.get_host_port_ipv4(HTTP_PORT).await?; + let https_port = container.get_host_port_ipv4(HTTPS_PORT).await?; + Ok((host, http_port, https_port)) +} + +async fn not_ready(container: &ContainerAsync, reason: String, last_reply: String) -> FixtureError { + FixtureError::ContainerNotReady { + service: SERVICE, + reason, + last_reply, + state: state(container).await, + logs: log_tail(container).await, + } +} + +async fn state(container: &ContainerAsync) -> String { + let inspected = match docker_client_instance().await { + Ok(docker) => docker + .inspect_container(container.id(), None::) + .await + .map_err(|error| error.to_string()), + Err(error) => Err(error.to_string()), + }; + match inspected.map(|response| response.state) { + Ok(Some(state)) => format!( + "status {}, exit code {}, OOM killed {}, error \"{}\"", + state.status.map(|status| status.to_string()).unwrap_or_default(), + state.exit_code.map(|code| code.to_string()).unwrap_or_default(), + state.oom_killed.unwrap_or_default(), + state.error.unwrap_or_default(), + ), + Ok(None) => "no state reported".to_owned(), + Err(error) => format!("unavailable: {error}"), + } +} + +async fn log_tail(container: &ContainerAsync) -> String { + let stdout = container.stdout_to_vec().await; + let stderr = container.stderr_to_vec().await; + format!( + "stdout:\n{}\nstderr:\n{}", + stream_tail(stdout.as_deref().map_err(ToString::to_string)), + stream_tail(stderr.as_deref().map_err(ToString::to_string)), + ) +} + +fn query_request(host: &str, credentials: &FixtureCredentials, query: &str) -> String { + format!( + "POST / HTTP/1.0\r\n\ + Host: {host}\r\n\ + X-ClickHouse-User: {}\r\n\ + X-ClickHouse-Key: {}\r\n\ + Content-Length: {}\r\n\ + \r\n\ + {query}", + credentials.username, + credentials.password, + query.len(), + ) +} + +async fn exchange(host: &str, port: u16, request: &str) -> std::io::Result { + let mut stream = TcpStream::connect((host, port)).await?; + stream.write_all(request.as_bytes()).await?; + let mut reply = Vec::new(); + stream.read_to_end(&mut reply).await?; + Ok(String::from_utf8_lossy(&reply).into_owned()) +} + +fn status_code(reply: &str) -> Option { + let status_line = reply.lines().next()?; + let mut parts = status_line.split_whitespace(); + parts.next().filter(|version| version.starts_with("HTTP/"))?; + parts.next()?.parse().ok() +} + +fn reply_summary(reply: &str) -> String { + let status = reply.lines().next().unwrap_or_default(); + let body = reply + .split_once("\r\n\r\n") + .and_then(|(_, body)| body.lines().find(|line| !line.trim().is_empty())) + .unwrap_or_default(); + match (status.is_empty(), body.is_empty()) { + (true, _) => "an empty reply".to_owned(), + (false, true) => status.to_owned(), + (false, false) => format!("{status}: {body}"), + } +} + +fn stream_tail(output: Result<&[u8], String>) -> String { + match output { + Ok(bytes) => { + let text = String::from_utf8_lossy(bytes); + let lines: Vec<&str> = text.lines().collect(); + lines[lines.len().saturating_sub(LOG_TAIL_LINES)..].join("\n") + } + Err(error) => format!("unavailable: {error}"), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config_file(name: &str, body: &str) -> ContainerFile { + ContainerFile::readable( + format!("/etc/clickhouse-server/config.d/{name}"), + format!("{body}\n"), + ) + } + + #[test] + fn the_query_request_carries_the_credentials_and_the_body_length() { + let request = query_request("localhost", &FixtureCredentials::new("tablepro", "secret"), "SELECT 1"); + + assert!(request.starts_with("POST / HTTP/1.0\r\nHost: localhost\r\n")); + assert!(request.contains("\r\nX-ClickHouse-User: tablepro\r\nX-ClickHouse-Key: secret\r\n")); + assert!(request.ends_with("\r\nContent-Length: 8\r\n\r\nSELECT 1")); + } + + #[test] + fn only_an_http_status_line_yields_a_status() { + assert_eq!(status_code("HTTP/1.0 200 OK\r\n\r\n"), Some(200)); + assert_eq!(status_code("HTTP/1.1 516 Unknown\r\n\r\nCode: 516."), Some(516)); + assert_eq!(status_code("SSH-2.0-OpenSSH_10.3\r\n"), None); + assert_eq!(status_code(""), None); + } + + #[test] + fn a_refused_query_is_summarised_with_the_server_message() { + let reply = "HTTP/1.0 516 Unknown\r\nX-ClickHouse-Exception-Code: 516\r\n\r\n\ + Code: 516. DB::Exception: tablepro: Authentication failed\n"; + + assert_eq!( + reply_summary(reply), + "HTTP/1.0 516 Unknown: Code: 516. DB::Exception: tablepro: Authentication failed" + ); + assert_eq!( + reply_summary("HTTP/1.0 500 Internal Server Error\r\n\r\n"), + "HTTP/1.0 500 Internal Server Error" + ); + assert_eq!(reply_summary(""), "an empty reply"); + } + + #[test] + fn the_log_tail_keeps_the_last_lines() { + let log: String = (1..=50).map(|line| format!("line {line}\n")).collect(); + + let tail = stream_tail(Ok(log.as_bytes())); + + assert!(tail.starts_with("line 11\n"), "{tail}"); + assert!(tail.ends_with("\nline 50"), "{tail}"); + assert_eq!(stream_tail(Err("gone".to_owned())), "unavailable: gone"); + } + + #[test] + fn the_server_also_logs_to_the_console() { + let pki = TestPki::generate().unwrap(); + let files = ClickHouseFixture::files(&pki, "secret").unwrap(); + + let console = files + .iter() + .find(|file| file.target() == "/etc/clickhouse-server/config.d/console.xml") + .unwrap(); + + let text = String::from_utf8_lossy(console.contents()); + assert!(text.contains("1"), "{text}"); + assert!( + text.contains("information"), + "{text}" + ); + } + + #[tokio::test] + #[ignore = "requires docker"] + async fn a_server_that_exits_is_reported_at_once_with_its_state_and_logs() { + let broken = config_file("zz-broken.xml", "Nowhere/Invalid"); + let started = Instant::now(); + + let error = ClickHouseFixture::launch(None, vec![broken], READY_TIMEOUT) + .await + .err() + .unwrap(); + + assert!(started.elapsed() < READY_TIMEOUT / 2, "took {:?}", started.elapsed()); + let FixtureError::ContainerNotReady { + reason, state, logs, .. + } = &error + else { + panic!("unexpected error: {error}"); + }; + assert_eq!(reason, "the container exited"); + assert!(state.starts_with("status exited, exit code "), "{state}"); + assert!(logs.contains("Nowhere/Invalid"), "{logs}"); + } + + #[tokio::test] + #[ignore = "requires docker"] + async fn a_server_that_never_answers_is_reported_with_its_state_and_logs() { + let elsewhere = config_file("zz-elsewhere.xml", "8124"); + + let error = ClickHouseFixture::launch(None, vec![elsewhere], Duration::from_secs(8)) + .await + .err() + .unwrap(); + + let FixtureError::ContainerNotReady { + reason, state, logs, .. + } = &error + else { + panic!("unexpected error: {error}"); + }; + assert_eq!(reason, "no answer within 8 seconds"); + assert!(state.starts_with("status running, "), "{state}"); + assert!(logs.contains("Listening for http://0.0.0.0:8124"), "{logs}"); + } } diff --git a/linux/crates/test-fixtures/src/fixture_error.rs b/linux/crates/test-fixtures/src/fixture_error.rs index 413315bdeb..46cfa6b571 100644 --- a/linux/crates/test-fixtures/src/fixture_error.rs +++ b/linux/crates/test-fixtures/src/fixture_error.rs @@ -16,4 +16,12 @@ pub enum FixtureError { }, #[error("{service} did not accept connections within {seconds} seconds")] NotReady { service: &'static str, seconds: u64 }, + #[error("{service} was not ready: {reason}; last reply: {last_reply}; container {state}\n{logs}")] + ContainerNotReady { + service: &'static str, + reason: String, + last_reply: String, + state: String, + logs: String, + }, } diff --git a/linux/crates/test-fixtures/tests/fixtures.rs b/linux/crates/test-fixtures/tests/fixtures.rs index 614712689c..c809b011fb 100644 --- a/linux/crates/test-fixtures/tests/fixtures.rs +++ b/linux/crates/test-fixtures/tests/fixtures.rs @@ -289,6 +289,26 @@ async fn clickhouse_https_fixture_serves_the_generated_certificate() -> TestResu Ok(()) } +#[tokio::test] +#[ignore = "requires docker"] +async fn clickhouse_fixture_starts_the_server_once_with_the_database() -> TestResult { + let fixture = ClickHouseFixture::start().await?; + let log = "/var/log/clickhouse-server/clickhouse-server.log"; + let query = "SELECT name FROM system.databases WHERE name = 'tablepro'"; + + let starts = fixture.exec(&["grep", "-c", "Starting ClickHouse", log]).await?; + let databases = fixture.exec(&["clickhouse-client", "--query", query]).await?; + + assert_eq!(starts.stdout_text().trim(), "1", "{}", starts.stderr_text()); + assert_eq!( + databases.stdout_text().trim(), + "tablepro", + "{}", + databases.stderr_text() + ); + Ok(()) +} + #[tokio::test] #[ignore = "requires docker"] async fn pgbouncer_fixture_runs_select_1_through_the_upstream() -> TestResult {