ryerraguntla commented on code in PR #3729:
URL: https://github.com/apache/iggy/pull/3729#discussion_r3706589460
##########
core/integration/tests/connectors/fixtures/elasticsearch/container.rs:
##########
@@ -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::<ClusterHealth>().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<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).
+///
+/// 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<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()))?;
+
+ 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"
Review Comment:
Fixed in
[58858e8](https://github.com/apache/iggy/pull/3729/commits/58858e820343ac4745449cf291f9354a6269c49b)
--
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]
