fix(tracing): add span processors shutdown (backport #3508) (#3516)

This is an automatic backport of pull request #3508 done by [Mergify](https://mergify.com).
Cherry-pick of e9b81b9 has failed:
```
On branch mergify/bp/1.0/pr-3508
Your branch is up to date with 'origin/1.0'.

You are currently cherry-picking commit e9b81b9.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	modified:   ddtrace/tracer.py
	modified:   tests/tracer/test_tracer.py

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   ddtrace/internal/processor/__init__.py
	both modified:   ddtrace/internal/processor/trace.py

```


To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/github/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

---


<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/processor/__init__.py

@@ -1,4 +1,5 @@
 import abc
+from typing import Optional
 
 import attr
 import six
@@ -51,3 +52,11 @@ def on_span_finish(self, span):
         applied afterwards.
         """
         pass
+
+    def shutdown(self, timeout):
+        # type: (Optional[float]) -> None
+        """Called when the processor is done being used.
+
+        Any clean-up or flushing should be performed with this method.
+        """
+        pass

ddtrace/internal/processor/trace.py

@@ -11,6 +11,7 @@
 
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.processor import SpanProcessor
+from ddtrace.internal.service import ServiceStatusError
 from ddtrace.internal.writer import TraceWriter
 from ddtrace.span import Span
 
@@ -201,3 +202,19 @@ def on_span_finish(self, span):
 
             log.debug("trace %d has %d spans, %d finished", span.trace_id, len(trace.spans), trace.num_finished)
             return None
+
+    def shutdown(self, timeout):
+        # type: (Optional[float]) -> None
+        """
+        This will stop the background writer/worker and flush any finished traces in the buffer. The tracer cannot be
+        used for tracing after this method has been called. A new tracer instance is required to continue tracing.
+
+        :param timeout: How long in seconds to wait for the background worker to flush traces
+            before exiting or :obj:`None` to block until flushing has successfully completed (default: :obj:`None`)
+        :type timeout: :obj:`int` | :obj:`float` | :obj:`None`
+        """
+        try:
+            self._writer.stop(timeout)
+        except ServiceStatusError:
+            # It's possible the writer never got started in the first place :(
+            pass

ddtrace/tracer.py

@@ -861,22 +861,20 @@ def set_tags(self, tags):
 
     def shutdown(self, timeout=None):
         # type: (Optional[float]) -> None
-        """Shutdown the tracer.
-
-        This will stop the background writer/worker and flush any finished traces in the buffer. The tracer cannot be
-        used for tracing after this method has been called. A new tracer instance is required to continue tracing.
+        """Shutdown the tracer and flush finished traces. Avoid calling shutdown multiple times.
 
         :param timeout: How long in seconds to wait for the background worker to flush traces
             before exiting or :obj:`None` to block until flushing has successfully completed (default: :obj:`None`)
         :type timeout: :obj:`int` | :obj:`float` | :obj:`None`
         """
-        try:
-            self._writer.stop(timeout=timeout)
-        except service.ServiceStatusError:
-            # It's possible the writer never got started in the first place :(
-            pass
-
         with self._shutdown_lock:
+            # Thread safety: Ensures tracer is shutdown synchronously
+            span_processors = self._span_processors
+            self._span_processors = []
+            for processor in span_processors:
+                if hasattr(processor, "shutdown"):
+                    processor.shutdown(timeout)
+
             atexit.unregister(self._atexit)
             forksafe.unregister(self._child_after_fork)
 

tests/tracer/test_tracer.py

@@ -517,13 +517,13 @@ def test_tracer_url():
 
 def test_tracer_shutdown_no_timeout():
     t = ddtrace.Tracer()
-    writer = mock.Mock(wraps=t._writer)
-    t._writer = writer
 
-    # The writer thread does not start until the first write.
-    t.shutdown()
-    assert t._writer.stop.called
-    assert not t._writer.join.called
+    with mock.patch.object(AgentWriter, "stop") as mock_stop:
+        with mock.patch.object(AgentWriter, "join") as mock_join:
+            t.shutdown()
+
+    mock_stop.assert_called()
+    mock_join.assert_not_called()
 
 
 def test_tracer_configure_writer_stop_unstarted():
@@ -551,13 +551,24 @@ def test_tracer_configure_writer_stop_started():
 
 def test_tracer_shutdown_timeout():
     t = ddtrace.Tracer()
-    t._writer = mock.Mock(wraps=t._writer)
 
-    with t.trace("something"):
-        pass
+    with mock.patch.object(AgentWriter, "stop") as mock_stop:
+        with t.trace("something"):
+            pass
+
+        t.shutdown(timeout=2)
+    mock_stop.assert_called_once_with(2)
+
+
+def test_tracer_shutdown():
+    t = ddtrace.Tracer()
+    t.shutdown()
+
+    with mock.patch.object(AgentWriter, "write") as mock_write:
+        with t.trace("something"):
+            pass
 
-    t.shutdown(timeout=2)
-    t._writer.stop.assert_called_once_with(timeout=2)
+    mock_write.assert_not_called()
 
 
 def test_tracer_dogstatsd_url():