diff --git a/core/connectors/sinks/elasticsearch_sink/README.md b/core/connectors/sinks/elasticsearch_sink/README.md index 905fbf496b..0c9c1038f4 100644 --- a/core/connectors/sinks/elasticsearch_sink/README.md +++ b/core/connectors/sinks/elasticsearch_sink/README.md @@ -8,7 +8,7 @@ A sink connector that consumes messages from Iggy streams and indexes them to El - `index`: Target index name - `username/password`: Optional authentication credentials - `batch_size`: Bulk indexing batch size (default: 100) -- `timeout_seconds`: Request timeout (default: 30s) +- `timeout_seconds`: Client-wide HTTP timeout for `open()` and bulk `consume()` (default: 30s). Values of `0` are clamped to 1s. Raise for slow bulk workloads; a timed-out bulk fails the batch after the poll offset is already committed - `create_index_if_not_exists`: Automatically create index (default: true) - `index_mapping`: Index mapping configuration diff --git a/core/connectors/sinks/elasticsearch_sink/src/lib.rs b/core/connectors/sinks/elasticsearch_sink/src/lib.rs index d4a8cb99d3..372fd2b8f6 100644 --- a/core/connectors/sinks/elasticsearch_sink/src/lib.rs +++ b/core/connectors/sinks/elasticsearch_sink/src/lib.rs @@ -31,11 +31,14 @@ use secrecy::{ExposeSecret, SecretString}; use serde::{Deserialize, Serialize}; use serde_json::json; use simd_json::{OwnedValue, prelude::*}; +use std::time::Duration; use tokio::sync::Mutex; use tracing::{info, warn}; sink_connector!(ElasticsearchSink); +const DEFAULT_TIMEOUT_SECONDS: u64 = 30; + #[derive(Debug)] struct State { invocations_count: usize, @@ -51,6 +54,11 @@ pub struct ElasticsearchSinkConfig { #[serde(serialize_with = "iggy_common::serde_secret::serialize_optional_secret")] pub password: Option, pub batch_size: Option, + /// Client-wide HTTP timeout for open() and bulk consume(). Values of `0` + /// are clamped to 1s (a zero duration would fail every request immediately). + /// Raise this for slow bulk workloads; until runtime ack/retry (#2927/#2928), + /// a timeout on consume drops the batch after the poll offset is already + /// committed. pub timeout_seconds: Option, pub create_index_if_not_exists: Option, pub index_mapping: Option, @@ -83,7 +91,17 @@ impl ElasticsearchSink { .map_err(|error| Error::Connection(format!("Invalid Elasticsearch URL: {error}")))?; let conn_pool = elasticsearch::http::transport::SingleNodeConnectionPool::new(url); - let mut transport_builder = TransportBuilder::new(conn_pool); + // elasticsearch-rs defaults to no timeout. This client-global timeout is + // an infinite-hang backstop for open() and for bulk consume() — not the + // primary #3728 flake fix (that is the harness readiness gate, which + // expires sooner than the 30s default). Values of 0 clamp to 1s. + let timeout_seconds = self + .config + .timeout_seconds + .unwrap_or(DEFAULT_TIMEOUT_SECONDS) + .max(1); + let mut transport_builder = + TransportBuilder::new(conn_pool).timeout(Duration::from_secs(timeout_seconds)); if let (Some(username), Some(password)) = (&self.config.username, &self.config.password) { let credentials = diff --git a/core/integration/src/harness/handle/connectors_runtime.rs b/core/integration/src/harness/handle/connectors_runtime.rs index f86eb91436..1118f55432 100644 --- a/core/integration/src/harness/handle/connectors_runtime.rs +++ b/core/integration/src/harness/handle/connectors_runtime.rs @@ -243,19 +243,41 @@ impl IggyServerDependent for ConnectorsRuntimeHandle { } async fn wait_ready(&mut self) -> Result<(), TestBinaryError> { - let http_address = self.http_url(); - let client = reqwest::Client::new(); + // Prefer /health over `/` so readiness means the connectors API is up, + // not some other listener that happened to bind the reserved port. + let health_url = format!("{}/health", self.http_url()); + // Bound each probe so a black-holed TCP accept (no HTTP response) cannot + // stall send() past the pid-crash / retry budget. Localhost dead process + // usually ECONNREFUSED, but a short timeout keeps both guards effective. + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to build connectors health client: {error}"), + })?; for retry in 0..common::DEFAULT_HEALTH_CHECK_RETRIES { - match client.get(&http_address).send().await { - Ok(_) => { + if let Some(pid) = self.pid() + && !common::is_process_alive(pid) + { + let (stdout, stderr) = self.collect_logs(); + return Err(TestBinaryError::ProcessCrashed { + binary: "iggy-connectors".to_string(), + exit_code: None, + stdout, + stderr, + }); + } + + match client.get(&health_url).send().await { + Ok(response) if response.status().is_success() => { return Ok(()); } - Err(_) => { + Ok(_) | Err(_) => { if retry == common::DEFAULT_HEALTH_CHECK_RETRIES - 1 { return Err(TestBinaryError::HealthCheckFailed { binary: "iggy-connectors".to_string(), - address: http_address, + address: health_url, retries: common::DEFAULT_HEALTH_CHECK_RETRIES, }); } diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs index 5749b18096..a0ea1a13ea 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/container.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/container.rs @@ -20,13 +20,16 @@ use reqwest_middleware::ClientWithMiddleware as HttpClient; use reqwest_retry::RetryTransientMiddleware; use reqwest_retry::policies::ExponentialBackoff; use serde::Deserialize; +use std::fs::{File, OpenOptions, TryLockError}; +use std::process::Stdio; +use std::time::{Duration, Instant}; use testcontainers_modules::testcontainers::core::wait::HttpWaitStrategy; use testcontainers_modules::testcontainers::core::{IntoContainerPort, WaitFor}; use testcontainers_modules::testcontainers::runners::AsyncRunner; use testcontainers_modules::testcontainers::{ ContainerAsync, GenericImage, ImageExt, ReuseDirective, }; -use tracing::info; +use tracing::{info, warn}; const ELASTICSEARCH_IMAGE: &str = "docker.io/library/elasticsearch"; const ELASTICSEARCH_TAG: &str = "9.3.0"; @@ -37,6 +40,25 @@ const ELASTICSEARCH_HEALTH_ENDPOINT: &str = "/_cluster/health"; // name. Per-test isolation comes from a unique index per fixture, not a fresh // container. const ELASTICSEARCH_CONTAINER_NAME: &str = "iggy-test-elasticsearch"; +// Short probe timeouts: create_http_client() uses 30s + retries and must not +// be used for readiness. One hung attempt there looks like a 60s+ test hang. +const CLUSTER_READY_ATTEMPTS: usize = 40; +const CLUSTER_READY_INTERVAL_MS: u64 = 250; +const CLUSTER_READY_REQUEST_TIMEOUT_MS: u64 = 2_000; +const DOCKER_RM_TIMEOUT_SECS: u64 = 15; +const DOCKER_INSPECT_TIMEOUT_SECS: u64 = 5; +// Serializes inspect+rm recovery across nextest processes sharing the reused +// container. Held only on the failure path. +const RECOVERY_LOCK_FILE_NAME: &str = "iggy-test-elasticsearch-recovery.lock"; +// Bound lock wait so a peer mid-recovery cannot park a current-thread runtime +// for two full container startups with no upper limit. +const RECOVERY_LOCK_TIMEOUT_SECS: u64 = 30; +const RECOVERY_LOCK_POLL_MS: u64 = 50; +// Indices from prior runs older than this are leftovers: a live concurrent +// test's index is seconds old, so age-based sweeping never races other tests +// sharing the reused container. +const STALE_INDEX_MAX_AGE_MS: u128 = 30 * 60 * 1000; +const STALE_INDEX_PATTERNS: &str = "iggy_messages_*,test_documents_*"; pub const DEFAULT_TEST_STREAM: &str = "test_stream"; pub const DEFAULT_TEST_TOPIC: &str = "test_topic"; @@ -101,6 +123,45 @@ pub struct ElasticsearchContainer { impl ElasticsearchContainer { pub async fn start() -> Result { + match Self::try_start().await { + Ok(started) => Ok(started), + Err(first_error) => { + // Shared Always-reuse container: never `docker rm -f` on a + // transient start/port/health flake. Another nextest worker may + // still be using a healthy instance. Recover only when inspect + // shows a removable wedged state, under a cross-process lock. + let recovery_lock = match acquire_recovery_lock().await { + Ok(lock) => lock, + Err(error) => { + warn!("Skipping Elasticsearch recovery, could not take lock: {error}"); + return Err(first_error); + } + }; + + if let Ok(started) = Self::try_start().await { + drop(recovery_lock); + return Ok(started); + } + + if !container_is_removable_wedged().await { + warn!( + "Elasticsearch start failed and '{ELASTICSEARCH_CONTAINER_NAME}' is not in a removable wedged state; leaving it in place: {first_error}" + ); + return Err(first_error); + } + + warn!( + "Elasticsearch container wedged, removing '{ELASTICSEARCH_CONTAINER_NAME}' and retrying once: {first_error}" + ); + force_remove_container().await; + let retry_result = Self::try_start().await; + drop(recovery_lock); + retry_result + } + } + } + + async fn try_start() -> Result { let container = GenericImage::new(ELASTICSEARCH_IMAGE, ELASTICSEARCH_TAG) .with_exposed_port(ELASTICSEARCH_PORT.tcp()) .with_wait_for(WaitFor::http( @@ -137,14 +198,320 @@ impl ElasticsearchContainer { message: "No mapping for Elasticsearch port".to_string(), })?; - let base_url = format!("http://localhost:{mapped_port}"); + // Prefer IPv4 loopback: Docker publishes 0.0.0.0:HOST→9200. `localhost` + // can resolve to ::1 first on macOS and black-hole the elasticsearch-rs + // client while the fixture's reqwest client still looks healthy. + let base_url = format!("http://127.0.0.1:{mapped_port}"); info!("Elasticsearch container available at {base_url}"); - Ok(Self { + let started = Self { container, base_url, + }; + // ReuseDirective::Always can attach to a days-old container without + // re-running HttpWaitStrategy; verify cluster health on every setup. + started.wait_until_ready().await?; + started.sweep_stale_indices().await; + Ok(started) + } + + async fn wait_until_ready(&self) -> Result<(), TestBinaryError> { + #[derive(Deserialize)] + struct ClusterHealth { + timed_out: bool, + status: String, + } + + // Dedicated probe client: short timeout, no retry middleware. The shared + // create_http_client() (30s + 3 retries) turns one black-holed request + // into a multi-minute hang that looks like the test is stuck. + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_millis( + CLUSTER_READY_REQUEST_TIMEOUT_MS, + )) + .build() + .map_err(|error| TestBinaryError::FixtureSetup { + fixture_type: "ElasticsearchContainer".to_string(), + message: format!("Failed to build readiness HTTP client: {error}"), + })?; + // wait_for_status=yellow makes ES block until yellow/green or timeout=1s + // elapses; without wait_for_*, timeout is ignored and red still returns + // 200 with timed_out=false. + let health_url = format!( + "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?wait_for_status=yellow&timeout=1s", + self.base_url + ); + let mut last_error = String::from("no attempts made"); + + for attempt in 1..=CLUSTER_READY_ATTEMPTS { + match client.get(&health_url).send().await { + Ok(response) if response.status().is_success() => { + match response.json::().await { + Ok(health) if !health.timed_out => { + info!( + "Elasticsearch cluster ready at {} (status={})", + self.base_url, health.status + ); + return Ok(()); + } + Ok(health) => { + last_error = format!( + "cluster health timed out on attempt {attempt}/{CLUSTER_READY_ATTEMPTS} (status={})", + health.status + ); + } + Err(error) => { + last_error = format!( + "cluster health body unparsable on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {error}" + ); + } + } + } + Ok(response) => { + last_error = format!( + "cluster health status {} on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}", + response.status() + ); + } + Err(error) => { + last_error = format!( + "cluster health request failed on attempt {attempt}/{CLUSTER_READY_ATTEMPTS}: {error}" + ); + } + } + tokio::time::sleep(std::time::Duration::from_millis(CLUSTER_READY_INTERVAL_MS)).await; + } + + Err(TestBinaryError::FixtureSetup { + fixture_type: "ElasticsearchContainer".to_string(), + message: format!( + "Elasticsearch at {} not ready after {CLUSTER_READY_ATTEMPTS} attempts: {last_error}", + self.base_url + ), }) } + + /// Delete leftover test indices (empty or partially filled) from previous + /// runs so accumulated shards do not degrade the reused container. Only + /// indices older than [`STALE_INDEX_MAX_AGE_MS`] are removed, which keeps + /// the sweep safe against tests running concurrently in other processes. + /// Best-effort: failures are logged, never fail the fixture. + async fn sweep_stale_indices(&self) { + #[derive(Deserialize)] + struct CatIndexEntry { + index: String, + #[serde(rename = "creation.date")] + creation_date: Option, + } + + // Short timeout, no retries: sweep is best-effort and must not stall setup. + let Ok(client) = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build() + else { + warn!("Skipping stale index sweep, failed to build HTTP client"); + return; + }; + let cat_url = format!( + "{}/_cat/indices/{STALE_INDEX_PATTERNS}?format=json&h=index,creation.date", + self.base_url + ); + + let entries = match client.get(&cat_url).send().await { + Ok(response) if response.status().is_success() => { + match response.json::>().await { + Ok(entries) => entries, + Err(error) => { + warn!("Skipping stale index sweep, unparsable _cat response: {error}"); + return; + } + } + } + Ok(response) => { + warn!( + "Skipping stale index sweep, _cat/indices returned {}", + response.status() + ); + return; + } + Err(error) => { + warn!("Skipping stale index sweep, _cat/indices failed: {error}"); + return; + } + }; + + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|duration| duration.as_millis()) + .unwrap_or(0); + + let stale: Vec = entries + .into_iter() + .filter_map(|entry| { + let created_ms = entry.creation_date.as_deref()?.parse::().ok()?; + (now_ms.saturating_sub(created_ms) > STALE_INDEX_MAX_AGE_MS).then_some(entry.index) + }) + .collect(); + + if stale.is_empty() { + return; + } + + // Chunked so the URL stays well under limits with many leftovers. + for chunk in stale.chunks(20) { + let delete_url = format!("{}/{}", self.base_url, chunk.join(",")); + match client.delete(&delete_url).send().await { + Ok(response) if response.status().is_success() => { + info!("Deleted {} stale Elasticsearch test indices", chunk.len()); + } + Ok(response) => { + warn!( + "Failed to delete stale Elasticsearch indices, status {}", + response.status() + ); + } + Err(error) => { + warn!("Failed to delete stale Elasticsearch indices: {error}"); + } + } + } + } +} + +/// Cross-process advisory lock for inspect+rm recovery of the shared reuse +/// container. Dropping the file releases the lock (including on process crash). +/// +/// Assumes every nextest process shares the same `std::env::temp_dir()` (and +/// thus the same lock file). A per-process `TMPDIR` makes each worker lock its +/// own path and recovery can race silently. +struct RecoveryLock { + _file: File, +} + +async fn acquire_recovery_lock() -> Result { + let path = std::env::temp_dir().join(RECOVERY_LOCK_FILE_NAME); + let file = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(&path) + .map_err(|error| format!("open {}: {error}", path.display()))?; + + let deadline = Instant::now() + Duration::from_secs(RECOVERY_LOCK_TIMEOUT_SECS); + loop { + match file.try_lock() { + Ok(()) => return Ok(RecoveryLock { _file: file }), + Err(TryLockError::WouldBlock) if Instant::now() < deadline => { + tokio::time::sleep(Duration::from_millis(RECOVERY_LOCK_POLL_MS)).await; + } + Err(TryLockError::WouldBlock) => { + return Err(format!( + "timed out after {RECOVERY_LOCK_TIMEOUT_SECS}s waiting for {}", + path.display() + )); + } + Err(TryLockError::Error(error)) => { + return Err(format!("lock {}: {error}", path.display())); + } + } + } +} + +/// True only when Docker says the shared container is in a state safe to +/// force-remove without yanking a healthy instance from a peer process. +/// +/// Removable: exited/dead/created/paused/restarting, or Docker healthcheck +/// `unhealthy`. A plain `running` container (even if ES HTTP is flaky) is left +/// alone: readiness flakes must not `docker rm -f` a shared Always-reuse box. +async fn container_is_removable_wedged() -> bool { + let inspect_format = "{{.State.Status}}|{{if .State.Health}}{{.State.Health.Status}}{{end}}"; + let Some(stdout) = docker_command_stdout( + &[ + "inspect", + "-f", + inspect_format, + ELASTICSEARCH_CONTAINER_NAME, + ], + DOCKER_INSPECT_TIMEOUT_SECS, + ) + .await + else { + return false; + }; + + let mut parts = stdout.trim().split('|'); + let status = parts.next().unwrap_or("").trim(); + let health = parts.next().unwrap_or("").trim(); + + matches!( + status, + "exited" | "dead" | "created" | "paused" | "restarting" + ) || health == "unhealthy" +} + +async fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option { + let mut command = tokio::process::Command::new("docker"); + command + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .kill_on_drop(true); + + match tokio::time::timeout(Duration::from_secs(timeout_secs), command.output()).await { + Ok(Ok(output)) if output.status.success() => { + Some(String::from_utf8_lossy(&output.stdout).into_owned()) + } + Ok(Ok(_)) => None, + Ok(Err(error)) => { + warn!("Failed to invoke docker {}: {error}", args.join(" ")); + None + } + Err(_) => { + warn!("docker {} timed out after {timeout_secs}s", args.join(" ")); + None + } + } +} + +/// Remove the shared reuse container so the next start creates a fresh one. +/// Uses the Docker CLI directly: testcontainers offers no "remove by name" API +/// and the wedged container was created by an earlier process anyway. +/// Caller must hold [`RecoveryLock`] and have confirmed +/// [`container_is_removable_wedged`]. +async fn force_remove_container() { + let mut command = tokio::process::Command::new("docker"); + command + .args(["rm", "-f", ELASTICSEARCH_CONTAINER_NAME]) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + + match tokio::time::timeout( + Duration::from_secs(DOCKER_RM_TIMEOUT_SECS), + command.output(), + ) + .await + { + Ok(Ok(output)) if output.status.success() => { + info!("Removed wedged Elasticsearch container '{ELASTICSEARCH_CONTAINER_NAME}'"); + } + Ok(Ok(output)) => { + let stderr = String::from_utf8_lossy(&output.stderr); + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} failed (exit {}): {stderr}", + output.status + ); + } + Ok(Err(error)) => { + warn!("Failed to invoke docker rm for '{ELASTICSEARCH_CONTAINER_NAME}': {error}"); + } + Err(_) => { + warn!( + "docker rm -f {ELASTICSEARCH_CONTAINER_NAME} timed out after {DOCKER_RM_TIMEOUT_SECS}s" + ); + } + } } pub fn create_http_client() -> HttpClient { diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs index 10462a2028..f8668ac254 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/sink.rs @@ -33,6 +33,7 @@ use uuid::Uuid; const SINK_INDEX_PREFIX: &str = "iggy_messages"; const POLL_ATTEMPTS: usize = 100; const POLL_INTERVAL_MS: u64 = 50; +const REFRESH_PROBE_TIMEOUT_MS: u64 = 2_000; pub struct ElasticsearchSinkFixture { container: ElasticsearchContainer, @@ -61,22 +62,39 @@ impl ElasticsearchSinkFixture { &self, expected_count: usize, ) -> Result { + let mut last_error: Option = None; + for _ in 0..POLL_ATTEMPTS { + // Short-timeout probe: create_http_client() is 30s + 3 retries and + // would inflate failure-path wait_for_documents up to minutes. + if let Err(error) = self.refresh_index_probe().await { + last_error = Some(error); + sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; + continue; + } + match self.count_documents(&self.index).await { Ok(count) if count >= expected_count => { info!("Found {count} documents in Elasticsearch (expected {expected_count})"); return Ok(count); } - Ok(_) => {} - Err(_) => {} + Ok(_) => { + last_error = None; + } + Err(error) => { + last_error = Some(error); + } } sleep(Duration::from_millis(POLL_INTERVAL_MS)).await; } let final_count = self.count_documents(&self.index).await.unwrap_or(0); + let detail = last_error + .map(|error| format!("; last error: {error}")) + .unwrap_or_default(); Err(TestBinaryError::InvalidState { message: format!( - "Expected at least {expected_count} documents, found {final_count} after {POLL_ATTEMPTS} attempts" + "Expected at least {expected_count} documents, found {final_count} after {POLL_ATTEMPTS} attempts{detail}" ), }) } @@ -88,6 +106,33 @@ impl ElasticsearchSinkFixture { pub async fn refresh_index(&self) -> Result<(), TestBinaryError> { ElasticsearchOps::refresh_index(self, &self.index).await } + + /// Refresh with a short-timeout client for the document poll loop. + async fn refresh_index_probe(&self) -> Result<(), TestBinaryError> { + let client = reqwest::Client::builder() + .timeout(Duration::from_millis(REFRESH_PROBE_TIMEOUT_MS)) + .build() + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to build refresh probe client: {error}"), + })?; + let url = format!("{}/{}/_refresh", self.container.base_url, self.index); + let response = + client + .post(&url) + .send() + .await + .map_err(|error| TestBinaryError::InvalidState { + message: format!("Failed to refresh index: {error}"), + })?; + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(TestBinaryError::InvalidState { + message: format!("Failed to refresh index: status={status}, body={body}"), + }); + } + Ok(()) + } } #[async_trait] @@ -97,8 +142,6 @@ impl TestFixture for ElasticsearchSinkFixture { let http_client = create_http_client(); let index = format!("{SINK_INDEX_PREFIX}_{}", Uuid::new_v4().simple()); - // Container startup already waits for /_cluster/health to return 200 - // via HttpWaitStrategy, so no additional health check is needed. Ok(Self { container, http_client, diff --git a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs index e6c42096bc..2368fce3d5 100644 --- a/core/integration/tests/connectors/fixtures/elasticsearch/source.rs +++ b/core/integration/tests/connectors/fixtures/elasticsearch/source.rs @@ -101,8 +101,6 @@ impl TestFixture for ElasticsearchSourceFixture { let http_client = create_http_client(); let index = format!("{TEST_INDEX_PREFIX}_{}", Uuid::new_v4().simple()); - // Container startup already waits for /_cluster/health to return 200 - // via HttpWaitStrategy, so no additional health check is needed. Ok(Self { container, http_client,