Skip to content

Commit 74c3ab4

Browse files
P403n1x87christophe-papazian
P403n1x87
and
christophe-papazian
authored
refactor: single RC subscriber (#16299)
# Description This PR refactors the Remote Configuration (RC) architecture from a multiple-subscriber model (one thread per product) to a single-subscriber model with product dispatching. This reduces thread overhead and simplifies the RC polling infrastructure. ## Key Changes ### Architectural Simplification - **Before**: Each product registered its own subscriber thread - **After**: Single global subscriber dispatches payloads to registered product callbacks - **Benefit**: Reduced thread count improves resource efficiency and simplifies lifecycle management ### New RCCallback Interface Products now register callbacks by implementing the RCCallback abstract base class: ```python from ddtrace.internal.remoteconfig import RCCallback, Payload from typing import Sequence class MyProductCallback(RCCallback): def __call__(self, payloads: Sequence[Payload]) -> None: """Process configuration payloads when received.""" for payload in payloads: # Process configuration updates ... def periodic(self) -> None: """Optional: Perform periodic operations at every polling cycle.""" # Emit metrics, check stale state, etc. pass ``` ## Key features: - `__call__(payloads)`: Abstract method called when configuration updates are received - `periodic()`: Optional method called at every polling interval (before payload processing), regardless of whether new configuration was received - Enables time-based operations like probe status emission, stale request cleanup, and periodic metrics ### Registration Pattern ```python from ddtrace.internal.remoteconfig.worker import remoteconfig_poller callback = MyProductCallback() remoteconfig_poller.register( product="MY_PRODUCT", callback=callback, capabilities=[...] ) ``` The single subscriber calls: 1. `callback.periodic()` for all registered callbacks at every poll cycle 2. `callback(payloads)` when new configuration is available for that product ## Benefits 1. **Reduced thread overhead**: One subscriber thread instead of N threads (one per product) 2. **Cleaner separation**: Periodic operations separated from payload processing logic 3. **Better testability**: Callbacks can test periodic() independently of payload handling 4. **Consistent interface**: All products follow the same registration pattern ## Migration Notes Existing callbacks automatically get no-op periodic() behavior. No breaking changes for products that don't need periodic operations. # Additional Notes Claude Code made the largest chunk of the changes. It has been guided step-by-step to perform the following tasks - Move to a single subscriber design whereby a single subscriber thread runs in each process. It then dispatches configurations to the appropriate callbacks. - Use the smaller RCCallback for registering product RC callbacks and remove most of the PubSub boilerplate code into the single subscriber. ## Specific changes in appsec - remove the appsec preprocessing part, as it's not required anymore to be outside of the usual appsec callback - update the tests accordingly the appsec RC layer is tested extensively on the system tests APPSEC-61160 Co-authored-by: christophe-papazian <[email protected]> Co-authored-by: gabriele.tornetta <[email protected]>
1 parent 88b4e89 commit 74c3ab4

31 files changed

Lines changed: 1447 additions & 1282 deletions

‎AGENTS.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
3. Ask when unsure about implementation details
2424
4. Update AIDEV- anchors when modifying related code
2525
5. Consider performance impact (this runs in production)
26+
6. Consider architecture (try to use well-established patterns for the problem at hand)
2627

2728
## Initial Setup for AI Assistants
2829

‎ddtrace/appsec/_remoteconfiguration.py‎

Lines changed: 61 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -12,10 +12,7 @@
1212
from ddtrace.internal.logger import get_logger
1313
from ddtrace.internal.remoteconfig import Payload
1414
from ddtrace.internal.remoteconfig import PayloadType
15-
from ddtrace.internal.remoteconfig._connectors import PublisherSubscriberConnector
16-
from ddtrace.internal.remoteconfig._publishers import RemoteConfigPublisher
17-
from ddtrace.internal.remoteconfig._pubsub import PubSub
18-
from ddtrace.internal.remoteconfig._subscribers import RemoteConfigSubscriber
15+
from ddtrace.internal.remoteconfig import RCCallback
1916
from ddtrace.internal.remoteconfig.worker import remoteconfig_poller
2017
from ddtrace.internal.settings.asm import config as asm_config
2118
from ddtrace.internal.telemetry import telemetry_writer
@@ -28,49 +25,32 @@
2825
APPSEC_PRODUCTS = {PRODUCTS.ASM_FEATURES, PRODUCTS.ASM, PRODUCTS.ASM_DATA, PRODUCTS.ASM_DD}
2926

3027

31-
class AppSecRC(PubSub):
32-
__subscriber_class__ = RemoteConfigSubscriber
33-
__publisher_class__ = RemoteConfigPublisher
34-
__shared_data__ = PublisherSubscriberConnector()
35-
36-
def __init__(self, callback):
37-
self._publisher = self.__publisher_class__(
38-
self.__shared_data__, preprocess_func=_preprocess_results_appsec_1click_activation
39-
)
40-
self._subscriber = self.__subscriber_class__(self.__shared_data__, callback, "ASM")
41-
42-
43-
def _forksafe_appsec_rc():
44-
remoteconfig_poller.start_subscribers_by_product(APPSEC_PRODUCTS)
45-
46-
4728
def enable_appsec_rc() -> None:
4829
"""Remote config will be used by ASM libraries to receive four different updates from the backend.
49-
Each update has it’s own product:
30+
Each update has it's own product:
5031
- ASM_FEATURES product - To allow users enable or disable ASM remotely
5132
- ASM product - To allow clients to activate or deactivate rules
5233
- ASM_DD product - To allow the library to receive rules updates
5334
- ASM_DATA product - To allow the library to receive list of blocked IPs and users
5435
5536
If environment variable `DD_APPSEC_ENABLED` is not set, registering ASM_FEATURE can enable ASM remotely.
5637
If it's set to true, we will register the rest of the products.
57-
58-
Parameters `start_subscribers` are needed for testing purposes
5938
"""
6039
log.debug("[%s][P: %s] Register ASM Remote Config Callback", os.getpid(), os.getppid())
61-
asm_callback = (
62-
remoteconfig_poller.get_registered(PRODUCTS.ASM_FEATURES)
63-
or remoteconfig_poller.get_registered(PRODUCTS.ASM)
64-
or AppSecRC(_appsec_callback)
65-
)
6640

6741
if _asm_feature_is_required():
68-
remoteconfig_poller.register(PRODUCTS.ASM_FEATURES, asm_callback, capabilities=[_rc_capabilities()])
42+
remoteconfig_poller.register(
43+
PRODUCTS.ASM_FEATURES,
44+
_appsec_callback,
45+
capabilities=[_rc_capabilities()],
46+
)
6947

48+
# Register other ASM products if AppSec is enabled
7049
if asm_config._asm_enabled and asm_config._asm_static_rule_file is None:
71-
remoteconfig_poller.register(PRODUCTS.ASM_DATA, asm_callback) # IP Blocking
72-
remoteconfig_poller.register(PRODUCTS.ASM, asm_callback) # Exclusion Filters & Custom Rules
73-
remoteconfig_poller.register(PRODUCTS.ASM_DD, asm_callback) # DD Rules
50+
remoteconfig_poller.register(PRODUCTS.ASM_DATA, _appsec_callback) # IP Blocking
51+
remoteconfig_poller.register(PRODUCTS.ASM, _appsec_callback) # Exclusion Filters & Custom Rules
52+
remoteconfig_poller.register(PRODUCTS.ASM_DD, _appsec_callback) # DD Rules
53+
7454
# ensure exploit prevention patches are loaded by one-click activation
7555
if asm_config._asm_enabled:
7656
from ddtrace.appsec._listeners import load_common_appsec_modules
@@ -89,28 +69,56 @@ def disable_appsec_rc():
8969
telemetry_writer.product_activated(TELEMETRY_APM_PRODUCT.APPSEC, False)
9070

9171

92-
def _appsec_callback(payload_list: Sequence[Payload]) -> None:
93-
if not payload_list:
94-
return
95-
debug_info = (
96-
f"appsec._remoteconfiguration.deb::_appsec_callback::payload"
97-
f"{tuple(p.path for p in payload_list)}[{os.getpid()}][P: {os.getppid()}]"
98-
)
99-
log.debug(debug_info)
72+
class AppSecCallback(RCCallback):
73+
"""Remote config callback for AppSec products."""
74+
75+
def __init__(self) -> None:
76+
"""Initialize the AppSec callback."""
77+
self._cache: dict[str, dict[str, Any]] = {}
78+
79+
def __call__(self, payloads: Sequence[Payload]) -> None:
80+
"""Process AppSec configuration payloads.
81+
82+
Args:
83+
payloads: Sequence of configuration payloads to process
84+
"""
85+
if not payloads:
86+
return
87+
result = _update_asm_features(payloads, self._cache)
88+
if "asm" in result:
89+
if asm_config._asm_static_rule_file is None:
90+
if result["asm"].get("enabled", False):
91+
# Register additional ASM products with the same callback
92+
remoteconfig_poller.register(PRODUCTS.ASM_DATA, self) # IP Blocking
93+
remoteconfig_poller.register(PRODUCTS.ASM, self) # Exclusion Filters & Custom Rules
94+
remoteconfig_poller.register(PRODUCTS.ASM_DD, self) # DD Rules
95+
else:
96+
remoteconfig_poller.unregister(PRODUCTS.ASM_DATA)
97+
remoteconfig_poller.unregister(PRODUCTS.ASM)
98+
remoteconfig_poller.unregister(PRODUCTS.ASM_DD)
99+
debug_info = (
100+
f"appsec._remoteconfiguration.deb::_appsec_callback::payload"
101+
f"{tuple(p.path for p in payloads)}[{os.getpid()}][P: {os.getppid()}]"
102+
)
103+
log.debug(debug_info)
104+
105+
for_the_waf_updates: list[tuple[str, str, PayloadType]] = []
106+
for_the_waf_removals: list[tuple[str, str]] = []
107+
for_the_tracer: list[Payload] = []
108+
for payload in payloads:
109+
if payload.metadata.product_name == "ASM_FEATURES":
110+
for_the_tracer.append(payload)
111+
elif payload.content is None:
112+
for_the_waf_removals.append((payload.metadata.product_name, payload.path))
113+
else:
114+
for_the_waf_updates.append((payload.metadata.product_name, payload.path, payload.content))
115+
_process_asm_features(for_the_tracer)
116+
if (for_the_waf_removals or for_the_waf_updates) and asm_config._asm_enabled:
117+
core.dispatch("waf.update", (for_the_waf_removals, for_the_waf_updates))
118+
100119

101-
for_the_waf_updates: list[tuple[str, str, PayloadType]] = []
102-
for_the_waf_removals: list[tuple[str, str]] = []
103-
for_the_tracer: list[Payload] = []
104-
for payload in payload_list:
105-
if payload.metadata.product_name == "ASM_FEATURES":
106-
for_the_tracer.append(payload)
107-
elif payload.content is None:
108-
for_the_waf_removals.append((payload.metadata.product_name, payload.path))
109-
else:
110-
for_the_waf_updates.append((payload.metadata.product_name, payload.path, payload.content))
111-
_process_asm_features(for_the_tracer)
112-
if (for_the_waf_removals or for_the_waf_updates) and asm_config._asm_enabled:
113-
core.dispatch("waf.update", (for_the_waf_removals, for_the_waf_updates))
120+
# Create singleton instance for global usage
121+
_appsec_callback = AppSecCallback()
114122

115123

116124
def _update_asm_features(payload_list: Sequence[Payload], cache: dict[str, dict[str, Any]]) -> dict[str, Any]:
@@ -181,23 +189,3 @@ def enable_asm():
181189
APIManager.enable()
182190
load_appsec()
183191
tracer.configure(appsec_enabled=True, appsec_enabled_origin=APPSEC.ENABLED_ORIGIN_RC)
184-
185-
186-
def _preprocess_results_appsec_1click_activation(
187-
payload_list: list[Payload], pubsub_instance: PubSub, cache: dict[str, dict[str, Any]] = {}
188-
) -> list[Payload]:
189-
"""The main process has the responsibility to enable or disable the ASM products. The child processes don't
190-
care about that, the children only need to know about payload content.
191-
"""
192-
result = _update_asm_features(payload_list, cache)
193-
if "asm" in result:
194-
if asm_config._asm_static_rule_file is None:
195-
if result["asm"].get("enabled", False):
196-
remoteconfig_poller.register(PRODUCTS.ASM_DATA, pubsub_instance) # IP Blocking
197-
remoteconfig_poller.register(PRODUCTS.ASM, pubsub_instance) # Exclusion Filters & Custom Rules
198-
remoteconfig_poller.register(PRODUCTS.ASM_DD, pubsub_instance) # DD Rules
199-
else:
200-
remoteconfig_poller.unregister(PRODUCTS.ASM_DATA)
201-
remoteconfig_poller.unregister(PRODUCTS.ASM)
202-
remoteconfig_poller.unregister(PRODUCTS.ASM_DD)
203-
return payload_list

‎ddtrace/debugging/_debugger.py‎

Lines changed: 14 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -30,9 +30,8 @@
3030
from ddtrace.debugging._probe.model import LineProbe
3131
from ddtrace.debugging._probe.model import Probe
3232
from ddtrace.debugging._probe.registry import ProbeRegistry
33+
from ddtrace.debugging._probe.remoteconfig import DebuggerRCCallback
3334
from ddtrace.debugging._probe.remoteconfig import ProbePollerEvent
34-
from ddtrace.debugging._probe.remoteconfig import ProbePollerEventType
35-
from ddtrace.debugging._probe.remoteconfig import ProbeRCAdapter
3635
from ddtrace.debugging._probe.remoteconfig import build_probe
3736
from ddtrace.debugging._probe.status import ProbeStatusLogger
3837
from ddtrace.debugging._signal.collector import SignalCollector
@@ -189,7 +188,7 @@ class Debugger(Service):
189188
_instance: Optional["Debugger"] = None
190189
_probe_meter = _probe_metrics.get_meter("probe")
191190

192-
__rc_adapter__ = ProbeRCAdapter
191+
__rc_adapter__ = DebuggerRCCallback
193192
__uploader__ = SignalUploader
194193
__watchdog__ = DebuggerModuleWatchdog
195194
__logger__ = ProbeStatusLogger
@@ -235,16 +234,16 @@ def disable(cls, join: bool = True) -> None:
235234

236235
log.debug("Disabling %s", cls.__name__)
237236

238-
adapter = remoteconfig_poller.get_registered("LIVE_DEBUGGING")
237+
callback = remoteconfig_poller.get_registered("LIVE_DEBUGGING")
239238

240239
remoteconfig_poller.unregister("LIVE_DEBUGGING")
241240

242241
# Currently the product enablement and the callback registration are
243242
# tied together within the RC client so here we have to pretend that
244243
# once we have disabled the debugger we also get an empty configuration
245244
# payload from RC.
246-
if adapter is not None:
247-
cast(ProbeRCAdapter, adapter).delete_all_probes()
245+
if callback is not None:
246+
cast(DebuggerRCCallback, callback).delete_all_probes()
248247

249248
unregister_post_run_module_hook(cls._on_run_module)
250249

@@ -288,8 +287,14 @@ def __init__(self, tracer: Optional[Tracer] = None) -> None:
288287
log.info("Disabled Remote Configuration enabled by Dynamic Instrumentation.")
289288

290289
# Register the debugger with the RCM client.
291-
di_callback = self.__rc_adapter__(None, self._on_configuration, status_logger=status_logger)
292-
remoteconfig_poller.register("LIVE_DEBUGGING", di_callback, restart_on_fork=True)
290+
# The callback handles periodic probe status emission internally
291+
di_callback = self.__rc_adapter__(
292+
self._on_configuration,
293+
status_logger,
294+
self._probe_registry,
295+
di_config.diagnostics_interval,
296+
)
297+
remoteconfig_poller.register("LIVE_DEBUGGING", di_callback)
293298

294299
# Load local probes from the probe file.
295300
self._load_local_config()
@@ -579,7 +584,7 @@ def _unwrap_functions(self, probes: list[FunctionProbe]) -> None:
579584
except ValueError:
580585
log.error("Cannot unregister wrapping import hook for module %r", module_name, exc_info=True)
581586

582-
def _on_configuration(self, event: ProbePollerEventType, probes: Iterable[Probe]) -> None:
587+
def _on_configuration(self, event: ProbePollerEvent, probes: Iterable[Probe]) -> None:
583588
log.debug("[%s][P: %s] Received poller event %r with probes %r", os.getpid(), os.getppid(), event, probes)
584589

585590
if event == ProbePollerEvent.STATUS_UPDATE:

0 commit comments

Comments
 (0)