hubcio commented on code in PR #3729:
URL: https://github.com/apache/iggy/pull/3729#discussion_r3636828449
##########
core/integration/tests/connectors/fixtures/elasticsearch/container.rs:
##########
@@ -137,14 +195,308 @@ 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> {
+ // 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}"),
+ })?;
+ // timeout=1s keeps ES from holding the request when the cluster is
slow.
+ let health_url = format!(
+ "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?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() => {
+ let body = response.text().await.unwrap_or_default();
+ if body.contains("\"timed_out\":true") {
+ last_error = format!(
+ "cluster health timed out on attempt
{attempt}/{CLUSTER_READY_ATTEMPTS}: {body}"
+ );
+ } else {
+ info!("Elasticsearch cluster ready at {}",
self.base_url);
+ return Ok(());
+ }
+ }
+ 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<String>,
+ }
+
+ // 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::<Vec<CatIndexEntry>>().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<String> = entries
+ .into_iter()
+ .filter_map(|entry| {
+ let created_ms =
entry.creation_date.as_deref()?.parse::<u128>().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).
+struct RecoveryLock {
+ _file: File,
+}
+
+fn acquire_recovery_lock() -> Result<RecoveryLock, String> {
+ 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()))?;
+ file.lock()
Review Comment:
`File::lock()` blocks with no acquire timeout - worst case a peer holds this
through its whole recovery (two 120s container starts plus readiness waits),
and since the test runtime is current-thread, the block parks the entire
runtime for that long. bounded and failure-path-only, so nice-to-have:
`try_lock()` with a deadline.
##########
core/integration/tests/connectors/fixtures/elasticsearch/sink.rs:
##########
@@ -61,22 +61,38 @@ impl ElasticsearchSinkFixture {
&self,
expected_count: usize,
) -> Result<usize, TestBinaryError> {
+ let mut last_error: Option<TestBinaryError> = None;
+
for _ in 0..POLL_ATTEMPTS {
+ // Refresh so near-real-time search/count sees recently indexed
docs.
+ if let Err(error) = self.refresh_index().await {
Review Comment:
`refresh_index()` goes through `create_http_client()` (30s timeout + 3
retries) inside a 50ms poll loop - on the failure path that is up to 100
refreshes, each with retry backoff. happy path exits in a few iterations so
this only inflates time-to-failure; the short-timeout probe-client pattern from
container.rs would fit here too.
##########
core/integration/tests/connectors/fixtures/elasticsearch/container.rs:
##########
@@ -137,14 +195,308 @@ 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> {
+ // 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}"),
+ })?;
+ // timeout=1s keeps ES from holding the request when the cluster is
slow.
+ let health_url = format!(
+ "{}{ELASTICSEARCH_HEALTH_ENDPOINT}?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() => {
+ let body = response.text().await.unwrap_or_default();
+ if body.contains("\"timed_out\":true") {
+ last_error = format!(
+ "cluster health timed out on attempt
{attempt}/{CLUSTER_READY_ATTEMPTS}: {body}"
+ );
+ } else {
+ info!("Elasticsearch cluster ready at {}",
self.base_url);
+ return Ok(());
+ }
+ }
+ 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<String>,
+ }
+
+ // 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::<Vec<CatIndexEntry>>().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<String> = entries
+ .into_iter()
+ .filter_map(|entry| {
+ let created_ms =
entry.creation_date.as_deref()?.parse::<u128>().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).
+struct RecoveryLock {
+ _file: File,
+}
+
+fn acquire_recovery_lock() -> Result<RecoveryLock, String> {
+ 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()))?;
+ file.lock()
+ .map_err(|error| format!("lock {}: {error}", path.display()))?;
+ Ok(RecoveryLock { _file: file })
+}
+
+/// 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.
+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,
+ ) 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"
+}
+
+fn docker_command_stdout(args: &[&str], timeout_secs: u64) -> Option<String> {
Review Comment:
`docker_command_stdout` and `force_remove_container` duplicate the spawn /
`try_wait` / deadline / kill loop, and both block the async caller
(`thread::sleep` on a current-thread runtime). `tokio::process::Command` +
`tokio::time::timeout(dur, cmd.output())` + `kill_on_drop(true)` collapses both
to a few lines with the same 5s/15s kill semantics - `delta/fixture.rs:254` is
the in-repo precedent and tokio `full` is already enabled here. it also removes
a latent footgun: stdout is only drained after exit, so docker output over the
pipe buffer (~64KB) would deadlock until the timeout kill. fine today with
one-line `inspect -f` output, but the helper reads as generic.
--
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]
