Merge removal of product verification with later 7.x changes

Author: Tim Bannister

elasticsearch/_async/transport.py

@@ -18,19 +18,16 @@
 import asyncio
 import logging
 import sys
-import warnings
 from itertools import chain
 
 from ..exceptions import (
- AuthenticationException,
- AuthorizationException,
  ConnectionError,
  ConnectionTimeout,
  ElasticsearchWarning,
  SerializationError,
  TransportError,
 )
-from ..transport import Transport, _ProductChecker
+from ..transport import Transport
 from .compat import get_running_loop
 from .http_aiohttp import AIOHttpConnection
 
@@ -117,10 +114,6 @@ async def _async_init(self):
  self.loop = get_running_loop()
  self.kwargs["loop"] = self.loop
 
- # Set our 'verified_once' implementation to one that
- # works with 'asyncio' instead of 'threading'
- self._verify_elasticsearch_lock = asyncio.Lock()
-
  # Now that we have a loop we can create all our HTTP connections...
  self.set_connections(self.hosts)
  self.seed_connections = list(self.connection_pool.connections[:])
@@ -335,14 +328,6 @@ async def perform_request(self, method, url, headers=None, params=None, body=Non
  method, headers, params, body
  )
 
- # Before we make the actual API call we verify the Elasticsearch instance.
- if self._verified_elasticsearch is None:
- await self._do_verify_elasticsearch(headers=headers, timeout=timeout)
-
- # If '_verified_elasticsearch' isn't 'True' then we raise an error.
- if self._verified_elasticsearch is not True:
- _ProductChecker.raise_error(self._verified_elasticsearch)
-
  for attempt in range(self.max_retries + 1):
  connection = self.get_connection()
 
@@ -414,84 +399,3 @@ async def close(self):
 
  for connection in self.connection_pool.connections:
  await connection.close()
-
- async def _do_verify_elasticsearch(self, headers, timeout):
- """Verifies that we're connected to an Elasticsearch cluster.
- This is done at least once before the first actual API call
- and makes a single request to the 'GET /' API endpoint and
- check version along with other details of the response.
-
- If we're unable to verify we're talking to Elasticsearch
- but we're also unable to rule it out due to a permission
- error we instead emit an 'ElasticsearchWarning'.
- """
- # Ensure that there's only one async exec within this section
- # at a time to not emit unnecessary index API calls.
- async with self._verify_elasticsearch_lock:
-
- # Product check has already been completed while we were
- # waiting our turn, no need to do again.
- if self._verified_elasticsearch is not None:
- return
-
- headers = {
- header.lower(): value for header, value in (headers or {}).items()
- }
- # We know we definitely want JSON so request it via 'accept'
- headers.setdefault("accept", "application/json")
-
- info_headers = {}
- info_response = {}
- error = None
-
- attempted_conns = []
- for conn in chain(self.connection_pool.connections, self.seed_connections):
- # 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
- )
-
- # 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.
- 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 = _ProductChecker.check_product(
- info_headers, info_response
- )

elasticsearch/compat.py

@@ -70,17 +70,6 @@ def to_bytes(x, encoding="ascii"):
 except (ImportError, AttributeError):
  pass
 
-try:
- from threading import Lock
-except ImportError: # Python <3.7 isn't guaranteed to have threading support.
-
- class Lock:
- def __enter__(self):
- pass
-
- def __exit__(self, *_):
- pass
-
 
 __all__ = [
  "string_types",

elasticsearch/exceptions.py

@@ -56,7 +56,6 @@ class UnsupportedProductError(ElasticsearchException):
  it's not connected to a supported product.
  """
 
-
 class TransportError(ElasticsearchException):
  """
  Exception raised when ES returns a non-OK (>=400) HTTP status code. Or when

elasticsearch/transport.py

@@ -15,19 +15,14 @@
 # specific language governing permissions and limitations
 # under the License.
 
-import re
 import time
-import warnings
 from itertools import chain
 from platform import python_version
 
 from ._version import __versionstr__
-from .compat import Lock
 from .connection import Urllib3HttpConnection
 from .connection_pool import ConnectionPool, DummyConnectionPool, EmptyConnectionPool
 from .exceptions import (
- AuthenticationException,
- AuthorizationException,
  ConnectionError,
  ConnectionTimeout,
  ElasticsearchWarning,
@@ -404,14 +399,6 @@ def perform_request(self, method, url, headers=None, params=None, body=None):
  method, headers, params, body
  )
 
- # Before we make the actual API call we verify the Elasticsearch instance.
- if self._verified_elasticsearch is None:
- self._do_verify_elasticsearch(headers=headers, timeout=timeout)
-
- # If '_verified_elasticsearch' isn't 'True' then we raise an error.
- if self._verified_elasticsearch is not True:
- _ProductChecker.raise_error(self._verified_elasticsearch)
-
  for attempt in range(self.max_retries + 1):
  connection = self.get_connection()
 
@@ -521,88 +508,6 @@ def _resolve_request_args(self, method, headers, params, body):
 
  return method, headers, params, body, ignore, timeout
 
- def _do_verify_elasticsearch(self, headers, timeout):
- """Verifies that we're connected to an Elasticsearch cluster.
- This is done at least once before the first actual API call
- and makes a single request to the 'GET /' API endpoint to
- check the version along with other details of the response.
-
- If we're unable to verify we're talking to Elasticsearch
- but we're also unable to rule it out due to a permission
- error we instead emit an 'ElasticsearchWarning'.
- """
- # Ensure that there's only one thread within this section
- # at a time to not emit unnecessary index API calls.
- with self._verify_elasticsearch_lock:
-
- # Product check has already been completed while we were
- # waiting our turn, no need to do again.
- if self._verified_elasticsearch is not None:
- return
-
- headers = {
- header.lower(): value for header, value in (headers or {}).items()
- }
- # We know we definitely want JSON so request it via 'accept'
- headers.setdefault("accept", "application/json")
-
- info_headers = {}
- info_response = {}
- error = None
-
- attempted_conns = []
- for conn in chain(self.connection_pool.connections, self.seed_connections):
- # 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 = _ProductChecker.check_product(
- info_headers, info_response
- )
-
-
 class _ProductChecker:
  """Class which verifies we're connected to a supported product"""
 
@@ -624,17 +529,17 @@ def raise_error(cls, state):
  )
  else: # UNSUPPORTED_PRODUCT
  message = (
- "The client noticed that the server is not Elasticsearch "
- "and we do not support this unknown product"
+ "The client noticed that the server is not Elasticsearch"
  )
  raise UnsupportedProductError(message)
 
  @classmethod
  def check_product(cls, headers, response):
  # type: (dict[str, str], dict[str, str]) -> int
- """Verifies that the server we're talking to is Elasticsearch.
+ """Checks whether the server we're talking to is Elasticsearch.
  Does this by checking HTTP headers and the deserialized
- response to the 'info' API. Returns one of the states above.
+ response to the 'info' API. Returns true - all servers are
+ acceptable.
  """
  try:
  version = response.get("version", {})
@@ -661,21 +566,4 @@ def check_product(cls, headers, response):
  bad_build_flavor = True
  bad_product_header = True
 
- # 7.0-7.13 and there's a bad 'tagline' or unsupported 'build_flavor'
- if (7, 0, 0) <= version_number < (7, 14, 0):
- if bad_tagline:
- return cls.UNSUPPORTED_PRODUCT
- elif bad_build_flavor:
- return cls.UNSUPPORTED_DISTRIBUTION
-
- elif (
- # No version or version less than 6.x
- version_number < (6, 0, 0)
- # 6.x and there's a bad 'tagline'
- or ((6, 0, 0) <= version_number < (7, 0, 0) and bad_tagline)
- # 7.14+ and there's a bad 'X-Elastic-Product' HTTP header
- or ((7, 14, 0) <= version_number and bad_product_header)
- ):
- return cls.UNSUPPORTED_PRODUCT
-
  return True

test_elasticsearch/test_async/test_connection.py

@@ -28,7 +28,7 @@
 from mock import patch
 from multidict import CIMultiDict
 
-from elasticsearch import AIOHttpConnection, AsyncElasticsearch, __versionstr__
+from elasticsearch import AIOHttpConnection, __versionstr__
 from elasticsearch.compat import reraise_exceptions
 from elasticsearch.exceptions import ConnectionError
 
@@ -410,9 +410,3 @@ async def test_aiohttp_connection_error(self):
  conn = AIOHttpConnection("not.a.host.name")
  with pytest.raises(ConnectionError):
  await conn.perform_request("GET", "/")
-
- async def test_elasticsearch_connection_error(self):
- es = AsyncElasticsearch("http://not.a.host.name")
-
- with pytest.raises(ConnectionError):
- await es.search()