chore(debugging): support probe mutability (#5389)

This change adds support for mutable probes to Dynamic Instrumentation.
This means that the DI agent can now update installed probes to match
whatever changes have been applied to an existing probe from the Dynamic
Instrumentation UI.

## Implementation details

Only certain probe fields are currently allowed to be mutated. We
leverage the annotation capabilities of `attrs` to mark such fields with
`eq=True` (in fact, we mark _immutable_ fields with `eq=False`). An
exception is the `probe_id` field which, whilst marked as mutable, it is
actually used for primary equality detection.

## Testing strategy

To comply with the backing RFC, the tests have been extended to check
that:

- probes are indeed mutable in the intended way
- probe status are re-emitted when a probe is mutated

Author: P403n1x87

ddtrace/debugging/_debugger.py

@@ -611,6 +611,7 @@ def _on_configuration(self, event, probes):
                         # We didn't have the probe. This shouldn't have happened!
                         log.error("Modified probe %r was not found in registry.", probe)
                         continue
+                    self._probe_registry.update(probe)
 
             return
 

ddtrace/debugging/_encoding.py

@@ -103,6 +103,7 @@ def _probe_details(probe):
         }
     return {
         "id": probe.probe_id,
+        "version": probe.version,
         "location": location,
     }
 

ddtrace/debugging/_probe/model.py

@@ -66,9 +66,10 @@ class CaptureLimits(object):
     max_fields = attr.ib(type=int, default=MAXFIELDS)  # type: int
 
 
-@attr.s(hash=True)
+@attr.s
 class Probe(six.with_metaclass(abc.ABCMeta)):
     probe_id = attr.ib(type=str)
+    version = attr.ib(type=int)
     tags = attr.ib(type=dict, eq=False)
     rate = attr.ib(type=float, eq=False)
     limiter = attr.ib(type=RateLimiter, init=False, repr=False, eq=False)  # type: RateLimiter
@@ -83,6 +84,22 @@ def _(self):
             raise_on_exceed=False,
         )
 
+    def update(self, other):
+        # type: (Probe) -> None
+        """Update the mutable fields from another probe."""
+        if self.probe_id != other.probe_id:
+            log.error("Probe ID mismatch when updating mutable fields")
+            return
+
+        if self.version == other.version:
+            return
+
+        for attrib in (_.name for _ in self.__attrs_attrs__ if _.eq):
+            setattr(self, attrib, getattr(other, attrib))
+
+    def __hash__(self):
+        return hash(self.probe_id)
+
 
 @attr.s
 class ProbeConditionMixin(object):
@@ -119,8 +136,8 @@ def location(self):
 
 @attr.s
 class LineLocationMixin(ProbeLocationMixin):
-    source_file = attr.ib(type=str, converter=_resolve_source_file)  # type: ignore[misc]
-    line = attr.ib(type=int)
+    source_file = attr.ib(type=str, converter=_resolve_source_file, eq=False)  # type: ignore[misc]
+    line = attr.ib(type=int, eq=False)
 
     def location(self):
         return (self.source_file, self.line)
@@ -135,8 +152,8 @@ class ProbeEvaluateTimingForMethod(object):
 
 @attr.s
 class FunctionLocationMixin(ProbeLocationMixin):
-    module = attr.ib(type=str)
-    func_qname = attr.ib(type=str)
+    module = attr.ib(type=str, eq=False)
+    func_qname = attr.ib(type=str, eq=False)
     evaluate_at = attr.ib(type=ProbeEvaluateTimingForMethod)
 
     def location(self):

ddtrace/debugging/_probe/registry.py

@@ -38,6 +38,10 @@ def set_message(self, message):
         # type: (str) -> None
         self.message = message
 
+    def update(self, probe):
+        # type: (Probe) -> None
+        self.probe.update(probe)
+
 
 def _get_probe_location(probe):
     # type: (Probe) -> Optional[str]
@@ -86,6 +90,16 @@ def register(self, *probes):
 
                 self.logger.received(probe)
 
+    def update(self, probe):
+        with self._lock:
+            if probe not in self:
+                logger.error("Attempted to update unregistered probe %s", probe.probe_id)
+                return
+
+            self[probe.probe_id].update(probe)
+
+            self.log_probe_status(probe)
+
     def set_installed(self, probe):
         # type: (Probe) -> None
         """Set the installed flag for a probe."""
@@ -111,19 +125,29 @@ def set_error(self, probe, message):
             self[probe.probe_id].set_message(message)
             self.logger.error(probe, message)
 
+    def _log_probe_status_unlocked(self, entry):
+        # type: (ProbeRegistryEntry) -> None
+        if entry.installed:
+            self.logger.installed(entry.probe)
+        elif entry.exc_info:
+            self.logger.error(entry.probe, exc_info=entry.exc_info)
+        elif entry.message:
+            self.logger.error(entry.probe, message=entry.message)
+        else:
+            self.logger.received(entry.probe)
+
+    def log_probe_status(self, probe):
+        # type: (Probe) -> None
+        """Log the status of a probe using the status logger."""
+        with self._lock:
+            self._log_probe_status_unlocked(self[probe.probe_id])
+
     def log_probes_status(self):
         # type: () -> None
         """Log the status of all the probes using the status logger."""
         with self._lock:
             for entry in self.values():
-                if entry.installed:
-                    self.logger.installed(entry.probe)
-                elif entry.exc_info:
-                    self.logger.error(entry.probe, exc_info=entry.exc_info)
-                elif entry.message:
-                    self.logger.error(entry.probe, message=entry.message)
-                else:
-                    self.logger.received(entry.probe)
+                self._log_probe_status_unlocked(entry)
 
     def _remove_pending(self, probe):
         # type: (Probe) -> None

ddtrace/debugging/_probe/remoteconfig.py

@@ -148,6 +148,7 @@ def probe_factory(attribs):
 
         args = dict(
             probe_id=_id,
+            version=attribs.get("version", 0),
             condition=_compile_expression(attribs.get("when")),
             tags=dict(_.split(":", 1) for _ in attribs.get("tags", [])),
             rate=rate,
@@ -175,6 +176,7 @@ def probe_factory(attribs):
     elif _type == ProbeType.METRIC_PROBE:
         args = dict(
             probe_id=_id,
+            version=attribs.get("version", 0),
             condition=_compile_expression(attribs.get("when")),
             tags=dict(_.split(":", 1) for _ in attribs.get("tags", [])),
             name=attribs["metricName"],
@@ -189,6 +191,7 @@ def probe_factory(attribs):
     elif _type == ProbeType.SPAN_PROBE:
         args = dict(
             probe_id=_id,
+            version=attribs.get("version", 0),
             condition=_compile_expression(attribs.get("when")),
             tags=dict(_.split(":", 1) for _ in attribs.get("tags", [])),
             rate=DEFAULT_PROBE_RATE,  # TODO: should we take rate limit out of Probe?
@@ -237,9 +240,10 @@ def _next_status_update_timestamp(self):
         self._status_timestamp = time.time() + config.diagnostics_interval
 
     def _dispatch_probe_events(self, prev_probes, next_probes):
+        # type: (Dict[str, Probe], Dict[str, Probe]) -> None
         new_probes = [p for _, p in next_probes.items() if _ not in prev_probes]
         deleted_probes = [p for _, p in prev_probes.items() if _ not in next_probes]
-        modified_probes = []  # DEV: Probes are currently immutable
+        modified_probes = [p for _, p in next_probes.items() if _ in prev_probes and p != prev_probes[_]]
 
         if deleted_probes:
             self._callback(ProbePollerEvent.DELETED_PROBES, deleted_probes)

ddtrace/debugging/_probe/status.py

@@ -36,6 +36,7 @@ def _payload(self, probe, status, message, timestamp, exc_info=None):
             "debugger": {
                 "diagnostics": {
                     "probeId": probe.probe_id,
+                    "probeVersion": probe.version,
                     "status": status,
                 }
             },

tests/debugging/mocking.py

@@ -82,6 +82,10 @@ def remove_probes(self, *probes):
         # type: (Probe) -> None
         self._on_configuration(ProbePollerEvent.DELETED_PROBES, probes)
 
+    def modify_probes(self, *probes):
+        # type: (Probe) -> None
+        self._on_configuration(ProbePollerEvent.MODIFIED_PROBES, probes)
+
     @property
     def test_queue(self):
         return self._collector.test_queue

tests/debugging/probe/test_model.py

@@ -3,7 +3,10 @@
 from os.path import relpath
 from os.path import sep
 
+from ddtrace.debugging._expressions import DDExpression
+from ddtrace.debugging._expressions import dd_compile
 from ddtrace.debugging._probe.model import _resolve_source_file
+from tests.debugging.utils import create_log_line_probe
 
 
 def test_resolve_source_file():
@@ -16,3 +19,30 @@ def test_resolve_source_file():
     # Test that we fail if we have incomplete source paths
     _, _, child = rpath.partition(sep)
     assert _resolve_source_file(child) is None
+
+
+def test_mutability():
+    before = create_log_line_probe(
+        probe_id="test_mutability",
+        version=1,
+        condition=None,
+        source_file="foo",
+        line=1,
+        template="",
+        segments=[],
+    )
+    after = create_log_line_probe(
+        probe_id="test_mutability",
+        version=2,
+        condition=DDExpression(dsl="True", callable=dd_compile(True)),
+        source_file="foo",
+        line=1,
+        template="",
+        segments=[],
+    )
+
+    assert before != after
+
+    before.update(after)
+
+    assert before == after

tests/debugging/probe/test_registry.py

@@ -63,6 +63,6 @@ def test_registry_location_error():
             "service": "test",
             "message": "Unable to resolve location information for probe 42",
             "ddsource": "dd_debugger",
-            "debugger": {"diagnostics": {"probeId": 42, "status": "ERROR"}},
+            "debugger": {"diagnostics": {"probeId": 42, "probeVersion": 0, "status": "ERROR"}},
         }
     ]

tests/debugging/probe/test_remoteconfig.py

@@ -205,6 +205,7 @@ def validate_events(expected):
             config_metadata("spanProbe_probe1"),
             {
                 "id": "probe1",
+                "version": 0,
                 "type": ProbeType.SPAN_PROBE,
                 "active": True,
                 "tags": ["foo:bar"],
@@ -223,6 +224,7 @@ def validate_events(expected):
             config_metadata("metricProbe_probe2"),
             {
                 "id": "probe2",
+                "version": 1,
                 "type": ProbeType.METRIC_PROBE,
                 "tags": ["foo:bar"],
                 "where": {"sourceFile": "tests/submod/stuff.p", "lines": ["36"]},
@@ -241,6 +243,7 @@ def validate_events(expected):
             config_metadata("logProbe_probe3"),
             {
                 "id": "probe3",
+                "version": 1,
                 "type": ProbeType.LOG_PROBE,
                 "tags": ["foo:bar"],
                 "where": {"sourceFile": "tests/submod/stuff.p", "lines": ["36"]},
@@ -359,3 +362,57 @@ def test_parse_log_probe_default_rates():
     )
 
     assert probe.rate == DEFAULT_PROBE_RATE
+
+
+def test_modified_probe_events(mock_config):
+    events = []
+
+    def cb(e, ps):
+        events.append((e, frozenset([p.probe_id if isinstance(p, Probe) else p for p in ps])))
+
+    mock_config.add_probes(
+        [
+            create_snapshot_line_probe(
+                probe_id="probe1",
+                version=1,
+                source_file="tests/debugger/submod/stuff.py",
+                line=36,
+                condition=None,
+            ),
+        ]
+    )
+
+    metadata = config_metadata()
+    old_interval = config.diagnostics_interval
+    config.diagnostics_interval = 0.5
+    try:
+        adapter = ProbeRCAdapter(cb)
+        # Wait to allow the next call to the adapter to generate a status event
+        sleep(0.5)
+        adapter(metadata, {})
+
+        mock_config.add_probes(
+            [
+                create_snapshot_line_probe(
+                    probe_id="probe1",
+                    version=2,
+                    source_file="tests/debugger/submod/stuff.py",
+                    line=36,
+                    condition=None,
+                )
+            ]
+        )
+        adapter(metadata, {})
+
+        # Wait to allow the next call to the adapter to generate a status event
+        sleep(0.5)
+        adapter(metadata, {})
+
+        assert events == [
+            (ProbePollerEvent.STATUS_UPDATE, frozenset()),
+            (ProbePollerEvent.NEW_PROBES, frozenset(["probe1"])),
+            (ProbePollerEvent.MODIFIED_PROBES, frozenset(["probe1"])),
+            (ProbePollerEvent.STATUS_UPDATE, frozenset(["probe1"])),
+        ]
+    finally:
+        config.diagnostics_interval = old_interval