fix(rcm): update the forking behavior in remote config (#7548)

# Context
The Remote Configuration Publisher/Subscriber System
#5464 restarts all pubsub
instances when the application forks (for example, [gunicorn
workers](https://docs.gunicorn.org/en/stable/design.html#server-model),
uwsgi workers, etc.).

![image](https://github.com/DataDog/dd-trace-py/assets/6352942/774fe838-5374-4edc-82c6-51cc563fe66e)

Dynamic Instrumentation needs to update the pubsub instance at this
point because the probe mechanism should run in the child process. For
that, DI needs the callback as the method of an instance of Debugger,
which lives in the child process.

https://github.com/DataDog/dd-trace-py/blob/2.x/ddtrace/debugging/_debugger.py#L276

# Problem description

When the application forks and restarts the subscribers, this happens
before the debugger updates its callback instance.

```
10348 starting subscribers
10348 restarting the debugger
10348 register callback 4429382528 <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.STOPPED: 'stopped'>)> 4406068560 shared_data 4398747792
```

This results in the registration of the callback in the child processes
but the execution of callbacks using the instance of the parent process.

* Parent process: register a callback
[on_configuration](https://github.com/DataDog/dd-trace-py/blob/2.x/ddtrace/debugging/_debugger.py#L654)
for DI

```
94621 register callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.STOPPED: 'stopped'>)> PubSub Isntance id: 4363974784 Debugger._on_configuration id: 4360362576
```

* Child processes: register callbacks
[on_configuration](https://github.com/DataDog/dd-trace-py/blob/2.x/ddtrace/debugging/_debugger.py#L654)
for DI

```
94638 register callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.STOPPED: 'stopped'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id: 4364006016
94639 register callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.STOPPED: 'stopped'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id: 4364006016
94640 register callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.STOPPED: 'stopped'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id: 4364006016
```

* Child processes: exec callbacks.
```
94621 _exec_callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.RUNNING: 'running'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id:  4360362576
94638 _exec_callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.RUNNING: 'running'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id:  4360362576
94639 _exec_callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.RUNNING: 'running'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id:  4360362576
94640 _exec_callback <bound method Debugger._on_configuration of Debugger(status=<ServiceStatus.RUNNING: 'running'>)> PubSub Isntance id: 4392569856 Debugger._on_configuration id:  4360362576
```

As a result, we're registering callback id **4364006016** but calling
**4360362576** (which is the instance of the parent process).

# PR Description
This PR removes the restart of publisher-subscriber instances in Remote
Configuration at fork and delegates it to the products that have
integration with Remote Config, namely AppSec and Dynamic
Instrumentation.

## 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] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/releasenotes.html)
are followed. If no release note is required, add label
`changelog/no-changelog`.
- [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))

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer has explicitly acknowledged and discussed the performance
implications of this PR as reported in the benchmarks PR comment.
- [x] 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)
- [x] If this PR 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`.
- [x] This PR doesn't touch any of that.

---------

Co-authored-by: Gabriele N. Tornetta <[email protected]>
(cherry picked from commit 1da4fc4)

Author: 2 people

ddtrace/appsec/_remoteconfiguration.py

@@ -13,6 +13,7 @@
 from ddtrace.appsec._constants import PRODUCTS
 from ddtrace.appsec._utils import _appsec_rc_features_is_enabled
 from ddtrace.constants import APPSEC_ENV
+from ddtrace.internal import forksafe
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.remoteconfig._connectors import PublisherSubscriberConnector
 from ddtrace.internal.remoteconfig._publishers import RemoteConfigPublisherMergeDicts
@@ -25,6 +26,8 @@
 
 log = get_logger(__name__)
 
+APPSEC_PRODUCTS = [PRODUCTS.ASM_FEATURES, PRODUCTS.ASM, PRODUCTS.ASM_DATA, PRODUCTS.ASM_DD]
+
 
 class AppSecRC(PubSub):
     __subscriber_class__ = RemoteConfigSubscriber
@@ -36,6 +39,10 @@ def __init__(self, _preprocess_results, callback):
         self._subscriber = self.__subscriber_class__(self.__shared_data__, callback, "ASM")
 
 
+def _forksafe_appsec_rc():
+    remoteconfig_poller.start_subscribers_by_product(APPSEC_PRODUCTS)
+
+
 def enable_appsec_rc(test_tracer: Optional[Tracer] = None) -> None:
     """Remote config will be used by ASM libraries to receive four different updates from the backend.
     Each update has it’s own product:
@@ -70,14 +77,13 @@ def enable_appsec_rc(test_tracer: Optional[Tracer] = None) -> None:
         remoteconfig_poller.register(PRODUCTS.ASM, asm_callback)  # Exclusion Filters & Custom Rules
         remoteconfig_poller.register(PRODUCTS.ASM_DD, asm_callback)  # DD Rules
 
+    forksafe.register(_forksafe_appsec_rc)
+
 
 def disable_appsec_rc():
     # only used to avoid data leaks between tests
-
-    remoteconfig_poller.unregister(PRODUCTS.ASM_FEATURES)
-    remoteconfig_poller.unregister(PRODUCTS.ASM_DATA)
-    remoteconfig_poller.unregister(PRODUCTS.ASM)
-    remoteconfig_poller.unregister(PRODUCTS.ASM_DD)
+    for product_name in APPSEC_PRODUCTS:
+        remoteconfig_poller.unregister(product_name)
 
 
 def _add_rules_to_list(features: Mapping[str, Any], feature: str, message: str, ruleset: Dict[str, Any]) -> None:

ddtrace/internal/remoteconfig/client.py

@@ -244,6 +244,13 @@ def update_product_callback(self, product_name, callback):
             return True
         return False
 
+    def start_products(self, products_list):
+        # type: (list) -> None
+        for product_name in products_list:
+            pubsub_instance = self._products.get(product_name)
+            if pubsub_instance:
+                pubsub_instance.restart_subscriber()
+
     def unregister_product(self, product_name):
         # type: (str) -> None
         self._products.pop(product_name, None)

ddtrace/internal/remoteconfig/worker.py

@@ -1,4 +1,5 @@
 import os
+from typing import List
 
 from ddtrace.internal import agent
 from ddtrace.internal import atexit
@@ -80,19 +81,21 @@ def enable(self):
                 return True
 
             self.start()
-            forksafe.register(self.start_subscribers)
+            forksafe.register(self.reset_at_fork)
             atexit.register(self.disable)
             return True
         return False
 
-    def start_subscribers(self):
+    def reset_at_fork(self):
         # type: () -> None
-        """Subscribers need to be restarted when application forks"""
+        """Client Id needs to be refreshed when application forks"""
         self._enable = False
-        log.debug("[%s][P: %s] Remote Config Poller fork. Starting Pubsub services", os.getpid(), os.getppid())
+        log.debug("[%s][P: %s] Remote Config Poller fork. Refreshing state", os.getpid(), os.getppid())
         self._client.renew_id()
-        for pubsub in self._client.get_pubsubs():
-            pubsub.restart_subscriber()
+
+    def start_subscribers_by_product(self, products_list):
+        # type: (List[str]) -> None
+        self._client.start_products(products_list)
 
     def _poll_data(self, test_tracer=None):
         """Force subscribers to poll new data. This function is only used in tests"""
@@ -121,7 +124,7 @@ def disable(self, join=False):
         if self.status == ServiceStatus.STOPPED:
             return
 
-        forksafe.unregister(self.start_subscribers)
+        forksafe.unregister(self.reset_at_fork)
         atexit.unregister(self.disable)
 
         self.stop(join=join)
@@ -147,14 +150,12 @@ def register(self, product, pubsub_instance, skip_enabled=False):
         try:
             # By enabling on registration we ensure we start the RCM client only
             # if there is at least one registered product.
-            enabled = True
             if not skip_enabled:
-                enabled = self.enable()
+                self.enable()
 
-            if enabled:
-                self._client.register_product(product, pubsub_instance)
-                if not self._client.is_subscriber_running(pubsub_instance):
-                    pubsub_instance.start_subscriber()
+            self._client.register_product(product, pubsub_instance)
+            if not self._client.is_subscriber_running(pubsub_instance):
+                pubsub_instance.start_subscriber()
         except Exception:
             log.debug("error starting the RCM client", exc_info=True)
 

releasenotes/notes/fix-debugger-rc-instance-954539a2e57b7887.yaml

@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    dynamic instrumentation: Needs to update the pubsub instance when the application forks because the probe 
+    mechanism should run in the child process. For that, DI needs the callback as the method of an instance of 
+    Debugger, which lives in the child process.

tests/internal/remoteconfig/test_remoteconfig.py

@@ -146,7 +146,7 @@ def get_mock_encoded_msg_with_signed_errors(msg, path, signed_errors):
 
 def test_remote_config_register_auto_enable(remote_config_worker):
     # ASM_FEATURES product is enabled by default, but LIVE_DEBUGGER isn't
-    class MockPubsub:
+    class MockPubsub(PubSub):
         def stop(self, *args, **kwargs):
             pass
 
@@ -165,10 +165,15 @@ def stop(self, *args, **kwargs):
 
 def test_remote_config_register_validate_rc_disabled(remote_config_worker):
     remoteconfig_poller.disable()
+
+    class MockPubsub(PubSub):
+        def stop(self, *args, **kwargs):
+            pass
+
     assert remoteconfig_poller.status == ServiceStatus.STOPPED
 
     with override_global_config(dict(_remote_config_enabled=False)):
-        remoteconfig_poller.register("LIVE_DEBUGGER", lambda m, c: None)
+        remoteconfig_poller.register("LIVE_DEBUGGER", MockPubsub())
 
         assert remoteconfig_poller.status == ServiceStatus.STOPPED