[7.x] Move client meta header logic to Transport

Co-authored-by: Seth Michael Larson <[email protected]>

elasticsearch/_async/helpers.py

@@ -367,7 +367,11 @@ async def async_scan(
 
     finally:
         if scroll_id and clear_scroll:
-            await client.clear_scroll(body={"scroll_id": [scroll_id]}, ignore=(404,))
+            await client.clear_scroll(
+                body={"scroll_id": [scroll_id]},
+                ignore=(404,),
+                params={"__elastic_client_meta": (("h", "s"),)},
+            )
 
 
 async def async_reindex(

elasticsearch/_async/http_aiohttp.py

@@ -24,8 +24,6 @@ from ._extra_imports import aiohttp_exceptions, aiohttp, yarl
 from .compat import get_running_loop
 from ..connection.base import (
     Connection,
-    _get_client_meta_header,
-    _python_to_meta_version,
 )
 from ..compat import urlencode
 from ..exceptions import (
@@ -34,6 +32,7 @@ from ..exceptions import (
     ImproperlyConfigured,
     SSLError,
 )
+from ..utils import _client_meta_version
 
 
 # sentinel value for `verify_certs`.
@@ -72,6 +71,9 @@ class AsyncConnection(Connection):
 
 
 class AIOHttpConnection(AsyncConnection):
+
+    HTTP_CLIENT_META = ("ai", _client_meta_version(aiohttp.__version__))
+
     def __init__(
         self,
         host="localhost",
@@ -222,11 +224,6 @@ 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:
@@ -277,13 +274,6 @@ 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/transport.py

@@ -286,8 +286,8 @@ class AsyncTransport(Transport):
         """
         await self._async_call()
 
-        method, params, body, ignore, timeout = self._resolve_request_args(
+        method, headers, params, body, ignore, timeout = self._resolve_request_args(
-            method, params, body
+            method, headers, params, body
         )
 
         for attempt in range(self.max_retries + 1):

elasticsearch/connection/base.py

@@ -65,10 +65,10 @@ 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.
     """
 
+    HTTP_CLIENT_META = None
+
     def __init__(
         self,
         host="localhost",
@@ -336,27 +336,3 @@ 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][0-9.]*[0-9]|[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_to_meta_version(python_version())
-    # 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/http_requests.py

@@ -18,14 +18,7 @@
 import time
 import warnings
 
-try:
+from .base import Connection
-    import requests
-
-    REQUESTS_AVAILABLE = True
-except ImportError:
-    REQUESTS_AVAILABLE = False
-
-from .base import Connection, _get_client_meta_header, _python_to_meta_version
 from ..exceptions import (
     ConnectionError,
     ImproperlyConfigured,
@@ -33,6 +26,16 @@ from ..exceptions import (
     SSLError,
 )
 from ..compat import urlencode, string_types
+from ..utils import _client_meta_version
+
+try:
+    import requests
+
+    REQUESTS_AVAILABLE = True
+    _REQUESTS_META_VERSION = _client_meta_version(requests.__version__)
+except ImportError:
+    REQUESTS_AVAILABLE = False
+    _REQUESTS_META_VERSION = ""
 
 
 class RequestsHttpConnection(Connection):
@@ -59,6 +62,8 @@ class RequestsHttpConnection(Connection):
         For tracing all requests made by this transport.
     """
 
+    HTTP_CLIENT_META = ("rq", _REQUESTS_META_VERSION)
+
     def __init__(
         self,
         host="localhost",
@@ -141,11 +146,6 @@ class RequestsHttpConnection(Connection):
     ):
         url = self.base_url + url
         headers = headers or {}
-        if params:
-            # Pop client metadata from parameters, if any.
-            client_meta = params.pop("_client_meta", ())
-        else:
-            client_meta = ()
         if params:
             url = "%s?%s" % (url, urlencode(params))
 
@@ -154,13 +154,6 @@ class RequestsHttpConnection(Connection):
             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_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, _get_client_meta_header, _python_to_meta_version
+from .base import Connection
 from ..exceptions import (
     ConnectionError,
     ImproperlyConfigured,
@@ -30,6 +30,7 @@ from ..exceptions import (
     SSLError,
 )
 from ..compat import urlencode
+from ..utils import _client_meta_version
 
 # sentinel value for `verify_certs` and `ssl_show_warn`.
 # This is used to detect if a user is passing in a value
@@ -96,6 +97,8 @@ class Urllib3HttpConnection(Connection):
         For tracing all requests made by this transport.
     """
 
+    HTTP_CLIENT_META = ("ur", _client_meta_version(urllib3.__version__))
+
     def __init__(
         self,
         host="localhost",
@@ -216,11 +219,6 @@ 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))
 
@@ -248,15 +246,6 @@ 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/helpers/actions.py

@@ -252,7 +252,7 @@ def _process_bulk_chunk(
 
 def _add_helper_meta_to_kwargs(kwargs, helper_meta):
     params = (kwargs or {}).pop("params", {})
-    params["_client_meta"] = (("h", helper_meta),)
+    params["__elastic_client_meta"] = (("h", helper_meta),)
     kwargs["params"] = params
     return kwargs
 
@@ -575,7 +575,7 @@ def scan(
             client.clear_scroll(
                 body={"scroll_id": [scroll_id]},
                 ignore=(404,),
-                params={"_client_meta": (("h", "s"),)},
+                params={"__elastic_client_meta": (("h", "s"),)},
             )
 
 

elasticsearch/transport.py

@@ -16,8 +16,10 @@
 #  under the License.
 
 import time
+from platform import python_version
 from itertools import chain
 
+from ._version import __versionstr__
 from .connection import Urllib3HttpConnection
 from .connection_pool import ConnectionPool, DummyConnectionPool, EmptyConnectionPool
 from .serializer import JSONSerializer, Deserializer, DEFAULT_SERIALIZERS
@@ -27,6 +29,7 @@ from .exceptions import (
     SerializationError,
     ConnectionTimeout,
 )
+from .utils import _client_meta_version
 
 
 def get_host_info(node_info, host):
@@ -76,6 +79,7 @@ class Transport(object):
         retry_on_status=(502, 503, 504),
         retry_on_timeout=False,
         send_get_body_as="GET",
+        meta_header=True,
         **kwargs
     ):
         """
@@ -110,6 +114,8 @@ class Transport(object):
             don't support passing bodies with GET requests. If you set this to
             'POST' a POST method will be used instead, if to 'source' then the body
             will be serialized and passed as a query parameter `source`.
+        :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.
 
         Any extra keyword arguments will be passed to the `connection_class`
         when creating and instance unless overridden by that connection's
@@ -117,6 +123,8 @@ class Transport(object):
         """
         if connection_class is None:
             connection_class = self.DEFAULT_CONNECTION_CLASS
+        if not isinstance(meta_header, bool):
+            raise TypeError("meta_header must be of type bool")
 
         # serialization config
         _serializers = DEFAULT_SERIALIZERS.copy()
@@ -132,6 +140,7 @@ class Transport(object):
         self.retry_on_timeout = retry_on_timeout
         self.retry_on_status = retry_on_status
         self.send_get_body_as = send_get_body_as
+        self.meta_header = meta_header
 
         # data serializer
         self.serializer = serializer
@@ -175,6 +184,20 @@ class Transport(object):
         if sniff_on_start:
             self.sniff_hosts(True)
 
+        # Create the default metadata for the x-elastic-client-meta
+        # HTTP header. Only requires adding the (service, service_version)
+        # tuple to the beginning of the client_meta
+        self._client_meta = (
+            ("es", _client_meta_version(__versionstr__)),
+            ("py", _client_meta_version(python_version())),
+            ("t", _client_meta_version(__versionstr__)),
+        )
+
+        # Grab the 'HTTP_CLIENT_META' property from the connection class
+        http_client_meta = getattr(connection_class, "HTTP_CLIENT_META", None)
+        if http_client_meta:
+            self._client_meta += (http_client_meta,)
+
     def add_connection(self, host):
         """
         Create a new :class:`~elasticsearch.Connection` instance and add it to the pool.
@@ -347,8 +370,8 @@ class Transport(object):
         :arg body: body of the request, will be serialized using serializer and
             passed to the connection
         """
-        method, params, body, ignore, timeout = self._resolve_request_args(
+        method, headers, params, body, ignore, timeout = self._resolve_request_args(
-            method, params, body
+            method, headers, params, body
         )
 
         for attempt in range(self.max_retries + 1):
@@ -410,7 +433,7 @@ class Transport(object):
         """
         self.connection_pool.close()
 
-    def _resolve_request_args(self, method, params, body):
+    def _resolve_request_args(self, method, headers, params, body):
         """Resolves parameters for .perform_request()"""
         if body is not None:
             body = self.serializer.dumps(body)
@@ -442,5 +465,15 @@ class Transport(object):
             ignore = params.pop("ignore", ())
             if isinstance(ignore, int):
                 ignore = (ignore,)
+            client_meta = params.pop("__elastic_client_meta", ())
+        else:
+            client_meta = ()
 
-        return method, params, body, ignore, timeout
+        if self.meta_header:
+            headers = headers or {}
+            client_meta = self._client_meta + client_meta
+            headers["x-elastic-client-meta"] = ",".join(
+                "%s=%s" % (k, v) for k, v in client_meta
+            )
+
+        return method, headers, params, body, ignore, timeout

elasticsearch/transport.pyi

@@ -77,6 +77,7 @@ class Transport(object):
         retry_on_status: Collection[int] = ...,
         retry_on_timeout: bool = ...,
         send_get_body_as: str = ...,
+        meta_header: bool = ...,
         **kwargs: Any
     ) -> None: ...
     def add_connection(self, host: Any) -> None: ...

elasticsearch/utils.py

@@ -0,0 +1,31 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+import re
+
+
+def _client_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][0-9.]*[0-9]|[0-9])(.*)$", version
+    ).groups()
+    if version_pre:
+        version += "p"
+    return version

elasticsearch/utils.pyi

@@ -0,0 +1,18 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+def _client_meta_version(version: str) -> str: ...

test_elasticsearch/test_async/test_connection.py

@@ -17,7 +17,6 @@
 #  under the License.
 
 import ssl
-import re
 import gzip
 import io
 from mock import patch
@@ -317,43 +316,3 @@ 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_async/test_transport.py

@@ -17,6 +17,7 @@
 #  under the License.
 
 from __future__ import unicode_literals
+import re
 import asyncio
 import json
 from mock import patch
@@ -118,7 +119,7 @@ class TestTransport:
         assert isinstance(t.connection_pool, DummyConnectionPool)
 
     async def test_request_timeout_extracted_from_params_and_passed(self):
-        t = AsyncTransport([{}], connection_class=DummyConnection)
+        t = AsyncTransport([{}], connection_class=DummyConnection, meta_header=False)
 
         await t.perform_request("GET", "/", params={"request_timeout": 42})
         assert 1 == len(t.get_connection().calls)
@@ -130,7 +131,9 @@ class TestTransport:
         } == t.get_connection().calls[0][1]
 
     async def test_opaque_id(self):
-        t = AsyncTransport([{}], opaque_id="app-1", connection_class=DummyConnection)
+        t = AsyncTransport(
+            [{}], opaque_id="app-1", connection_class=DummyConnection, meta_header=False
+        )
 
         await t.perform_request("GET", "/")
         assert 1 == len(t.get_connection().calls)
@@ -152,7 +155,7 @@ class TestTransport:
         } == t.get_connection().calls[1][1]
 
     async def test_request_with_custom_user_agent_header(self):
-        t = AsyncTransport([{}], connection_class=DummyConnection)
+        t = AsyncTransport([{}], connection_class=DummyConnection, meta_header=False)
 
         await t.perform_request(
             "GET", "/", headers={"user-agent": "my-custom-value/1.2.3"}
@@ -182,6 +185,39 @@ class TestTransport:
         assert 1 == len(t.get_connection().calls)
         assert ("POST", "/", None, b"{}") == t.get_connection().calls[0][0]
 
+    async def test_client_meta_header(self):
+        t = AsyncTransport([{}], connection_class=DummyConnection)
+
+        await t.perform_request("GET", "/", body={})
+        assert len(t.get_connection().calls) == 1
+        headers = t.get_connection().calls[0][1]["headers"]
+        assert re.match(
+            r"^es=[0-9.]+p?,py=[0-9.]+p?,t=[0-9.]+p?$",
+            headers["x-elastic-client-meta"],
+        )
+
+        class DummyConnectionWithMeta(DummyConnection):
+            HTTP_CLIENT_META = ("dm", "1.2.3")
+
+        t = AsyncTransport([{}], connection_class=DummyConnectionWithMeta)
+
+        await t.perform_request("GET", "/", body={}, headers={"Custom": "header"})
+        assert len(t.get_connection().calls) == 1
+        headers = t.get_connection().calls[0][1]["headers"]
+        assert re.match(
+            r"^es=[0-9.]+p?,py=[0-9.]+p?,t=[0-9.]+p?,dm=1.2.3$",
+            headers["x-elastic-client-meta"],
+        )
+        assert headers["Custom"] == "header"
+
+    async def test_client_meta_header_not_sent(self):
+        t = AsyncTransport([{}], meta_header=False, connection_class=DummyConnection)
+
+        await t.perform_request("GET", "/", body={})
+        assert len(t.get_connection().calls) == 1
+        headers = t.get_connection().calls[0][1]["headers"]
+        assert headers is None
+
     async def test_body_gets_encoded_into_bytes(self):
         t = AsyncTransport([{}], connection_class=DummyConnection)
 

test_elasticsearch/test_connection.py

@@ -40,12 +40,8 @@ from elasticsearch.connection import (
     RequestsHttpConnection,
     Urllib3HttpConnection,
 )
-from elasticsearch.connection.base import (
-    _get_client_meta_header,
-    _python_to_meta_version,
-)
 
-from elasticsearch import __version__, __versionstr__
+from elasticsearch import __versionstr__
 from .test_cases import TestCase, SkipTest
 
 
@@ -176,23 +172,6 @@ class TestBaseConnection(TestCase):
             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_to_meta_version(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"
-        assert _python_to_meta_version("7.10.a1") == "7.10p"
-
 
 class TestUrllib3Connection(TestCase):
     def _get_mock_connection(self, connection_params={}, response_body=b"{}"):
@@ -466,45 +445,6 @@ 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(
@@ -906,42 +846,3 @@ 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

@@ -208,15 +208,15 @@ class TestChunkActions(TestCase):
     def test_add_helper_meta_to_kwargs(self):
         self.assertEqual(
             actions._add_helper_meta_to_kwargs({}, "b"),
-            {"params": {"_client_meta": (("h", "b"),)}},
+            {"params": {"__elastic_client_meta": (("h", "b"),)}},
         )
         self.assertEqual(
             actions._add_helper_meta_to_kwargs({"params": {}}, "b"),
-            {"params": {"_client_meta": (("h", "b"),)}},
+            {"params": {"__elastic_client_meta": (("h", "b"),)}},
         )
         self.assertEqual(
             actions._add_helper_meta_to_kwargs({"params": {"key": "value"}}, "b"),
-            {"params": {"_client_meta": (("h", "b"),), "key": "value"}},
+            {"params": {"__elastic_client_meta": (("h", "b"),), "key": "value"}},
         )
 
 

test_elasticsearch/test_transport.py

@@ -19,6 +19,7 @@
 from __future__ import unicode_literals
 import json
 import time
+import pytest
 from mock import patch
 
 from elasticsearch.transport import Transport, get_host_info
@@ -120,7 +121,7 @@ class TestTransport(TestCase):
         self.assertIsInstance(t.connection_pool, DummyConnectionPool)
 
     def test_request_timeout_extracted_from_params_and_passed(self):
-        t = Transport([{}], connection_class=DummyConnection)
+        t = Transport([{}], meta_header=False, connection_class=DummyConnection)
 
         t.perform_request("GET", "/", params={"request_timeout": 42})
         self.assertEqual(1, len(t.get_connection().calls))
@@ -131,7 +132,9 @@ class TestTransport(TestCase):
         )
 
     def test_opaque_id(self):
-        t = Transport([{}], opaque_id="app-1", connection_class=DummyConnection)
+        t = Transport(
+            [{}], opaque_id="app-1", meta_header=False, connection_class=DummyConnection
+        )
 
         t.perform_request("GET", "/")
         self.assertEqual(1, len(t.get_connection().calls))
@@ -151,7 +154,7 @@ class TestTransport(TestCase):
         )
 
     def test_request_with_custom_user_agent_header(self):
-        t = Transport([{}], connection_class=DummyConnection)
+        t = Transport([{}], meta_header=False, connection_class=DummyConnection)
 
         t.perform_request("GET", "/", headers={"user-agent": "my-custom-value/1.2.3"})
         self.assertEqual(1, len(t.get_connection().calls))
@@ -180,6 +183,43 @@ class TestTransport(TestCase):
         self.assertEqual(1, len(t.get_connection().calls))
         self.assertEqual(("POST", "/", None, b"{}"), t.get_connection().calls[0][0])
 
+    def test_client_meta_header(self):
+        t = Transport([{}], connection_class=DummyConnection)
+
+        t.perform_request("GET", "/", body={})
+        self.assertEqual(1, len(t.get_connection().calls))
+        headers = t.get_connection().calls[0][1]["headers"]
+        self.assertRegexpMatches(
+            headers["x-elastic-client-meta"], r"^es=[0-9.]+p?,py=[0-9.]+p?,t=[0-9.]+p?$"
+        )
+
+        class DummyConnectionWithMeta(DummyConnection):
+            HTTP_CLIENT_META = ("dm", "1.2.3")
+
+        t = Transport([{}], connection_class=DummyConnectionWithMeta)
+
+        t.perform_request("GET", "/", body={}, headers={"Custom": "header"})
+        self.assertEqual(1, len(t.get_connection().calls))
+        headers = t.get_connection().calls[0][1]["headers"]
+        self.assertRegexpMatches(
+            headers["x-elastic-client-meta"],
+            r"^es=[0-9.]+p?,py=[0-9.]+p?,t=[0-9.]+p?,dm=1.2.3$",
+        )
+        self.assertEqual(headers["Custom"], "header")
+
+    def test_client_meta_header_not_sent(self):
+        t = Transport([{}], meta_header=False, connection_class=DummyConnection)
+
+        t.perform_request("GET", "/", body={})
+        self.assertEqual(1, len(t.get_connection().calls))
+        headers = t.get_connection().calls[0][1]["headers"]
+        self.assertIs(headers, None)
+
+    def test_meta_header_type_error(self):
+        with pytest.raises(TypeError) as e:
+            Transport([{}], meta_header=1)
+        assert str(e.value) == "meta_header must be of type bool"
+
     def test_body_gets_encoded_into_bytes(self):
         t = Transport([{}], connection_class=DummyConnection)
 

test_elasticsearch/test_utils.py

@@ -0,0 +1,27 @@
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.
+
+import pytest
+from elasticsearch.utils import _client_meta_version
+
+
+@pytest.mark.parametrize(
+    ["version", "meta_version"],
+    [("1.26.3", "1.26.3"), ("7.10.1a1", "7.10.1p"), ("7.10.pre", "7.10p")],
+)
+def test_client_meta_version(version, meta_version):
+    assert _client_meta_version(version) == meta_version