From b894e359df7dc6cb38c983d74bf51ea1276a4bf8 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Mon, 14 Dec 2020 17:50:14 -0600 Subject: [PATCH] [7.x] Add the 'X-Elastic-Client-Meta' header Co-authored-by: Seth Michael Larson --- elasticsearch/_async/http_aiohttp.py | 18 ++- elasticsearch/_async/http_aiohttp.pyi | 7 +- elasticsearch/connection/base.py | 30 +++++ elasticsearch/connection/base.pyi | 8 +- elasticsearch/connection/http_requests.py | 16 ++- elasticsearch/connection/http_requests.pyi | 1 + elasticsearch/connection/http_urllib3.py | 17 ++- elasticsearch/connection/http_urllib3.pyi | 1 + elasticsearch/helpers/actions.py | 16 ++- .../test_async/test_connection.py | 41 +++++++ test_elasticsearch/test_connection.py | 111 +++++++++++++++++- test_elasticsearch/test_helpers.py | 15 +++ 12 files changed, 269 insertions(+), 12 deletions(-) diff --git a/elasticsearch/_async/http_aiohttp.py b/elasticsearch/_async/http_aiohttp.py index 0deb7ee5..3fb1e91c 100644 --- a/elasticsearch/_async/http_aiohttp.py +++ b/elasticsearch/_async/http_aiohttp.py @@ -22,7 +22,11 @@ import urllib3 # type: ignore import warnings from ._extra_imports import aiohttp_exceptions, aiohttp, yarl from .compat import get_running_loop -from ..connection import Connection +from ..connection.base import ( + Connection, + _get_client_meta_header, + _python_to_meta_version, +) from ..compat import urlencode from ..exceptions import ( ConnectionError, @@ -218,6 +222,11 @@ class AIOHttpConnection(AsyncConnection): orig_body = body url_path = self.url_prefix + url + if params: + # Pop client metadata from parameters, if any. + client_meta = tuple(params.pop("_client_meta", ())) + else: + client_meta = () if params: query_string = urlencode(params) else: @@ -268,6 +277,13 @@ class AIOHttpConnection(AsyncConnection): body = self._gzip_compress(body) req_headers["content-encoding"] = "gzip" + # Create meta header for aiohttp + if self.meta_header: + client_meta = ( + ("ai", _python_to_meta_version(aiohttp.__version__)), + ) + client_meta + req_headers["x-elastic-client-meta"] = _get_client_meta_header(client_meta) + start = self.loop.time() try: async with self.session.request( diff --git a/elasticsearch/_async/http_aiohttp.pyi b/elasticsearch/_async/http_aiohttp.pyi index 971f98ce..b46d48b0 100644 --- a/elasticsearch/_async/http_aiohttp.pyi +++ b/elasticsearch/_async/http_aiohttp.pyi @@ -16,7 +16,7 @@ # under the License. from ._extra_imports import aiohttp # type: ignore -from typing import Optional, Mapping, Collection, Union, Any, Tuple +from typing import Optional, Mapping, MutableMapping, Collection, Union, Any, Tuple from ..connection import Connection class AsyncConnection(Connection): @@ -24,11 +24,11 @@ class AsyncConnection(Connection): self, method: str, url: str, - params: Optional[Mapping[str, Any]] = ..., + params: Optional[MutableMapping[str, Any]] = ..., body: Optional[bytes] = ..., timeout: Optional[Union[int, float]] = ..., ignore: Collection[int] = ..., - headers: Optional[Mapping[str, str]] = ..., + headers: Optional[MutableMapping[str, str]] = ..., ) -> Tuple[int, Mapping[str, str], str]: ... async def close(self) -> None: ... @@ -55,6 +55,7 @@ class AIOHttpConnection(AsyncConnection): cloud_id: Optional[str] = ..., api_key: Optional[Any] = ..., opaque_id: Optional[str] = ..., + meta_header: bool = ..., loop: Any = ..., **kwargs: Any, ) -> None: ... diff --git a/elasticsearch/connection/base.py b/elasticsearch/connection/base.py index 30f26530..f7a98040 100644 --- a/elasticsearch/connection/base.py +++ b/elasticsearch/connection/base.py @@ -21,6 +21,7 @@ import gzip import io import re from platform import python_version +import sys import warnings try: @@ -65,6 +66,8 @@ class Connection(object): :arg cloud_id: The Cloud ID from ElasticCloud. Convenient way to connect to cloud instances. :arg opaque_id: Send this value in the 'X-Opaque-Id' HTTP header For tracing all requests made by this transport. + :arg meta_header: If True will send the 'X-Elastic-Client-Meta' HTTP header containing + simple client metadata. Setting to False will disable the header. Defaults to True. """ def __init__( @@ -79,6 +82,7 @@ class Connection(object): cloud_id=None, api_key=None, opaque_id=None, + meta_header=True, **kwargs ): @@ -148,6 +152,10 @@ class Connection(object): self.url_prefix = url_prefix self.timeout = timeout + if not isinstance(meta_header, bool): + raise TypeError("meta_header must be of type bool") + self.meta_header = meta_header + def __repr__(self): return "<%s: %s>" % (self.__class__.__name__, self.host) @@ -329,3 +337,25 @@ class Connection(object): s = "{0}:{1}".format(api_key[0], api_key[1]).encode("utf-8") return "ApiKey " + binascii.b2a_base64(s).rstrip(b"\r\n").decode("utf-8") return "ApiKey " + api_key + + +def _python_to_meta_version(version): + """Transforms a Python package version to one + compatible with 'X-Elastic-Client-Meta'. Essentially + replaces any pre-release information with a 'p' suffix. + """ + version, version_pre = re.match(r"^([0-9.]+)(.*)$", version).groups() + if version_pre: + version += "p" + return version + + +def _get_client_meta_header(client_meta=()): + """Builds an 'X-Elastic-Client-Meta' HTTP header""" + es_version = _python_to_meta_version(__versionstr__) + py_version = python_version() + ("p" if sys.version_info[3] != "final" else "") + # First three values have to be 'service', 'language', 'transport' + client_meta = (("es", es_version), ("py", py_version), ("t", es_version)) + tuple( + client_meta + ) + return ",".join("%s=%s" % (k, v) for k, v in client_meta) diff --git a/elasticsearch/connection/base.pyi b/elasticsearch/connection/base.pyi index 82d266bb..d2eca4ed 100644 --- a/elasticsearch/connection/base.pyi +++ b/elasticsearch/connection/base.pyi @@ -21,13 +21,13 @@ from typing import ( Union, Optional, Mapping, + MutableMapping, Tuple, List, NoReturn, Dict, Sequence, Any, - AnyStr, Collection, ) @@ -44,6 +44,7 @@ class Connection(object): host: str url_prefix: str timeout: Optional[Union[float, int]] + meta_header: bool def __init__( self, host: str = ..., @@ -56,6 +57,7 @@ class Connection(object): cloud_id: Optional[str] = ..., api_key: Optional[Union[Tuple[str, str], List[str], str]] = ..., opaque_id: Optional[str] = ..., + meta_header: bool = ..., **kwargs: Any ) -> None: ... def __repr__(self) -> str: ... @@ -77,11 +79,11 @@ class Connection(object): self, method: str, url: str, - params: Optional[Mapping[str, Any]] = ..., + params: Optional[MutableMapping[str, Any]] = ..., body: Optional[bytes] = ..., timeout: Optional[Union[int, float]] = ..., ignore: Collection[int] = ..., - headers: Optional[Mapping[str, str]] = ..., + headers: Optional[MutableMapping[str, str]] = ..., ) -> Tuple[int, Mapping[str, str], str]: ... def log_request_success( self, diff --git a/elasticsearch/connection/http_requests.py b/elasticsearch/connection/http_requests.py index bd4e3835..05011384 100644 --- a/elasticsearch/connection/http_requests.py +++ b/elasticsearch/connection/http_requests.py @@ -25,7 +25,7 @@ try: except ImportError: REQUESTS_AVAILABLE = False -from .base import Connection +from .base import Connection, _get_client_meta_header, _python_to_meta_version from ..exceptions import ( ConnectionError, ImproperlyConfigured, @@ -142,13 +142,25 @@ class RequestsHttpConnection(Connection): url = self.base_url + url headers = headers or {} if params: - url = "%s?%s" % (url, urlencode(params or {})) + # Pop client metadata from parameters, if any. + client_meta = params.pop("_client_meta", ()) + else: + client_meta = () + if params: + url = "%s?%s" % (url, urlencode(params)) orig_body = body if self.http_compress and body: body = self._gzip_compress(body) headers["content-encoding"] = "gzip" + # Create meta header for requests + if self.meta_header: + client_meta = ( + ("rq", _python_to_meta_version(requests.__version__)), + ) + client_meta + headers["x-elastic-client-meta"] = _get_client_meta_header(client_meta) + start = time.time() request = requests.Request(method=method, headers=headers, url=url, data=body) prepared_request = self.session.prepare_request(request) diff --git a/elasticsearch/connection/http_requests.pyi b/elasticsearch/connection/http_requests.pyi index a9964837..df86edf0 100644 --- a/elasticsearch/connection/http_requests.pyi +++ b/elasticsearch/connection/http_requests.pyi @@ -37,5 +37,6 @@ class RequestsHttpConnection(Connection): cloud_id: Optional[str] = ..., api_key: Optional[Any] = ..., opaque_id: Optional[str] = ..., + meta_header: bool = ..., **kwargs: Any ) -> None: ... diff --git a/elasticsearch/connection/http_urllib3.py b/elasticsearch/connection/http_urllib3.py index bb2a2a9b..596c7026 100644 --- a/elasticsearch/connection/http_urllib3.py +++ b/elasticsearch/connection/http_urllib3.py @@ -22,7 +22,7 @@ from urllib3.exceptions import ReadTimeoutError, SSLError as UrllibSSLError # t from urllib3.util.retry import Retry # type: ignore import warnings -from .base import Connection +from .base import Connection, _get_client_meta_header, _python_to_meta_version from ..exceptions import ( ConnectionError, ImproperlyConfigured, @@ -216,8 +216,14 @@ class Urllib3HttpConnection(Connection): self, method, url, params=None, body=None, timeout=None, ignore=(), headers=None ): url = self.url_prefix + url + # Pop client metadata from parameters, if any. + if params: + client_meta = tuple(params.pop("_client_meta", ())) + else: + client_meta = () if params: url = "%s?%s" % (url, urlencode(params)) + full_url = self.host + url start = time.time() @@ -242,6 +248,15 @@ class Urllib3HttpConnection(Connection): body = self._gzip_compress(body) request_headers["content-encoding"] = "gzip" + # Create meta header for urllib3 + if self.meta_header: + client_meta = ( + ("ur", _python_to_meta_version(urllib3.__version__)), + ) + client_meta + request_headers["x-elastic-client-meta"] = _get_client_meta_header( + client_meta + ) + response = self.pool.urlopen( method, url, body, retries=Retry(False), headers=request_headers, **kw ) diff --git a/elasticsearch/connection/http_urllib3.pyi b/elasticsearch/connection/http_urllib3.pyi index 47b126b3..26b4fc6b 100644 --- a/elasticsearch/connection/http_urllib3.pyi +++ b/elasticsearch/connection/http_urllib3.pyi @@ -51,5 +51,6 @@ class Urllib3HttpConnection(Connection): cloud_id: Optional[str] = ..., api_key: Optional[Any] = ..., opaque_id: Optional[str] = ..., + meta_header: bool = ..., **kwargs: Any ) -> None: ... diff --git a/elasticsearch/helpers/actions.py b/elasticsearch/helpers/actions.py index 94089654..1a988778 100644 --- a/elasticsearch/helpers/actions.py +++ b/elasticsearch/helpers/actions.py @@ -230,6 +230,8 @@ def _process_bulk_chunk( """ Send a bulk request to elasticsearch and process the output. """ + kwargs = _add_helper_meta_to_kwargs(kwargs, "bp") + try: # send the actual request resp = client.bulk("\n".join(bulk_actions) + "\n", *args, **kwargs) @@ -248,6 +250,13 @@ def _process_bulk_chunk( yield item +def _add_helper_meta_to_kwargs(kwargs, helper_meta): + params = (kwargs or {}).pop("params", {}) + params["_client_meta"] = (("h", helper_meta),) + kwargs["params"] = params + return kwargs + + def streaming_bulk( client, actions, @@ -515,6 +524,7 @@ def scan( """ scroll_kwargs = scroll_kwargs or {} + _add_helper_meta_to_kwargs(scroll_kwargs, "s") if not preserve_order: query = query.copy() if query else {} @@ -562,7 +572,11 @@ def scan( finally: if scroll_id and clear_scroll: - client.clear_scroll(body={"scroll_id": [scroll_id]}, ignore=(404,)) + client.clear_scroll( + body={"scroll_id": [scroll_id]}, + ignore=(404,), + params={"_client_meta": (("h", "s"),)}, + ) def reindex( diff --git a/test_elasticsearch/test_async/test_connection.py b/test_elasticsearch/test_async/test_connection.py index 77e4c6d7..fc3a9b58 100644 --- a/test_elasticsearch/test_async/test_connection.py +++ b/test_elasticsearch/test_async/test_connection.py @@ -17,6 +17,7 @@ # under the License. import ssl +import re import gzip import io from mock import patch @@ -316,3 +317,43 @@ class TestAIOHttpConnection: con = await self._get_mock_connection(response_body=buf) status, headers, data = await con.perform_request("GET", "/") assert u"你好\uda6a" == data + + async def test_meta_header_value(self): + con = await self._get_mock_connection() + assert con.meta_header is True + + await con.perform_request("GET", "/", body=b"{}") + + _, kwargs = con.session.request.call_args + headers = kwargs["headers"] + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,ai=[0-9]+\.[0-9]+\.[0-9]+p?$", + headers["x-elastic-client-meta"], + ) + + con = await self._get_mock_connection() + assert con.meta_header is True + + await con.perform_request( + "GET", "/", body=b"{}", params={"_client_meta": (("h", "bp"),)} + ) + + (method, url), kwargs = con.session.request.call_args + headers = kwargs["headers"] + assert method == "GET" + assert str(url) == "http://localhost:9200/" + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,ai=[0-9]+\.[0-9]+\.[0-9]+p?,h=bp$", + headers["x-elastic-client-meta"], + ) + + con = await self._get_mock_connection(connection_params={"meta_header": False}) + assert con.meta_header is False + + await con.perform_request("GET", "/", body=b"{}") + + _, kwargs = con.session.request.call_args + headers = kwargs["headers"] + assert "x-elastic-client-meta" not in (x.lower() for x in headers) diff --git a/test_elasticsearch/test_connection.py b/test_elasticsearch/test_connection.py index 128cbe78..9d6ee1c1 100644 --- a/test_elasticsearch/test_connection.py +++ b/test_elasticsearch/test_connection.py @@ -40,7 +40,12 @@ from elasticsearch.connection import ( RequestsHttpConnection, Urllib3HttpConnection, ) -from elasticsearch import __versionstr__ +from elasticsearch.connection.base import ( + _get_client_meta_header, + _python_to_meta_version, +) + +from elasticsearch import __version__, __versionstr__ from .test_cases import TestCase, SkipTest @@ -161,6 +166,32 @@ class TestBaseConnection(TestCase): conn = Connection(**kwargs) assert conn.host == expected_host + def test_meta_header(self): + conn = Connection(meta_header=True) + assert conn.meta_header is True + conn = Connection(meta_header=False) + assert conn.meta_header is False + + with pytest.raises(TypeError) as e: + Connection(meta_header=1) + assert str(e.value) == "meta_header must be of type bool" + + def test_get_client_meta_header(self): + meta_header = _get_client_meta_header() + assert ("es=%s" % (".".join(str(x) for x in __version__),)) in meta_header + assert ("t=%s" % (".".join(str(x) for x in __version__),)) in meta_header + assert ("py=%s" % python_version()) in meta_header + + meta_header = _get_client_meta_header((("h", "bp"),)) + assert meta_header.endswith(",h=bp") + + meta_header = _get_client_meta_header((("ur", "1.26.3"), ("h", "bp"))) + assert meta_header.endswith(",ur=1.26.3,h=bp") + + def test_python_to_meta_version(self): + assert _python_to_meta_version("1.26.3") == "1.26.3" + assert _python_to_meta_version("7.10.1a1") == "7.10.1p" + class TestUrllib3Connection(TestCase): def _get_mock_connection(self, connection_params={}, response_body=b"{}"): @@ -434,6 +465,45 @@ class TestUrllib3Connection(TestCase): status, headers, data = con.perform_request("GET", "/") self.assertEqual(u"你好\uda6a", data) + def test_meta_header_value(self): + con = self._get_mock_connection() + self.assertTrue(con.meta_header) + + con.perform_request("GET", "/", body=b"{}") + + _, kwargs = con.pool.urlopen.call_args + headers = kwargs["headers"] + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,ur=[0-9]+\.[0-9]+\.[0-9]+p?$", + headers["x-elastic-client-meta"], + ) + + con = self._get_mock_connection({"meta_header": True}) + self.assertTrue(con.meta_header) + + con.perform_request( + "GET", "/", body=b"{}", params={"_client_meta": (("h", "bp"),)} + ) + + args, kwargs = con.pool.urlopen.call_args + headers = kwargs["headers"] + assert args == ("GET", "/", b"{}") + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,ur=[0-9]+\.[0-9]+\.[0-9]+p?,h=bp$", + headers["x-elastic-client-meta"], + ) + + con = self._get_mock_connection({"meta_header": False}) + self.assertFalse(con.meta_header) + + con.perform_request("GET", "/", body=b"{}") + + _, kwargs = con.pool.urlopen.call_args + headers = kwargs["headers"] + assert "x-elastic-client-meta" not in (x.lower() for x in headers) + class TestRequestsConnection(TestCase): def _get_mock_connection( @@ -835,3 +905,42 @@ class TestRequestsConnection(TestCase): con = self._get_mock_connection(response_body=buf) status, headers, data = con.perform_request("GET", "/") self.assertEqual(u"你好\uda6a", data) + + def test_meta_header_value(self): + con = self._get_mock_connection() + self.assertTrue(con.meta_header) + + con.perform_request("GET", "/", body=b"{}") + + args, _ = con.session.send.call_args + headers = args[0].headers + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,rq=[0-9]+\.[0-9]+\.[0-9]+p?$", + headers["x-elastic-client-meta"], + ) + + con = self._get_mock_connection() + self.assertTrue(con.meta_header) + + con.perform_request( + "GET", "/", body=b"{}", params={"_client_meta": (("h", "bp"),)} + ) + + args, _ = con.session.send.call_args + headers = args[0].headers + assert args[0].url == "http://localhost:9200/" + assert re.match( + r"^es=[0-9]+\.[0-9]+\.[0-9]+p?,py=[0-9]+\.[0-9]+\.[0-9]+p?," + r"t=[0-9]+\.[0-9]+\.[0-9]+p?,rq=[0-9]+\.[0-9]+\.[0-9]+p?,h=bp$", + headers["x-elastic-client-meta"], + ) + + con = self._get_mock_connection({"meta_header": False}) + self.assertFalse(con.meta_header) + + con.perform_request("GET", "/", body=b"{}") + + args, _ = con.session.send.call_args + headers = args[0].headers + assert "x-elastic-client-meta" not in (x.lower() for x in headers) diff --git a/test_elasticsearch/test_helpers.py b/test_elasticsearch/test_helpers.py index d3640d89..becb9e88 100644 --- a/test_elasticsearch/test_helpers.py +++ b/test_elasticsearch/test_helpers.py @@ -21,6 +21,7 @@ import time import threading import pytest from elasticsearch import helpers, Elasticsearch +from elasticsearch.helpers import actions from elasticsearch.serializer import JSONSerializer from .test_cases import TestCase @@ -204,6 +205,20 @@ class TestChunkActions(TestCase): chunk = chunk if isinstance(chunk, str) else chunk.encode("utf-8") self.assertLessEqual(len(chunk), max_byte_size) + def test_add_helper_meta_to_kwargs(self): + self.assertEqual( + actions._add_helper_meta_to_kwargs({}, "b"), + {"params": {"_client_meta": (("h", "b"),)}}, + ) + self.assertEqual( + actions._add_helper_meta_to_kwargs({"params": {}}, "b"), + {"params": {"_client_meta": (("h", "b"),)}}, + ) + self.assertEqual( + actions._add_helper_meta_to_kwargs({"params": {"key": "value"}}, "b"), + {"params": {"_client_meta": (("h", "b"),), "key": "value"}}, + ) + class TestExpandActions(TestCase): def test_string_actions_are_marked_as_simple_inserts(self):