Don't swallow unexpected errors during Elasticsearch verification

elasticsearch/_async/transport.py

@@ -120,7 +120,7 @@ class AsyncTransport(Transport):
 
         # Set our 'verified_once' implementation to one that
         # works with 'asyncio' instead of 'threading'
-        self._verified_once = Once()
+        self._verify_elasticsearch_lock = asyncio.Lock()
 
         # Now that we have a loop we can create all our HTTP connections...
         self.set_connections(self.hosts)
@@ -338,9 +338,7 @@ class AsyncTransport(Transport):
 
         # Before we make the actual API call we verify the Elasticsearch instance.
         if self._verified_elasticsearch is None:
-            await self._verified_once.call(
+            await self._do_verify_elasticsearch(headers=headers, timeout=timeout)
-                self._do_verify_elasticsearch, headers=headers, timeout=timeout
-            )
 
         # If '_verified_elasticsearch' is False we know we're not connected to Elasticsearch.
         if self._verified_elasticsearch is False:
@@ -431,74 +429,73 @@ class AsyncTransport(Transport):
         but we're also unable to rule it out due to a permission
         error we instead emit an 'ElasticsearchWarning'.
         """
-        # Product check has already been done, no need to do again.
+        # Ensure that there's only one async exec within this section
-        if self._verified_elasticsearch:
+        # at a time to not emit unnecessary index API calls.
-            return
+        async with self._verify_elasticsearch_lock:
 
-        headers = {header.lower(): value for header, value in (headers or {}).items()}
+            # Product check has already been completed while we were
-        # We know we definitely want JSON so request it via 'accept'
+            # waiting our turn, no need to do again.
-        headers.setdefault("accept", "application/json")
+            if self._verified_elasticsearch is not None:
-
-        info_headers = {}
-        info_response = {}
-        error = None
-
-        for conn in chain(self.connection_pool.connections, self.seed_connections):
-            try:
-                _, info_headers, info_response = await conn.perform_request(
-                    "GET", "/", headers=headers, timeout=timeout
-                )
-
-                # Lowercase all the header names for consistency in accessing them.
-                info_headers = {
-                    header.lower(): value for header, value in info_headers.items()
-                }
-
-                info_response = self.deserializer.loads(
-                    info_response, mimetype="application/json"
-                )
-                break
-
-            # Previous versions of 7.x Elasticsearch required a specific
-            # permission so if we receive HTTP 401/403 we should warn
-            # instead of erroring out.
-            except (AuthenticationException, AuthorizationException):
-                warnings.warn(
-                    (
-                        "The client is unable to verify that the server is "
-                        "Elasticsearch due security privileges on the server side"
-                    ),
-                    ElasticsearchWarning,
-                    stacklevel=4,
-                )
-                self._verified_elasticsearch = True
                 return
 
-            # This connection didn't work, we'll try another.
+            headers = {
-            except (ConnectionError, SerializationError) as err:
+                header.lower(): value for header, value in (headers or {}).items()
-                if error is None:
+            }
-                    error = err
+            # We know we definitely want JSON so request it via 'accept'
+            headers.setdefault("accept", "application/json")
 
-        # If we received a connection error and weren't successful
+            info_headers = {}
-        # anywhere then we reraise the more appropriate error.
+            info_response = {}
-        if error and not info_response:
+            error = None
-            raise error
 
-        # Check the information we got back from the index request.
+            attempted_conns = []
-        self._verified_elasticsearch = _verify_elasticsearch(
+            for conn in chain(self.connection_pool.connections, self.seed_connections):
-            info_headers, info_response
+                # Only attempt once per connection max.
-        )
+                if conn in attempted_conns:
+                    continue
+                attempted_conns.append(conn)
 
+                try:
+                    _, info_headers, info_response = await conn.perform_request(
+                        "GET", "/", headers=headers, timeout=timeout
+                    )
 
-class Once:
+                    # Lowercase all the header names for consistency in accessing them.
-    """Simple class which forces an async function to only execute once."""
+                    info_headers = {
+                        header.lower(): value for header, value in info_headers.items()
+                    }
 
-    def __init__(self):
+                    info_response = self.deserializer.loads(
-        self._lock = asyncio.Lock()
+                        info_response, mimetype="application/json"
-        self._called = False
+                    )
+                    break
 
-    async def call(self, func, *args, **kwargs):
+                # Previous versions of 7.x Elasticsearch required a specific
-        async with self._lock:
+                # permission so if we receive HTTP 401/403 we should warn
-            if not self._called:
+                # instead of erroring out.
-                self._called = True
+                except (AuthenticationException, AuthorizationException):
-                await func(*args, **kwargs)
+                    warnings.warn(
+                        (
+                            "The client is unable to verify that the server is "
+                            "Elasticsearch due security privileges on the server side"
+                        ),
+                        ElasticsearchWarning,
+                        stacklevel=4,
+                    )
+                    self._verified_elasticsearch = True
+                    return
+
+                # This connection didn't work, we'll try another.
+                except (ConnectionError, SerializationError, TransportError) as err:
+                    if error is None:
+                        error = err
+
+            # If we received a connection error and weren't successful
+            # anywhere then we re-raise the more appropriate error.
+            if error and not info_response:
+                raise error
+
+            # Check the information we got back from the index request.
+            self._verified_elasticsearch = _verify_elasticsearch(
+                info_headers, info_response
+            )

elasticsearch/transport.py

@@ -220,7 +220,7 @@ class Transport(object):
 
         # Ensures that the ES verification request only fires once and that
         # all requests block until this request returns back.
-        self._verified_once = Once()
+        self._verify_elasticsearch_lock = Lock()
 
     def add_connection(self, host):
         """
@@ -406,9 +406,7 @@ class Transport(object):
 
         # Before we make the actual API call we verify the Elasticsearch instance.
         if self._verified_elasticsearch is None:
-            self._verified_once.call(
+            self._do_verify_elasticsearch(headers=headers, timeout=timeout)
-                self._do_verify_elasticsearch, headers=headers, timeout=timeout
-            )
 
         # If '_verified_elasticsearch' is False we know we're not connected to Elasticsearch.
         if self._verified_elasticsearch is False:
@@ -536,63 +534,76 @@ class Transport(object):
         but we're also unable to rule it out due to a permission
         error we instead emit an 'ElasticsearchWarning'.
         """
-        # Product check has already been done, no need to do again.
+        # Ensure that there's only one thread within this section
-        if self._verified_elasticsearch is not None:
+        # at a time to not emit unnecessary index API calls.
-            return
+        with self._verify_elasticsearch_lock:
 
-        headers = {header.lower(): value for header, value in (headers or {}).items()}
+            # Product check has already been completed while we were
-        # We know we definitely want JSON so request it via 'accept'
+            # waiting our turn, no need to do again.
-        headers.setdefault("accept", "application/json")
+            if self._verified_elasticsearch is not None:
-
-        info_headers = {}
-        info_response = {}
-        error = None
-
-        for conn in chain(self.connection_pool.connections, self.seed_connections):
-            try:
-                _, info_headers, info_response = conn.perform_request(
-                    "GET", "/", headers=headers, timeout=timeout
-                )
-
-                # Lowercase all the header names for consistency in accessing them.
-                info_headers = {
-                    header.lower(): value for header, value in info_headers.items()
-                }
-
-                info_response = self.deserializer.loads(
-                    info_response, mimetype="application/json"
-                )
-                break
-
-            # Previous versions of 7.x Elasticsearch required a specific
-            # permission so if we receive HTTP 401/403 we should warn
-            # instead of erroring out.
-            except (AuthenticationException, AuthorizationException):
-                warnings.warn(
-                    (
-                        "The client is unable to verify that the server is "
-                        "Elasticsearch due security privileges on the server side"
-                    ),
-                    ElasticsearchWarning,
-                    stacklevel=5,
-                )
-                self._verified_elasticsearch = True
                 return
 
-            # This connection didn't work, we'll try another.
+            headers = {
-            except (ConnectionError, SerializationError) as err:
+                header.lower(): value for header, value in (headers or {}).items()
-                if error is None:
+            }
-                    error = err
+            # We know we definitely want JSON so request it via 'accept'
+            headers.setdefault("accept", "application/json")
 
-        # If we received a connection error and weren't successful
+            info_headers = {}
-        # anywhere then we reraise the more appropriate error.
+            info_response = {}
-        if error and not info_response:
+            error = None
-            raise error
 
-        # Check the information we got back from the index request.
+            attempted_conns = []
-        self._verified_elasticsearch = _verify_elasticsearch(
+            for conn in chain(self.connection_pool.connections, self.seed_connections):
-            info_headers, info_response
+                # Only attempt once per connection max.
-        )
+                if conn in attempted_conns:
+                    continue
+                attempted_conns.append(conn)
+
+                try:
+                    _, info_headers, info_response = conn.perform_request(
+                        "GET", "/", headers=headers, timeout=timeout
+                    )
+
+                    # Lowercase all the header names for consistency in accessing them.
+                    info_headers = {
+                        header.lower(): value for header, value in info_headers.items()
+                    }
+
+                    info_response = self.deserializer.loads(
+                        info_response, mimetype="application/json"
+                    )
+                    break
+
+                # Previous versions of 7.x Elasticsearch required a specific
+                # permission so if we receive HTTP 401/403 we should warn
+                # instead of erroring out.
+                except (AuthenticationException, AuthorizationException):
+                    warnings.warn(
+                        (
+                            "The client is unable to verify that the server is "
+                            "Elasticsearch due security privileges on the server side"
+                        ),
+                        ElasticsearchWarning,
+                        stacklevel=5,
+                    )
+                    self._verified_elasticsearch = True
+                    return
+
+                # This connection didn't work, we'll try another.
+                except (ConnectionError, SerializationError, TransportError) as err:
+                    if error is None:
+                        error = err
+
+            # If we received a connection error and weren't successful
+            # anywhere then we re-raise the more appropriate error.
+            if error and not info_response:
+                raise error
+
+            # Check the information we got back from the index request.
+            self._verified_elasticsearch = _verify_elasticsearch(
+                info_headers, info_response
+            )
 
 
 def _verify_elasticsearch(headers, response):
@@ -640,17 +651,3 @@ def _verify_elasticsearch(headers, response):
         return False
 
     return True
-
-
-class Once:
-    """Simple class which forces a function to only execute once."""
-
-    def __init__(self):
-        self._lock = Lock()
-        self._called = False
-
-    def call(self, func, *args, **kwargs):
-        with self._lock:
-            if not self._called:
-                self._called = True
-                func(*args, **kwargs)

test_elasticsearch/test_async/test_transport.py

@@ -34,6 +34,7 @@ from elasticsearch.exceptions import (
     ConnectionError,
     ElasticsearchWarning,
     NotElasticsearchError,
+    NotFoundError,
     TransportError,
 )
 
@@ -770,7 +771,9 @@ class TestTransport:
         # The rest of the requests are 'GET /_search' afterwards
         assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
 
-    async def test_multiple_requests_verify_elasticsearch_errors(self, event_loop):
+    async def test_multiple_requests_verify_elasticsearch_product_error(
+        self, event_loop
+    ):
         t = AsyncTransport(
             [
                 {
@@ -823,3 +826,53 @@ class TestTransport:
 
         # The rest of the requests are 'GET /_search' afterwards
         assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
+
+    @pytest.mark.parametrize("error_cls", [ConnectionError, NotFoundError])
+    async def test_multiple_requests_verify_elasticsearch_retry_on_errors(
+        self, event_loop, error_cls
+    ):
+        t = AsyncTransport(
+            [
+                {
+                    "exception": error_cls(),
+                    "delay": 0.1,
+                }
+            ],
+            connection_class=DummyConnection,
+        )
+
+        results = []
+        completed_at = []
+
+        async def request_task():
+            try:
+                results.append(await t.perform_request("GET", "/_search"))
+            except Exception as e:
+                results.append(e)
+            completed_at.append(event_loop.time())
+
+        # Execute a bunch of requests concurrently.
+        tasks = []
+        start_time = event_loop.time()
+        for _ in range(5):
+            tasks.append(event_loop.create_task(request_task()))
+        await asyncio.gather(*tasks)
+        end_time = event_loop.time()
+
+        # Exactly 5 results completed
+        assert len(results) == 5
+
+        # All results were errors and not wrapped in 'NotElasticsearchError'
+        assert all(isinstance(result, error_cls) for result in results)
+
+        # Assert that 5 requests were made in total (5 transport requests per x 0.1s/conn request)
+        duration = end_time - start_time
+        assert 0.5 <= duration <= 0.6
+
+        # Assert that the cluster is still in the unknown/unverified stage.
+        assert t._verified_elasticsearch is None
+
+        # See that the API isn't hit, instead it's the index requests that are failing.
+        calls = t.connection_pool.connections[0].calls
+        assert len(calls) == 5
+        assert all(call[0] == ("GET", "/") for call in calls)

test_elasticsearch/test_transport.py

@@ -32,6 +32,7 @@ from elasticsearch.exceptions import (
     ConnectionError,
     ElasticsearchWarning,
     NotElasticsearchError,
+    NotFoundError,
     TransportError,
 )
 from elasticsearch.transport import Transport
@@ -748,7 +749,7 @@ def test_multiple_requests_verify_elasticsearch_success():
     assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
 
 
-def test_multiple_requests_verify_elasticsearch_errors():
+def test_multiple_requests_verify_elasticsearch_product_error():
     try:
         import threading
     except ImportError:
@@ -810,3 +811,61 @@ def test_multiple_requests_verify_elasticsearch_errors():
 
     # The rest of the requests are 'GET /_search' afterwards
     assert all(call[0][:2] == ("GET", "/_search") for call in calls[1:])
+
+
+@pytest.mark.parametrize("error_cls", [ConnectionError, NotFoundError])
+def test_multiple_requests_verify_elasticsearch_retry_on_errors(error_cls):
+    try:
+        import threading
+    except ImportError:
+        return pytest.skip("Requires the 'threading' module")
+
+    t = Transport(
+        [
+            {
+                "exception": error_cls(),
+                "delay": 0.1,
+            }
+        ],
+        connection_class=DummyConnection,
+    )
+
+    results = []
+    completed_at = []
+
+    class RequestThread(threading.Thread):
+        def run(self):
+            try:
+                results.append(t.perform_request("GET", "/_search"))
+            except Exception as e:
+                results.append(e)
+            completed_at.append(time.time())
+
+    # Execute a bunch of requests concurrently.
+    threads = []
+    start_time = time.time()
+    for _ in range(5):
+        thread = RequestThread()
+        thread.start()
+        threads.append(thread)
+    for thread in threads:
+        thread.join()
+    end_time = time.time()
+
+    # Exactly 5 results completed
+    assert len(results) == 5
+
+    # All results were errors and not wrapped in 'NotElasticsearchError'
+    assert all(isinstance(result, error_cls) for result in results)
+
+    # Assert that 5 requests were made in total (5 transport requests per x 0.1s/conn request)
+    duration = end_time - start_time
+    assert 0.5 <= duration <= 0.6
+
+    # Assert that the cluster is still in the unknown/unverified stage.
+    assert t._verified_elasticsearch is None
+
+    # See that the API isn't hit, instead it's the index requests that are failing.
+    calls = t.connection_pool.connections[0].calls
+    assert len(calls) == 5
+    assert all(call[0] == ("GET", "/") for call in calls)