elasticapm: properly cleanup buffer and data views

With Python 3.13 our pattern of buffering revealed some issues because
the underlying BytesIO fileobj may get released before gzip.GzipFile.
This requires a fix in CPython but also some improvements on our side by
properly closing the GzipFile in case of error and also releasing the
memoryview we can from the BytesIO buffer.

These problems manifests as following warnings from unraisable exceptions
running tests:

The closing of the gzip buffer helps with:

  Traceback (most recent call last):
    File "/usr/lib/python3.13/gzip.py", line 362, in close
      fileobj.write(self.compress.flush())
      ~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^
  ValueError: I/O operation on closed file.

Exception ignored in: <_io.BytesIO object at 0x7fbc4335fbf0>

  Traceback (most recent call last):
    File "/venv313/lib/python3.13/site-packages/ecs_logging/_stdlib.py", line 272, in _record_attribute
      def _record_attribute(

  BufferError: Existing exports of data: object cannot be re-sized

Python 3.12 shows the same warnings with the `-X dev` flag.

Author: xrmx

elasticapm/transport/base.py

@@ -250,6 +250,7 @@ def _flush(self, buffer, forced_flush=False) -> None:
         """
         if not self.state.should_try():
             logger.error("dropping flushed data due to transport failure back-off")
+            buffer.close()
         else:
             fileobj = buffer.fileobj  # get a reference to the fileobj before closing the gzip file
             buffer.close()
@@ -261,6 +262,8 @@ def _flush(self, buffer, forced_flush=False) -> None:
             except Exception as e:
                 self.handle_transport_fail(e)
 
+            data.release()
+
     def start_thread(self, pid=None) -> None:
         super(Transport, self).start_thread(pid=pid)
         if (not self._thread or self.pid != self._thread.pid) and not self._closed:

tests/transports/test_base.py

@@ -107,18 +107,25 @@ def test_empty_queue_flush(mock_send, elasticapm_client):
         transport.close()
 
 
-@mock.patch("elasticapm.transport.base.Transport.send")
+@mock.patch("elasticapm.transport.base.Transport._flush")
 @pytest.mark.parametrize("elasticapm_client", [{"api_request_time": "5s"}], indirect=True)
-def test_metadata_prepended(mock_send, elasticapm_client):
+def test_metadata_prepended(mock_flush, elasticapm_client):
     transport = Transport(client=elasticapm_client, compress_level=0)
     transport.start_thread()
     transport.queue("error", {}, flush=True)
     transport.close()
-    assert mock_send.call_count == 1
-    args, kwargs = mock_send.call_args
-    data = gzip.decompress(args[0])
+    assert mock_flush.call_count == 1
+    args, kwargs = mock_flush.call_args
+    buffer = args[0]
+    # this used to mock send but after we fixed a leak of not releasing a memoryview containing
+    # the data we cannot read it anymore. Soreimplement _flush and read the data ourselves
+    fileobj = buffer.fileobj
+    buffer.close()
+    compressed_data = fileobj.getbuffer()
+    data = gzip.decompress(compressed_data)
     data = data.decode("utf-8").split("\n")
     assert "metadata" in data[0]
+    compressed_data.release()
 
 
 @mock.patch("elasticapm.transport.base.Transport.send")