Surface OpenSearch errors in readiness check

Replace the previous ping-based readiness probe with a cluster.health() call so transport/auth/connection errors are propagated instead of swallowed. Track the last error and include structured details (error_type, status_code, info) in per-attempt logs, and embed the last error in the final OpenSearchNotReadyError message. Add unit tests to verify green/yellow success, auth failures surface status_code/info, and connection errors without status_code are reported.

Author: edwinjosechittilappilly

src/utils/opensearch_utils.py

@@ -71,6 +71,7 @@ async def wait_for_opensearch(
     Raises:
         OpenSearchNotReadyError: If OpenSearch fails to become ready within the retry limit.
     """
+    last_error: str | None = None
     for attempt in range(max_retries):
         display_attempt: int = attempt + 1
 
@@ -81,34 +82,34 @@ async def wait_for_opensearch(
         )
 
         try:
-            # Simple ping to check connection
-            if await opensearch_client.ping():
-                # Also check cluster health
-                health = await opensearch_client.cluster.health()
-                status = health.get("status")
-                if status in ["green", "yellow"]:
-                    logger.info(
-                        "Successfully verified that OpenSearch is ready.",
-                        attempt=display_attempt,
-                        status=status,
-                    )
-                    return
-                else:
-                    logger.warning(
-                        "OpenSearch is up but cluster health is red.",
-                        attempt=display_attempt,
-                        status=status,
-                    )
-            else:
-                logger.warning(
-                    "OpenSearch ping failed.",
+            # cluster.health() (GET /_cluster/health) raises on
+            # connection/auth/transport errors, unlike ping() (HEAD /), which
+            # swallows them and returns False — hiding the real cause (e.g. a
+            # rejected JWT, a 403, or a connection refusal).
+            health = await opensearch_client.cluster.health()
+            status = health.get("status")
+            if status in ["green", "yellow"]:
+                logger.info(
+                    "Successfully verified that OpenSearch is ready.",
                     attempt=display_attempt,
+                    status=status,
                 )
+                return
+            last_error = f"cluster health status={status}"
+            logger.warning(
+                "OpenSearch is up but cluster health is red.",
+                attempt=display_attempt,
+                status=status,
+            )
         except Exception as e:
+            last_error = f"{type(e).__name__}: {e}"
             logger.warning(
-                "OpenSearch is not ready.",
+                "OpenSearch readiness check failed.",
                 attempt=display_attempt,
+                error_type=type(e).__name__,
+                status_code=getattr(e, "status_code", None),
                 error=str(e),
+                info=getattr(e, "info", None),
             )
 
         if attempt < max_retries - 1:
@@ -123,8 +124,8 @@ async def wait_for_opensearch(
 
             await asyncio.sleep(delay)
 
-    message: str = "Failed to verify whether OpenSearch is ready."
-    logger.error(message)
+    message: str = f"Failed to verify whether OpenSearch is ready. Last error: {last_error}"
+    logger.error(message, last_error=last_error)
     raise OpenSearchNotReadyError(message)
 
 

tests/unit/test_wait_for_opensearch_error_surfacing.py

@@ -0,0 +1,96 @@
+"""wait_for_opensearch must surface the real failure reason.
+
+Previously the readiness probe gated on ``client.ping()`` (HEAD /), which
+opensearchpy implements by swallowing transport/auth/connection errors and
+returning ``False`` — collapsing a rejected JWT (401/403), a connection
+refusal, and a TLS error into a single uninformative "ping failed" log.
+
+These tests pin the new behaviour: the probe calls ``cluster.health()``
+(which raises), the per-attempt warning carries ``status_code``/``info``, and
+the final ``OpenSearchNotReadyError`` message embeds the last error.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent.parent
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from utils.opensearch_utils import (  # noqa: E402
+    OpenSearchNotReadyError,
+    wait_for_opensearch,
+)
+
+
+class _FakeTransportError(Exception):
+    """Mimics opensearchpy's TransportError: carries status_code + info."""
+
+    def __init__(self, status_code, error, info):
+        super().__init__(error)
+        self.status_code = status_code
+        self.error = error
+        self.info = info
+
+
+def _client_with_health(health_side_effect):
+    client = MagicMock()
+    client.cluster.health = AsyncMock(side_effect=health_side_effect)
+    return client
+
+
+@pytest.mark.asyncio
+async def test_ready_when_health_green():
+    client = _client_with_health([{"status": "green"}])
+    # Should return without raising; no sleeps on first-attempt success.
+    await wait_for_opensearch(client, max_retries=1)
+    client.cluster.health.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_auth_failure_is_surfaced(monkeypatch):
+    err = _FakeTransportError(403, "AuthorizationException", {"reason": "invalid jwt"})
+    client = _client_with_health(err)
+
+    captured = []
+    monkeypatch.setattr(
+        "utils.opensearch_utils.logger.warning",
+        lambda msg, **kw: captured.append((msg, kw)),
+    )
+
+    with pytest.raises(OpenSearchNotReadyError) as excinfo:
+        # tiny delays so the retry loop doesn't actually sleep meaningfully
+        await wait_for_opensearch(client, max_retries=2, base_delay=0.0, max_delay=0.0)
+
+    # The real reason must reach the final exception message...
+    assert "403" in str(excinfo.value) or "AuthorizationException" in str(excinfo.value)
+    # ...and the per-attempt warning must carry structured detail.
+    assert captured, "expected at least one readiness-failure warning"
+    _, kw = captured[0]
+    assert kw["status_code"] == 403
+    assert kw["info"] == {"reason": "invalid jwt"}
+    assert kw["error_type"] == "_FakeTransportError"
+
+
+@pytest.mark.asyncio
+async def test_connection_error_without_status_code(monkeypatch):
+    # A plain ConnectionError-style failure has no status_code attribute.
+    client = _client_with_health(ConnectionError("connection refused"))
+
+    captured = []
+    monkeypatch.setattr(
+        "utils.opensearch_utils.logger.warning",
+        lambda msg, **kw: captured.append((msg, kw)),
+    )
+
+    with pytest.raises(OpenSearchNotReadyError) as excinfo:
+        await wait_for_opensearch(client, max_retries=1, base_delay=0.0, max_delay=0.0)
+
+    assert "connection refused" in str(excinfo.value)
+    _, kw = captured[0]
+    assert kw["status_code"] is None
+    assert kw["error_type"] == "ConnectionError"