fetch server version on transport thread start

elasticapm/base.py

@@ -102,6 +102,8 @@ def __init__(self, config=None, **inline):
         self.processors = []
         self.filter_exception_types_dict = {}
         self._service_info = None
+        # setting server_version here is mainly used for testing
+        self.server_version = inline.pop("server_version", None)
 
         self.check_python_version()
 

elasticapm/conf/constants.py

@@ -59,6 +59,7 @@ def _starmatch_to_regex(pattern):
 
 EVENTS_API_PATH = "intake/v2/events"
 AGENT_CONFIG_PATH = "config/v1/agents"
+SERVER_INFO_PATH = "/"
 
 TRACE_CONTEXT_VERSION = 0
 TRACEPARENT_HEADER_NAME = "traceparent"

elasticapm/transport/http.py

@@ -168,6 +168,29 @@ def get_config(self, current_version=None, keys=None):
             logger.warning("Failed decoding APM Server response as JSON: %s", body)
             return current_version, None, max_age
 
+    def _process_queue(self):
+        if not self.client.server_version:
+            self.fetch_server_info()
+        super()._process_queue()
+
+    def fetch_server_info(self):
+        headers = self._headers.copy() if self._headers else {}
+        headers.update(self.auth_headers)
+        headers["accept"] = "text/plain"
+        try:
+            response = self.http.urlopen("GET", self._server_info_url, headers=headers, timeout=self._timeout)
+            body = response.data
+            data = json_encoder.loads(body.decode("utf8"))
+            version = data["version"]
+            logger.info("Fetched APM Server version %s", version)
+            self.client.server_version = version_string_to_tuple(version)
+        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:
+            logger.warning("HTTP error while fetching server information: %s", str(e))
+        except json.JSONDecodeError as e:
+            logger.warning("JSON decoding error while fetching server information: %s", str(e))
+        except KeyError:
+            logger.warning("No version key found in server response: %s", response.data)
+
     @property
     def cert_fingerprint(self):
         if self._server_cert:
@@ -192,5 +215,12 @@ def ca_certs(self):
         return certifi.where() if (certifi and self.client.config.use_certifi) else None
 
 
+def version_string_to_tuple(version):
+    if version:
+        version_parts = re.split(r"[.\-]", version)
+        return tuple(int(p) if p.isdigit() else p for p in version_parts)
+    return ()
+
+
 # left for backwards compatibility
 AsyncTransport = Transport

elasticapm/transport/http_base.py

@@ -61,6 +61,7 @@ def __init__(
         }
         base, sep, tail = self._url.rpartition(constants.EVENTS_API_PATH)
         self._config_url = "".join((base, constants.AGENT_CONFIG_PATH, tail))
+        self._server_info_url = "".join((base, constants.SERVER_INFO_PATH, tail))
         super(HTTPTransportBase, self).__init__(client, compress_level=compress_level, **kwargs)
 
     def send(self, data):

tests/contrib/django/fixtures.py

@@ -79,6 +79,7 @@ def django_sending_elasticapm_client(request, validating_httpserver):
     validating_httpserver.serve_content(code=202, content="", headers={"Location": "http://example.com/foo"})
     client_config = getattr(request, "param", {})
     client_config.setdefault("server_url", validating_httpserver.url)
+    client_config.setdefault("server_version", (8, 0, 0))
     client_config.setdefault("service_name", "app")
     client_config.setdefault("secret_token", "secret")
     client_config.setdefault("transport_class", "elasticapm.transport.http.Transport")

tests/fixtures.py

@@ -292,6 +292,7 @@ def sending_elasticapm_client(request, validating_httpserver):
     client_config.setdefault("include_paths", ("*/tests/*",))
     client_config.setdefault("metrics_interval", "0ms")
     client_config.setdefault("central_config", "false")
+    client_config.setdefault("server_version", (8, 0, 0))
     client = Client(**client_config)
     client.httpserver = validating_httpserver
     yield client

tests/transports/test_urllib3.py

@@ -39,8 +39,9 @@
 
 from elasticapm.conf import constants
 from elasticapm.transport.exceptions import TransportException
-from elasticapm.transport.http import Transport
+from elasticapm.transport.http import Transport, version_string_to_tuple
 from elasticapm.utils import compat
+from tests.utils import assert_any_record_contains
 
 try:
     import urlparse
@@ -354,3 +355,65 @@ def test_use_certifi(elasticapm_client):
     assert transport.ca_certs == certifi.where()
     elasticapm_client.config.update("2", use_certifi=False)
     assert not transport.ca_certs
+
+
+@pytest.mark.parametrize(
+    "version,expected",
+    [
+        (
+            "1.2.3",
+            (1, 2, 3),
+        ),
+        (
+            "1.2.3-alpha1",
+            (1, 2, 3, "alpha1"),
+        ),
+        (
+            "1.2.3alpha1",
+            (1, 2, "3alpha1"),
+        ),
+        (
+            "",
+            (),
+        ),
+    ],
+)
+def test_server_version_to_tuple(version, expected):
+    assert version_string_to_tuple(version) == expected
+
+
+def test_fetch_server_info(waiting_httpserver, elasticapm_client):
+    waiting_httpserver.serve_content(
+        code=200,
+        content=b'{"version": "8.0.0-alpha1"}',
+    )
+    url = waiting_httpserver.url
+    transport = Transport(url + "/" + constants.EVENTS_API_PATH, client=elasticapm_client)
+    transport.fetch_server_info()
+    assert elasticapm_client.server_version == (8, 0, 0, "alpha1")
+
+
+def test_fetch_server_info_no_json(waiting_httpserver, caplog, elasticapm_client):
+    waiting_httpserver.serve_content(
+        code=200,
+        content=b'"version": "8.0.0-alpha1"',
+    )
+    url = waiting_httpserver.url
+    transport = Transport(url + "/" + constants.EVENTS_API_PATH, client=elasticapm_client)
+    with caplog.at_level("WARNING"):
+        transport.fetch_server_info()
+    assert elasticapm_client.server_version is None
+    assert_any_record_contains(caplog.records, "JSON decoding error while fetching server information")
+
+
+def test_fetch_server_info_no_version(waiting_httpserver, caplog, elasticapm_client):
+    waiting_httpserver.serve_content(
+        code=200,
+        content=b"{}",
+    )
+    url = waiting_httpserver.url
+    transport = Transport(url + "/" + constants.EVENTS_API_PATH, client=elasticapm_client)
+    with caplog.at_level("WARNING"):
+        transport.fetch_server_info()
+    assert elasticapm_client.server_version is None
+    assert_any_record_contains(caplog.records, "No version key found in server response")