We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[META 690] Ensure a minimum retry interval of 5 seconds in fetching central configuration
See meta issue for the description and details:
- Meta issue: https://github.com/elastic/apm/issues/690


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `elasticapm/transport/http.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 
3 #  BSD 3-Clause License
4 #
5 #  Copyright (c) 2019, Elasticsearch BV
6 #  All rights reserved.
7 #
8 #  Redistribution and use in source and binary forms, with or without
9 #  modification, are permitted provided that the following conditions are met:
10 #
11 #  * Redistributions of source code must retain the above copyright notice, this
12 #    list of conditions and the following disclaimer.
13 #
14 #  * Redistributions in binary form must reproduce the above copyright notice,
15 #    this list of conditions and the following disclaimer in the documentation
16 #    and/or other materials provided with the distribution.
17 #
18 #  * Neither the name of the copyright holder nor the names of its
19 #    contributors may be used to endorse or promote products derived from
20 #    this software without specific prior written permission.
21 #
22 #  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
23 #  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
24 #  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
25 #  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
26 #  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
27 #  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
28 #  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
29 #  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
30 #  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
31 #  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32 
33 import hashlib
34 import json
35 import re
36 import ssl
37 import urllib.parse
38 from urllib.request import getproxies_environment, proxy_bypass_environment
39 
40 import urllib3
41 from urllib3.exceptions import MaxRetryError, TimeoutError
42 
43 from elasticapm.transport.exceptions import TransportException
44 from elasticapm.transport.http_base import HTTPTransportBase
45 from elasticapm.utils import json_encoder, read_pem_file
46 from elasticapm.utils.logging import get_logger
47 
48 try:
49     import certifi
50 except ImportError:
51     certifi = None
52 
53 logger = get_logger("elasticapm.transport.http")
54 
55 
56 class Transport(HTTPTransportBase):
57     def __init__(self, url: str, *args, **kwargs) -> None:
58         super(Transport, self).__init__(url, *args, **kwargs)
59         pool_kwargs = {"cert_reqs": "CERT_REQUIRED", "ca_certs": self.ca_certs, "block": True}
60         if url.startswith("https"):
61             if self._server_cert:
62                 pool_kwargs.update(
63                     {"assert_fingerprint": self.cert_fingerprint, "assert_hostname": False, "cert_reqs": ssl.CERT_NONE}
64                 )
65                 del pool_kwargs["ca_certs"]
66             elif not self._verify_server_cert:
67                 pool_kwargs["cert_reqs"] = ssl.CERT_NONE
68                 pool_kwargs["assert_hostname"] = False
69         self._pool_kwargs = pool_kwargs
70         self._http = None
71         self._url = url
72 
73     def send(self, data, forced_flush=False):
74         response = None
75 
76         headers = self._headers.copy() if self._headers else {}
77         headers.update(self.auth_headers)
78         headers.update(
79             {
80                 b"Content-Type": b"application/x-ndjson",
81                 b"Content-Encoding": b"gzip",
82             }
83         )
84 
85         url = self._url
86         if forced_flush:
87             url = f"{url}?flushed=true"
88         try:
89             try:
90                 response = self.http.urlopen(
91                     "POST", url, body=data, headers=headers, timeout=self._timeout, preload_content=False
92                 )
93                 logger.debug("Sent request, url=%s size=%.2fkb status=%s", url, len(data) / 1024.0, response.status)
94             except Exception as e:
95                 print_trace = True
96                 if isinstance(e, MaxRetryError) and isinstance(e.reason, TimeoutError):
97                     message = "Connection to APM Server timed out " "(url: %s, timeout: %s seconds)" % (
98                         self._url,
99                         self._timeout,
100                     )
101                     print_trace = False
102                 else:
103                     message = "Unable to reach APM Server: %s (url: %s)" % (e, self._url)
104                 raise TransportException(message, data, print_trace=print_trace)
105             body = response.read()
106             if response.status >= 400:
107                 if response.status == 429:  # rate-limited
108                     message = "Temporarily rate limited: "
109                     print_trace = False
110                 else:
111                     message = "HTTP %s: " % response.status
112                     print_trace = True
113                 message += body.decode("utf8", errors="replace")[:10000]
114                 raise TransportException(message, data, print_trace=print_trace)
115             return response.getheader("Location")
116         finally:
117             if response:
118                 response.close()
119 
120     @property
121     def http(self) -> urllib3.PoolManager:
122         if not self._http:
123             url_parts = urllib.parse.urlparse(self._url)
124             proxies = getproxies_environment()
125             proxy_url = proxies.get("https", proxies.get("http", None))
126             if proxy_url and not proxy_bypass_environment(url_parts.netloc):
127                 self._http = urllib3.ProxyManager(proxy_url, **self._pool_kwargs)
128             else:
129                 self._http = urllib3.PoolManager(**self._pool_kwargs)
130         return self._http
131 
132     def handle_fork(self) -> None:
133         # reset http pool to avoid sharing connections with the parent process
134         self._http = None
135 
136     def get_config(self, current_version=None, keys=None):
137         """
138         Gets configuration from a remote APM Server
139 
140         :param current_version: version of the current configuration
141         :param keys: a JSON-serializable dict to identify this instance, e.g.
142                 {
143                     "service": {
144                         "name": "foo",
145                         "environment": "bar"
146                     }
147                 }
148         :return: a three-tuple of new version, config dictionary and validity in seconds.
149                  Any element of the tuple can be None.
150         """
151         url = self._config_url
152         data = json_encoder.dumps(keys).encode("utf-8")
153         headers = self._headers.copy()
154         headers[b"Content-Type"] = "application/json"
155         headers.update(self.auth_headers)
156         max_age = 300
157         if current_version:
158             headers["If-None-Match"] = current_version
159         try:
160             response = self.http.urlopen(
161                 "POST", url, body=data, headers=headers, timeout=self._timeout, preload_content=False
162             )
163         except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:
164             logger.debug("HTTP error while fetching remote config: %s", str(e))
165             return current_version, None, max_age
166         body = response.read()
167         if "Cache-Control" in response.headers:
168             try:
169                 max_age = int(next(re.finditer(r"max-age=(\d+)", response.headers["Cache-Control"])).groups()[0])
170             except StopIteration:
171                 logger.debug("Could not parse Cache-Control header: %s", response.headers["Cache-Control"])
172         if response.status == 304:
173             # config is unchanged, return
174             logger.debug("Configuration unchanged")
175             return current_version, None, max_age
176         elif response.status >= 400:
177             return None, None, max_age
178 
179         if not body:
180             logger.debug("APM Server answered with empty body and status code %s", response.status)
181             return current_version, None, max_age
182         body = body.decode("utf-8")
183         try:
184             data = json_encoder.loads(body)
185             return response.headers.get("Etag"), data, max_age
186         except json.JSONDecodeError:
187             logger.warning("Failed decoding APM Server response as JSON: %s", body)
188             return current_version, None, max_age
189 
190     def _process_queue(self):
191         if not self.client.server_version:
192             self.fetch_server_info()
193         super()._process_queue()
194 
195     def fetch_server_info(self):
196         headers = self._headers.copy() if self._headers else {}
197         headers.update(self.auth_headers)
198         headers[b"accept"] = b"text/plain"
199         try:
200             response = self.http.urlopen("GET", self._server_info_url, headers=headers, timeout=self._timeout)
201             body = response.data
202             data = json_encoder.loads(body.decode("utf8"))
203             version = data["version"]
204             logger.debug("Fetched APM Server version %s", version)
205             self.client.server_version = version_string_to_tuple(version)
206         except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:
207             logger.warning("HTTP error while fetching server information: %s", str(e))
208         except json.JSONDecodeError as e:
209             logger.warning("JSON decoding error while fetching server information: %s", str(e))
210         except (KeyError, TypeError):
211             logger.warning("No version key found in server response: %s", response.data)
212 
213     @property
214     def cert_fingerprint(self):
215         if self._server_cert:
216             with open(self._server_cert, "rb") as f:
217                 cert_data = read_pem_file(f)
218             digest = hashlib.sha256()
219             digest.update(cert_data)
220             return digest.hexdigest()
221         return None
222 
223     @property
224     def auth_headers(self):
225         headers = super(Transport, self).auth_headers
226         return {k.encode("ascii"): v.encode("ascii") for k, v in headers.items()}
227 
228     @property
229     def ca_certs(self):
230         """
231         Return location of certificate store. If it is available and not disabled via setting,
232         this will return the location of the certifi certificate store.
233         """
234         return certifi.where() if (certifi and self.client.config.use_certifi) else None
235 
236 
237 def version_string_to_tuple(version):
238     if version:
239         version_parts = re.split(r"[.\-]", version)
240         return tuple(int(p) if p.isdigit() else p for p in version_parts)
241     return ()
242 
243 
244 # left for backwards compatibility
245 AsyncTransport = Transport
246
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/elasticapm/transport/http.py b/elasticapm/transport/http.py
--- a/elasticapm/transport/http.py
+++ b/elasticapm/transport/http.py
@@ -164,11 +164,9 @@
             logger.debug("HTTP error while fetching remote config: %s", str(e))
             return current_version, None, max_age
         body = response.read()
-        if "Cache-Control" in response.headers:
-            try:
-                max_age = int(next(re.finditer(r"max-age=(\d+)", response.headers["Cache-Control"])).groups()[0])
-            except StopIteration:
-                logger.debug("Could not parse Cache-Control header: %s", response.headers["Cache-Control"])
+
+        max_age = self._get_cache_control_max_age(response.headers) or max_age
+
         if response.status == 304:
             # config is unchanged, return
             logger.debug("Configuration unchanged")
@@ -187,6 +185,22 @@
             logger.warning("Failed decoding APM Server response as JSON: %s", body)
             return current_version, None, max_age
 
+    def _get_cache_control_max_age(self, response_headers):
+        max_age = None
+        if "Cache-Control" in response_headers:
+            try:
+                cc_max_age = int(next(re.finditer(r"max-age=(\d+)", response_headers["Cache-Control"])).groups()[0])
+                if cc_max_age <= 0:
+                    # max_age remains at default value
+                    pass
+                elif cc_max_age < 5:
+                    max_age = 5
+                else:
+                    max_age = cc_max_age
+            except StopIteration:
+                logger.debug("Could not parse Cache-Control header: %s", response_headers["Cache-Control"])
+        return max_age
+
     def _process_queue(self):
         if not self.client.server_version:
             self.fetch_server_info()

{"golden_diff": "diff --git a/elasticapm/transport/http.py b/elasticapm/transport/http.py\n--- a/elasticapm/transport/http.py\n+++ b/elasticapm/transport/http.py\n@@ -164,11 +164,9 @@\n             logger.debug(\"HTTP error while fetching remote config: %s\", str(e))\n             return current_version, None, max_age\n         body = response.read()\n-        if \"Cache-Control\" in response.headers:\n-            try:\n-                max_age = int(next(re.finditer(r\"max-age=(\\d+)\", response.headers[\"Cache-Control\"])).groups()[0])\n-            except StopIteration:\n-                logger.debug(\"Could not parse Cache-Control header: %s\", response.headers[\"Cache-Control\"])\n+\n+        max_age = self._get_cache_control_max_age(response.headers) or max_age\n+\n         if response.status == 304:\n             # config is unchanged, return\n             logger.debug(\"Configuration unchanged\")\n@@ -187,6 +185,22 @@\n             logger.warning(\"Failed decoding APM Server response as JSON: %s\", body)\n             return current_version, None, max_age\n \n+    def _get_cache_control_max_age(self, response_headers):\n+        max_age = None\n+        if \"Cache-Control\" in response_headers:\n+            try:\n+                cc_max_age = int(next(re.finditer(r\"max-age=(\\d+)\", response_headers[\"Cache-Control\"])).groups()[0])\n+                if cc_max_age <= 0:\n+                    # max_age remains at default value\n+                    pass\n+                elif cc_max_age < 5:\n+                    max_age = 5\n+                else:\n+                    max_age = cc_max_age\n+            except StopIteration:\n+                logger.debug(\"Could not parse Cache-Control header: %s\", response_headers[\"Cache-Control\"])\n+        return max_age\n+\n     def _process_queue(self):\n         if not self.client.server_version:\n             self.fetch_server_info()\n", "issue": "[META 690] Ensure a minimum retry interval of 5 seconds in fetching central configuration\nSee meta issue for the description and details:\r\n- Meta issue: https://github.com/elastic/apm/issues/690\r\n\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n\n#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nimport hashlib\nimport json\nimport re\nimport ssl\nimport urllib.parse\nfrom urllib.request import getproxies_environment, proxy_bypass_environment\n\nimport urllib3\nfrom urllib3.exceptions import MaxRetryError, TimeoutError\n\nfrom elasticapm.transport.exceptions import TransportException\nfrom elasticapm.transport.http_base import HTTPTransportBase\nfrom elasticapm.utils import json_encoder, read_pem_file\nfrom elasticapm.utils.logging import get_logger\n\ntry:\n    import certifi\nexcept ImportError:\n    certifi = None\n\nlogger = get_logger(\"elasticapm.transport.http\")\n\n\nclass Transport(HTTPTransportBase):\n    def __init__(self, url: str, *args, **kwargs) -> None:\n        super(Transport, self).__init__(url, *args, **kwargs)\n        pool_kwargs = {\"cert_reqs\": \"CERT_REQUIRED\", \"ca_certs\": self.ca_certs, \"block\": True}\n        if url.startswith(\"https\"):\n            if self._server_cert:\n                pool_kwargs.update(\n                    {\"assert_fingerprint\": self.cert_fingerprint, \"assert_hostname\": False, \"cert_reqs\": ssl.CERT_NONE}\n                )\n                del pool_kwargs[\"ca_certs\"]\n            elif not self._verify_server_cert:\n                pool_kwargs[\"cert_reqs\"] = ssl.CERT_NONE\n                pool_kwargs[\"assert_hostname\"] = False\n        self._pool_kwargs = pool_kwargs\n        self._http = None\n        self._url = url\n\n    def send(self, data, forced_flush=False):\n        response = None\n\n        headers = self._headers.copy() if self._headers else {}\n        headers.update(self.auth_headers)\n        headers.update(\n            {\n                b\"Content-Type\": b\"application/x-ndjson\",\n                b\"Content-Encoding\": b\"gzip\",\n            }\n        )\n\n        url = self._url\n        if forced_flush:\n            url = f\"{url}?flushed=true\"\n        try:\n            try:\n                response = self.http.urlopen(\n                    \"POST\", url, body=data, headers=headers, timeout=self._timeout, preload_content=False\n                )\n                logger.debug(\"Sent request, url=%s size=%.2fkb status=%s\", url, len(data) / 1024.0, response.status)\n            except Exception as e:\n                print_trace = True\n                if isinstance(e, MaxRetryError) and isinstance(e.reason, TimeoutError):\n                    message = \"Connection to APM Server timed out \" \"(url: %s, timeout: %s seconds)\" % (\n                        self._url,\n                        self._timeout,\n                    )\n                    print_trace = False\n                else:\n                    message = \"Unable to reach APM Server: %s (url: %s)\" % (e, self._url)\n                raise TransportException(message, data, print_trace=print_trace)\n            body = response.read()\n            if response.status >= 400:\n                if response.status == 429:  # rate-limited\n                    message = \"Temporarily rate limited: \"\n                    print_trace = False\n                else:\n                    message = \"HTTP %s: \" % response.status\n                    print_trace = True\n                message += body.decode(\"utf8\", errors=\"replace\")[:10000]\n                raise TransportException(message, data, print_trace=print_trace)\n            return response.getheader(\"Location\")\n        finally:\n            if response:\n                response.close()\n\n    @property\n    def http(self) -> urllib3.PoolManager:\n        if not self._http:\n            url_parts = urllib.parse.urlparse(self._url)\n            proxies = getproxies_environment()\n            proxy_url = proxies.get(\"https\", proxies.get(\"http\", None))\n            if proxy_url and not proxy_bypass_environment(url_parts.netloc):\n                self._http = urllib3.ProxyManager(proxy_url, **self._pool_kwargs)\n            else:\n                self._http = urllib3.PoolManager(**self._pool_kwargs)\n        return self._http\n\n    def handle_fork(self) -> None:\n        # reset http pool to avoid sharing connections with the parent process\n        self._http = None\n\n    def get_config(self, current_version=None, keys=None):\n        \"\"\"\n        Gets configuration from a remote APM Server\n\n        :param current_version: version of the current configuration\n        :param keys: a JSON-serializable dict to identify this instance, e.g.\n                {\n                    \"service\": {\n                        \"name\": \"foo\",\n                        \"environment\": \"bar\"\n                    }\n                }\n        :return: a three-tuple of new version, config dictionary and validity in seconds.\n                 Any element of the tuple can be None.\n        \"\"\"\n        url = self._config_url\n        data = json_encoder.dumps(keys).encode(\"utf-8\")\n        headers = self._headers.copy()\n        headers[b\"Content-Type\"] = \"application/json\"\n        headers.update(self.auth_headers)\n        max_age = 300\n        if current_version:\n            headers[\"If-None-Match\"] = current_version\n        try:\n            response = self.http.urlopen(\n                \"POST\", url, body=data, headers=headers, timeout=self._timeout, preload_content=False\n            )\n        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:\n            logger.debug(\"HTTP error while fetching remote config: %s\", str(e))\n            return current_version, None, max_age\n        body = response.read()\n        if \"Cache-Control\" in response.headers:\n            try:\n                max_age = int(next(re.finditer(r\"max-age=(\\d+)\", response.headers[\"Cache-Control\"])).groups()[0])\n            except StopIteration:\n                logger.debug(\"Could not parse Cache-Control header: %s\", response.headers[\"Cache-Control\"])\n        if response.status == 304:\n            # config is unchanged, return\n            logger.debug(\"Configuration unchanged\")\n            return current_version, None, max_age\n        elif response.status >= 400:\n            return None, None, max_age\n\n        if not body:\n            logger.debug(\"APM Server answered with empty body and status code %s\", response.status)\n            return current_version, None, max_age\n        body = body.decode(\"utf-8\")\n        try:\n            data = json_encoder.loads(body)\n            return response.headers.get(\"Etag\"), data, max_age\n        except json.JSONDecodeError:\n            logger.warning(\"Failed decoding APM Server response as JSON: %s\", body)\n            return current_version, None, max_age\n\n    def _process_queue(self):\n        if not self.client.server_version:\n            self.fetch_server_info()\n        super()._process_queue()\n\n    def fetch_server_info(self):\n        headers = self._headers.copy() if self._headers else {}\n        headers.update(self.auth_headers)\n        headers[b\"accept\"] = b\"text/plain\"\n        try:\n            response = self.http.urlopen(\"GET\", self._server_info_url, headers=headers, timeout=self._timeout)\n            body = response.data\n            data = json_encoder.loads(body.decode(\"utf8\"))\n            version = data[\"version\"]\n            logger.debug(\"Fetched APM Server version %s\", version)\n            self.client.server_version = version_string_to_tuple(version)\n        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:\n            logger.warning(\"HTTP error while fetching server information: %s\", str(e))\n        except json.JSONDecodeError as e:\n            logger.warning(\"JSON decoding error while fetching server information: %s\", str(e))\n        except (KeyError, TypeError):\n            logger.warning(\"No version key found in server response: %s\", response.data)\n\n    @property\n    def cert_fingerprint(self):\n        if self._server_cert:\n            with open(self._server_cert, \"rb\") as f:\n                cert_data = read_pem_file(f)\n            digest = hashlib.sha256()\n            digest.update(cert_data)\n            return digest.hexdigest()\n        return None\n\n    @property\n    def auth_headers(self):\n        headers = super(Transport, self).auth_headers\n        return {k.encode(\"ascii\"): v.encode(\"ascii\") for k, v in headers.items()}\n\n    @property\n    def ca_certs(self):\n        \"\"\"\n        Return location of certificate store. If it is available and not disabled via setting,\n        this will return the location of the certifi certificate store.\n        \"\"\"\n        return certifi.where() if (certifi and self.client.config.use_certifi) else None\n\n\ndef version_string_to_tuple(version):\n    if version:\n        version_parts = re.split(r\"[.\\-]\", version)\n        return tuple(int(p) if p.isdigit() else p for p in version_parts)\n    return ()\n\n\n# left for backwards compatibility\nAsyncTransport = Transport\n", "path": "elasticapm/transport/http.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n\n#  BSD 3-Clause License\n#\n#  Copyright (c) 2019, Elasticsearch BV\n#  All rights reserved.\n#\n#  Redistribution and use in source and binary forms, with or without\n#  modification, are permitted provided that the following conditions are met:\n#\n#  * Redistributions of source code must retain the above copyright notice, this\n#    list of conditions and the following disclaimer.\n#\n#  * Redistributions in binary form must reproduce the above copyright notice,\n#    this list of conditions and the following disclaimer in the documentation\n#    and/or other materials provided with the distribution.\n#\n#  * Neither the name of the copyright holder nor the names of its\n#    contributors may be used to endorse or promote products derived from\n#    this software without specific prior written permission.\n#\n#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS \"AS IS\"\n#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE\n#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE\n#  DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE\n#  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL\n#  DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR\n#  SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER\n#  CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,\n#  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n#  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n\nimport hashlib\nimport json\nimport re\nimport ssl\nimport urllib.parse\nfrom urllib.request import getproxies_environment, proxy_bypass_environment\n\nimport urllib3\nfrom urllib3.exceptions import MaxRetryError, TimeoutError\n\nfrom elasticapm.transport.exceptions import TransportException\nfrom elasticapm.transport.http_base import HTTPTransportBase\nfrom elasticapm.utils import json_encoder, read_pem_file\nfrom elasticapm.utils.logging import get_logger\n\ntry:\n    import certifi\nexcept ImportError:\n    certifi = None\n\nlogger = get_logger(\"elasticapm.transport.http\")\n\n\nclass Transport(HTTPTransportBase):\n    def __init__(self, url: str, *args, **kwargs) -> None:\n        super(Transport, self).__init__(url, *args, **kwargs)\n        pool_kwargs = {\"cert_reqs\": \"CERT_REQUIRED\", \"ca_certs\": self.ca_certs, \"block\": True}\n        if url.startswith(\"https\"):\n            if self._server_cert:\n                pool_kwargs.update(\n                    {\"assert_fingerprint\": self.cert_fingerprint, \"assert_hostname\": False, \"cert_reqs\": ssl.CERT_NONE}\n                )\n                del pool_kwargs[\"ca_certs\"]\n            elif not self._verify_server_cert:\n                pool_kwargs[\"cert_reqs\"] = ssl.CERT_NONE\n                pool_kwargs[\"assert_hostname\"] = False\n        self._pool_kwargs = pool_kwargs\n        self._http = None\n        self._url = url\n\n    def send(self, data, forced_flush=False):\n        response = None\n\n        headers = self._headers.copy() if self._headers else {}\n        headers.update(self.auth_headers)\n        headers.update(\n            {\n                b\"Content-Type\": b\"application/x-ndjson\",\n                b\"Content-Encoding\": b\"gzip\",\n            }\n        )\n\n        url = self._url\n        if forced_flush:\n            url = f\"{url}?flushed=true\"\n        try:\n            try:\n                response = self.http.urlopen(\n                    \"POST\", url, body=data, headers=headers, timeout=self._timeout, preload_content=False\n                )\n                logger.debug(\"Sent request, url=%s size=%.2fkb status=%s\", url, len(data) / 1024.0, response.status)\n            except Exception as e:\n                print_trace = True\n                if isinstance(e, MaxRetryError) and isinstance(e.reason, TimeoutError):\n                    message = \"Connection to APM Server timed out \" \"(url: %s, timeout: %s seconds)\" % (\n                        self._url,\n                        self._timeout,\n                    )\n                    print_trace = False\n                else:\n                    message = \"Unable to reach APM Server: %s (url: %s)\" % (e, self._url)\n                raise TransportException(message, data, print_trace=print_trace)\n            body = response.read()\n            if response.status >= 400:\n                if response.status == 429:  # rate-limited\n                    message = \"Temporarily rate limited: \"\n                    print_trace = False\n                else:\n                    message = \"HTTP %s: \" % response.status\n                    print_trace = True\n                message += body.decode(\"utf8\", errors=\"replace\")[:10000]\n                raise TransportException(message, data, print_trace=print_trace)\n            return response.getheader(\"Location\")\n        finally:\n            if response:\n                response.close()\n\n    @property\n    def http(self) -> urllib3.PoolManager:\n        if not self._http:\n            url_parts = urllib.parse.urlparse(self._url)\n            proxies = getproxies_environment()\n            proxy_url = proxies.get(\"https\", proxies.get(\"http\", None))\n            if proxy_url and not proxy_bypass_environment(url_parts.netloc):\n                self._http = urllib3.ProxyManager(proxy_url, **self._pool_kwargs)\n            else:\n                self._http = urllib3.PoolManager(**self._pool_kwargs)\n        return self._http\n\n    def handle_fork(self) -> None:\n        # reset http pool to avoid sharing connections with the parent process\n        self._http = None\n\n    def get_config(self, current_version=None, keys=None):\n        \"\"\"\n        Gets configuration from a remote APM Server\n\n        :param current_version: version of the current configuration\n        :param keys: a JSON-serializable dict to identify this instance, e.g.\n                {\n                    \"service\": {\n                        \"name\": \"foo\",\n                        \"environment\": \"bar\"\n                    }\n                }\n        :return: a three-tuple of new version, config dictionary and validity in seconds.\n                 Any element of the tuple can be None.\n        \"\"\"\n        url = self._config_url\n        data = json_encoder.dumps(keys).encode(\"utf-8\")\n        headers = self._headers.copy()\n        headers[b\"Content-Type\"] = \"application/json\"\n        headers.update(self.auth_headers)\n        max_age = 300\n        if current_version:\n            headers[\"If-None-Match\"] = current_version\n        try:\n            response = self.http.urlopen(\n                \"POST\", url, body=data, headers=headers, timeout=self._timeout, preload_content=False\n            )\n        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:\n            logger.debug(\"HTTP error while fetching remote config: %s\", str(e))\n            return current_version, None, max_age\n        body = response.read()\n\n        max_age = self._get_cache_control_max_age(response.headers) or max_age\n\n        if response.status == 304:\n            # config is unchanged, return\n            logger.debug(\"Configuration unchanged\")\n            return current_version, None, max_age\n        elif response.status >= 400:\n            return None, None, max_age\n\n        if not body:\n            logger.debug(\"APM Server answered with empty body and status code %s\", response.status)\n            return current_version, None, max_age\n        body = body.decode(\"utf-8\")\n        try:\n            data = json_encoder.loads(body)\n            return response.headers.get(\"Etag\"), data, max_age\n        except json.JSONDecodeError:\n            logger.warning(\"Failed decoding APM Server response as JSON: %s\", body)\n            return current_version, None, max_age\n\n    def _get_cache_control_max_age(self, response_headers):\n        max_age = None\n        if \"Cache-Control\" in response_headers:\n            try:\n                cc_max_age = int(next(re.finditer(r\"max-age=(\\d+)\", response_headers[\"Cache-Control\"])).groups()[0])\n                if cc_max_age <= 0:\n                    # max_age remains at default value\n                    pass\n                elif cc_max_age < 5:\n                    max_age = 5\n                else:\n                    max_age = cc_max_age\n            except StopIteration:\n                logger.debug(\"Could not parse Cache-Control header: %s\", response_headers[\"Cache-Control\"])\n        return max_age\n\n    def _process_queue(self):\n        if not self.client.server_version:\n            self.fetch_server_info()\n        super()._process_queue()\n\n    def fetch_server_info(self):\n        headers = self._headers.copy() if self._headers else {}\n        headers.update(self.auth_headers)\n        headers[b\"accept\"] = b\"text/plain\"\n        try:\n            response = self.http.urlopen(\"GET\", self._server_info_url, headers=headers, timeout=self._timeout)\n            body = response.data\n            data = json_encoder.loads(body.decode(\"utf8\"))\n            version = data[\"version\"]\n            logger.debug(\"Fetched APM Server version %s\", version)\n            self.client.server_version = version_string_to_tuple(version)\n        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:\n            logger.warning(\"HTTP error while fetching server information: %s\", str(e))\n        except json.JSONDecodeError as e:\n            logger.warning(\"JSON decoding error while fetching server information: %s\", str(e))\n        except (KeyError, TypeError):\n            logger.warning(\"No version key found in server response: %s\", response.data)\n\n    @property\n    def cert_fingerprint(self):\n        if self._server_cert:\n            with open(self._server_cert, \"rb\") as f:\n                cert_data = read_pem_file(f)\n            digest = hashlib.sha256()\n            digest.update(cert_data)\n            return digest.hexdigest()\n        return None\n\n    @property\n    def auth_headers(self):\n        headers = super(Transport, self).auth_headers\n        return {k.encode(\"ascii\"): v.encode(\"ascii\") for k, v in headers.items()}\n\n    @property\n    def ca_certs(self):\n        \"\"\"\n        Return location of certificate store. If it is available and not disabled via setting,\n        this will return the location of the certifi certificate store.\n        \"\"\"\n        return certifi.where() if (certifi and self.client.config.use_certifi) else None\n\n\ndef version_string_to_tuple(version):\n    if version:\n        version_parts = re.split(r\"[.\\-]\", version)\n        return tuple(int(p) if p.isdigit() else p for p in version_parts)\n    return ()\n\n\n# left for backwards compatibility\nAsyncTransport = Transport\n", "path": "elasticapm/transport/http.py"}]}