fix(telemetry): flush telemetry writer when a process terminates (backport #3598) (#3621)

This is an automatic backport of pull request #3598 done by [Mergify](https://mergify.com).


---


<details>
<summary>Mergify commands and options</summary>

<br />

More conditions and actions can be found in the [documentation](https://docs.mergify.com/).

You can also trigger Mergify actions by commenting on this pull request:

- `@Mergifyio refresh` will re-evaluate the rules
- `@Mergifyio rebase` will rebase this PR on its base branch
- `@Mergifyio update` will merge the base branch into this PR
- `@Mergifyio backport <destination>` will backport this PR on `<destination>` branch

Additionally, on Mergify [dashboard](https://dashboard.mergify.com/) you can:

- look at your merge queues
- generate the Mergify configuration with the config editor.

Finally, you can contact us on https://mergify.com
</details>

Author: mergify[bot]

ddtrace/internal/telemetry/writer.py

@@ -4,6 +4,7 @@
 from typing import List
 from typing import Optional
 
+from ...internal import atexit
 from ...internal import forksafe
 from ...settings import _config as config
 from ..agent import get_connection
@@ -116,11 +117,15 @@ def periodic(self):
                     exc_info=True,
                 )
 
-    def shutdown(self):
-        # type: () -> None
+    def on_shutdown(self):
         self._app_closing_event()
         self.periodic()
 
+    def _stop_service(self, *args, **kwargs):
+        # type: (...) -> None
+        super(TelemetryWriter, self)._stop_service()
+        self.join()
+
     def add_event(self, payload, payload_type):
         # type: (Dict, str) -> None
         """
@@ -229,10 +234,9 @@ def disable(self):
                 return
 
             forksafe.unregister(self._restart)
+            atexit.unregister(self.stop)
 
-            self.stop()
-
-            self.join()
+        self.stop()
 
     def enable(self):
         # type: () -> None
@@ -248,6 +252,7 @@ def enable(self):
             self._enabled = True
 
             forksafe.register(self._restart)
+            atexit.register(self.stop)
 
         # add_event _locks around adding to the events queue
         self.app_started_event()

tests/tracer/telemetry/test_writer.py

@@ -100,9 +100,9 @@ def test_add_app_started_event(mock_time, mock_send_request, telemetry_writer):
 
 
 def test_add_app_closing_event(mock_time, mock_send_request, telemetry_writer):
-    """asserts that shutdown() queues and sends an app-closing telemetry request"""
+    """asserts that on_shutdown() queues and sends an app-closing telemetry request"""
     # send app closed event
-    telemetry_writer.shutdown()
+    telemetry_writer.on_shutdown()
     # assert that one request was sent
     assert len(httpretty.latest_requests()) == 1
     # ensure an app closing event was sent
@@ -191,7 +191,7 @@ def test_send_failing_request(mock_status, mock_send_request, telemetry_writer):
     """asserts that a warning is logged when an unsuccessful response is returned by the http client"""
     with mock.patch("ddtrace.internal.telemetry.writer.log") as log:
         # sends failing app-closing event
-        telemetry_writer.shutdown()
+        telemetry_writer.on_shutdown()
         # asserts unsuccessful status code was logged
         log.warning.assert_called_with(
             "failed to send telemetry to the Datadog Agent at %s/%s. response: %s",
@@ -203,7 +203,7 @@ def test_send_failing_request(mock_status, mock_send_request, telemetry_writer):
     assert len(httpretty.latest_requests()) == 1
 
 
-def test_send_request_exception(telemetry_writer):
+def test_send_request_exception():
     """asserts that an error is logged when an exception is raised by the http client"""
     # create a telemetry writer with an invalid agent url.
     # this will raise an Exception on _send_request
@@ -212,7 +212,7 @@ def test_send_request_exception(telemetry_writer):
 
     with mock.patch("ddtrace.internal.telemetry.writer.log") as log:
         # sends failing app-closing event
-        telemetry_writer.shutdown()
+        telemetry_writer.on_shutdown()
         # assert an exception was logged
         log.warning.assert_called_with(
             "failed to send telemetry to the Datadog Agent at %s/%s.",
@@ -222,6 +222,12 @@ def test_send_request_exception(telemetry_writer):
         )
 
 
+def test_telemetry_graceful_shutdown(mock_time, mock_send_request, telemetry_writer):
+    telemetry_writer.start()
+    telemetry_writer.stop()
+    assert httpretty.last_request().parsed_body == _get_request_body({}, "app-closing")
+
+
 def _get_request_body(payload, payload_type, seq_id=1):
     """used to test the body of requests intercepted by httpretty"""
     return {