chore(di): implement EMITTING probe status (#8188)

We implement the `EMITTING` probe status that is logged whenever a probe
sends a signal for the first time. The message is then replicated
periodically, like the other status messages.

## Checklist

- [x] Change(s) are motivated and described in the PR description
- [x] Testing strategy is described if automated tests are not included
in the PR
- [x] Risks are described (performance impact, potential for breakage,
maintainability)
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed or label `changelog/no-changelog` is set
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/))
- [x] Backport labels are set (if
[applicable](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting))
- [x] If this PR changes the public interface, I've notified
`@DataDog/apm-tees`.
- [x] If change touches code that signs or publishes builds or packages,
or handles credentials of any kind, I've requested a review from
`@DataDog/security-design-and-guidance`.

## Reviewer Checklist

- [ ] Title is accurate
- [ ] All changes are related to the pull request's stated goal
- [ ] Description motivates each change
- [ ] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- [ ] Testing strategy adequately addresses listed risks
- [ ] Change is maintainable (easy to change, telemetry, documentation)
- [ ] Release note makes sense to a user of the library
- [ ] Author has acknowledged and discussed the performance implications
of this PR as reported in the benchmarks PR comment
- [ ] Backport labels are set in a manner that is consistent with the
[release branch maintenance
policy](https://ddtrace.readthedocs.io/en/latest/contributing.html#backporting)

ddtrace/debugging/_debugger.py

@@ -52,6 +52,7 @@
 from ddtrace.debugging._signal.collector import SignalContext
 from ddtrace.debugging._signal.metric_sample import MetricSample
 from ddtrace.debugging._signal.model import Signal
+from ddtrace.debugging._signal.model import SignalState
 from ddtrace.debugging._signal.snapshot import Snapshot
 from ddtrace.debugging._signal.tracing import DynamicSpan
 from ddtrace.debugging._signal.tracing import SpanDecoration
@@ -322,6 +323,9 @@ def _dd_debugger_hook(self, probe: Probe) -> None:
             log.debug("[%s][P: %s] Debugger. Report signal %s", os.getpid(), os.getppid(), signal)
             self._collector.push(signal)
 
+            if signal.state is SignalState.DONE:
+                self._probe_registry.set_emitting(probe)
+
         except Exception:
             log.error("Failed to execute probe hook", exc_info=True)
 
@@ -421,6 +425,9 @@ def _(wrapped: FunctionType, args: Tuple[Any], kwargs: Dict[str, Any]) -> Any:
 
             for context in open_contexts:
                 context.exit(retval, exc_info, end_time - start_time)
+                signal = context.signal
+                if signal.state is SignalState.DONE:
+                    self._probe_registry.set_emitting(signal.probe)
 
             exc = exc_info[1]
             if exc is not None:

ddtrace/debugging/_probe/registry.py

@@ -3,6 +3,7 @@
 from typing import Dict
 from typing import List
 from typing import Optional
+from typing import cast
 
 from ddtrace.debugging._probe.model import Probe
 from ddtrace.debugging._probe.model import ProbeLocationMixin
@@ -15,17 +16,27 @@
 
 
 class ProbeRegistryEntry(object):
-    __slots__ = ("probe", "installed", "error_type", "message")
+    __slots__ = (
+        "probe",
+        "installed",
+        "emitting",
+        "error_type",
+        "message",
+    )
 
     def __init__(self, probe: Probe) -> None:
         self.probe = probe
         self.installed = False
+        self.emitting = False
         self.error_type: Optional[str] = None
         self.message: Optional[str] = None
 
     def set_installed(self) -> None:
         self.installed = True
 
+    def set_emitting(self) -> None:
+        self.emitting = True
+
     def set_error(self, error_type: str, message: str) -> None:
         self.error_type = error_type
         self.message = message
@@ -102,14 +113,24 @@ def set_installed(self, probe: Probe) -> None:
 
             self.logger.installed(probe)
 
+    def set_emitting(self, probe: Probe) -> None:
+        """Set the emitting flag for a probe."""
+        with self._lock:
+            entry = cast(ProbeRegistryEntry, self[probe.probe_id])
+            if not entry.emitting:
+                entry.set_emitting()
+                self.logger.emitting(probe)
+
     def set_error(self, probe: Probe, error_type: str, message: str) -> None:
         """Set the error message for a probe."""
         with self._lock:
             self[probe.probe_id].set_error(error_type, message)
             self.logger.error(probe, (error_type, message))
 
     def _log_probe_status_unlocked(self, entry: ProbeRegistryEntry) -> None:
-        if entry.installed:
+        if entry.emitting:
+            self.logger.emitting(entry.probe)
+        elif entry.installed:
             self.logger.installed(entry.probe)
         elif entry.error_type:
             assert entry.message is not None, entry  # nosec

ddtrace/debugging/_probe/status.py

@@ -136,5 +136,12 @@ def installed(self, probe: Probe, message: t.Optional[str] = None) -> None:
             message or "Probe %s instrumented correctly" % probe.probe_id,
         )
 
+    def emitting(self, probe: Probe, message: t.Optional[str] = None) -> None:
+        self._enqueue(
+            probe,
+            "EMITTING",
+            message or "Probe %s is emitting data" % probe.probe_id,
+        )
+
     def error(self, probe: Probe, error: t.Optional[ErrorInfo] = None) -> None:
         self._enqueue(probe, "ERROR", "Failed to instrument probe %s" % probe.probe_id, error)

tests/debugging/exploration/debugger.py

@@ -171,6 +171,9 @@ def received(self, *args, **kwargs):
     def installed(self, *args, **kwargs):
         pass
 
+    def emitting(self, *args, **kwargs):
+        pass
+
     def error(self, *args, **kwargs):
         pass
 

tests/debugging/test_debugger.py

@@ -637,7 +637,7 @@ def test_debugger_line_probe_on_wrapped_function(stuff):
             assert snapshot.probe.probe_id == "line-probe-wrapped-method"
 
 
-def test_probe_status_logging(remote_config_worker):
+def test_probe_status_logging(remote_config_worker, stuff):
     assert remoteconfig_poller.status == ServiceStatus.STOPPED
 
     with rcm_endpoint(), debugger(diagnostics_interval=float("inf"), enabled=True) as d:
@@ -647,6 +647,7 @@ def test_probe_status_logging(remote_config_worker):
                 source_file="tests/submod/stuff.py",
                 line=36,
                 condition=None,
+                rate=float("inf"),
             ),
             create_snapshot_function_probe(
                 probe_id="line-probe-error",
@@ -656,18 +657,22 @@ def test_probe_status_logging(remote_config_worker):
             ),
         )
 
+        # Call the function multiple times to ensure that we emit only once.
+        for _ in range(10):
+            stuff.Stuff().instancestuff(42)
+
         logger = d.probe_status_logger
 
         def count_status(queue):
             return Counter(_["debugger"]["diagnostics"]["status"] for _ in queue)
 
-        logger.wait(lambda q: count_status(q) == {"INSTALLED": 1, "RECEIVED": 2, "ERROR": 1})
+        logger.wait(lambda q: count_status(q) == {"INSTALLED": 1, "RECEIVED": 2, "ERROR": 1, "EMITTING": 1})
 
         d.log_probe_status()
-        logger.wait(lambda q: count_status(q) == {"INSTALLED": 2, "RECEIVED": 2, "ERROR": 2})
+        logger.wait(lambda q: count_status(q) == {"INSTALLED": 1, "RECEIVED": 2, "ERROR": 2, "EMITTING": 2})
 
         d.log_probe_status()
-        logger.wait(lambda q: count_status(q) == {"INSTALLED": 3, "RECEIVED": 2, "ERROR": 3})
+        logger.wait(lambda q: count_status(q) == {"INSTALLED": 1, "RECEIVED": 2, "ERROR": 3, "EMITTING": 3})
 
 
 def test_probe_status_logging_reemit_on_modify(remote_config_worker):