[7.x] Add the 'X-Elastic-Client-Meta' header

Co-authored-by: Seth Michael Larson <seth.larson@elastic.co>

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(

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: ...

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)

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,

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)

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: ...

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
             )

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: ...

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(

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)

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)

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):