chore(rcm): Remote Config refactor, publisher subscriber system

[avara1986]

Remote Configuration Publisher-Subscriber system.

Motivation

A common Python web application uses a WSGI server (e.g., Gunicorn) that employs multiple workers. For each worker, a Remote Config instance polls new configuration from the Agent. In the context of ASM, the typical scenario follows this flow:

• Remote Config enables the ASM_FEATURES product (with 1 click activation). The RC starts polling.
• RC receives a payload to enable ASM_FEATURES. RC activates ASM and registers new products (ASM_DD, ASM_DATA, ASM_DD).
• RC receives a payload with all products and updates the WAF configuration.

This flow takes around 10-15 seconds (i.e., with a polling interval of 5s, RC needs 2-3 requests to the agent to keep all information updated).

However, there is a problem with the Gunicorn application: sometimes, Gunicorn may restart a worker, which needs to retrieve all RC data again and repeat the aforementioned flow. During this interval, ASM with Remote Config could raise unexpected behaviors, the new worker wouldn’t block it since it may not have yet applied all the new configuration.

Example

The following example runs a Gunicorn application.

ddtrace-run gunicorn -w 4 app:app

We use Locust to request the /block endpoint with a blocked IP. Then, we kill a Gunicorn child process. As a result, we have the following table:

/block: 19495 (request) 19325 (request blocked)

The 19495 > 19325 means we have 170 requests that return 200 instead of 403 (blocked).

Description

Remote Configuration needs to keep all workers updated as soon as possible. Therefore, Remote Configuration may start BEFORE the Gunicorn server in sitecustomize.py, it starts to poll information from the RC Agent, and for each new payload, through this Pub-sub system, share this information with all child processes.

In addition to this, there are different Remote Configuration behaviors:

• When the Remote Configuration Client receives a new product target file payload, we need to call a callback.
• When the Remote Configuration Client receives a new product target file payload, we need to aggregate this target file data for each product. After that, call the callback with all aggregated information.
• Remote Configuration may have a callback for each product.
• Remote Configuration may have a callback for one or more products.
• For each payload, Remote Configuration needs to execute specific actions on the main process and a different action on child processes.

To achieve this goal, a Remote Configuration product may register a PubSub instance. A PubSub class contains a publisher that receives the Remote Configuration payload and shares it with Pubsub Subscriber instance. The Subscriber starts a thread on each child process, waiting for a new update of the shared data between the Publisher on the main process and the child process. Remote Configuration creates a thread listening to the main process for each instance of PubSub. To connect this publisher and the child processes subscribers, we need a connector class: Shared Memory or File. Each instance of PubSub works as a singleton when Remote Configuration dispatches the callbacks. That means if we register the same instance of PubSub class on different products, we would have one thread waiting to the main process.

Each DD Product (APM, ASM, DI, CI) may implement its PubSub Class.

Example 1: A callback for one or more Remote Configuration Products

AppSec needs to aggregate different products in the same callback for all child processes.

class AppSecRC(PubSubMergeFirst):
    __shared_data = ConnectorSharedMemory()

    def __init__(self, _preprocess_results, callback, name="Default"):
        self._publisher = self.__publisher_class__(self.__shared_data, _preprocess_results)
        self._subscriber = self.__subscriber_class__(self.__shared_data, callback, name)

asm_callback = AppSecRC(preprocess_1click_activation, appsec_callback, "ASM")

remoteconfig_poller.register("ASM_PRODUCT", asm_callback)
remoteconfig_poller.register("ASM_FEATURES_PRODUCT", asm_callback)

Example 2: One Callback for each product
DI needs to aggregate different products in the same callback for all child processes.

class DynamicInstrumentationRC(PubSub):
    __shared_data = ConnectorSharedMemory()

    def __init__(self, _preprocess_results, callback, name="Default"):
        self._publisher = self.__publisher_class__(self.__shared_data, _preprocess_results)
        self._subscriber = self.__subscriber_class__(self.__shared_data, callback, name)


di_callback_1 = DynamicInstrumentationRC(callback=di_callback_1, name="ASM")
di_callback_2 = DynamicInstrumentationRC(callback=di_callback_2, name="ASM")

remoteconfig_poller.register("DI_1_PRODUCT", di_callback)
remoteconfig_poller.register("DI_2_PRODUCT", di_callback_2)

Results

Following the previous example, if we run a application with gunicorn and we kill some Gunicorn child worker, the result is:

/block: 15070 (request) 15070 (request blocked)

19495 == 19325 Hurray!

Extra improvements of this refactor

if we compare with docker stats both branches (1.x and this branch) we see an improvement of memory usage and CPU:

1.x branch:

CONTAINER ID   NAME                        CPU %     MEM USAGE / LIMIT     MEM %     NET I/O           BLOCK I/O    PIDS
e487b87bc5e0   showcase-flask_showcase_1   34.23%    204.1MiB / 31.07GiB   0.64%     44.8MB / 62.3MB   0B / 451kB   17

avara1986/rc_refactor branch:

CONTAINER ID   NAME                        CPU %     MEM USAGE / LIMIT     MEM %     NET I/O           BLOCK I/O        PIDS
b90ac73a23b6   showcase-flask_showcase_1   14.99%    169.2MiB / 31.07GiB   0.53%     31.3MB / 55.6MB   4.36MB / 303kB   18

PR related

This PR fix this system test problem: DataDog/system-tests#932

Checklist

• Change(s) are motivated and described in the PR description.
• Testing strategy is described if automated tests are not included in the PR.
• Risk is outlined (performance impact, potential for breakage, maintainability, etc).
• Change is maintainable (easy to change, telemetry, documentation).
• Library release note guidelines are followed.
• Documentation is included (in-code, generated user docs, public corp docs).
• PR description includes explicit acknowledgement/acceptance of the performance implications of this PR as reported in the benchmarks PR comment.

Reviewer Checklist

• Title is accurate.
• No unnecessary changes are introduced.
• Description motivates each change.
• Avoids breaking API changes unless absolutely necessary.
• Testing strategy adequately addresses listed risk(s).
• Change is maintainable (easy to change, telemetry, documentation).
• Release note makes sense to a user of the library.
• Reviewer has explicitly acknowledged and discussed the performance implications of this PR as reported in the benchmarks PR comment.

[pr-commenter]

Benchmarks

Comparing candidate commit 1ffeb0b in PR branch avara1986/rc_refactor with baseline commit 8e8f196 in branch 1.x.

Found 37 performance improvements and 1 performance regressions! Performance is the same for 52 cases.

scenario:flasksimple-baseline

• 🟩 max_rss_usage [-0.978MB; -0.838MB] or [-2.986%; -2.557%]

scenario:flasksimple-tracer

• 🟩 max_rss_usage [-0.995MB; -0.844MB] or [-3.030%; -2.569%]

scenario:flasksimple-profiler

• 🟩 max_rss_usage [-1.027MB; -0.857MB] or [-3.133%; -2.613%]

scenario:flasksimple-iast-get

• 🟩 max_rss_usage [-1.079MB; -0.938MB] or [-3.287%; -2.858%]

scenario:flasksimple-appsec-get

• 🟩 max_rss_usage [-1.025MB; -0.891MB] or [-3.129%; -2.720%]

scenario:flasksimple-appsec-post

• 🟩 max_rss_usage [-1.066MB; -0.910MB] or [-3.244%; -2.771%]

scenario:flasksimple-appsec-telemetry

• 🟩 max_rss_usage [-1.105MB; -0.945MB] or [-3.371%; -2.882%]

scenario:otelspan-start-finish

• 🟩 max_rss_usage [-1.689MB; -1.570MB] or [-5.576%; -5.183%]

scenario:samplingrules-high_match

• 🟩 max_rss_usage [-1.800MB; -1.684MB] or [-6.198%; -5.797%]

scenario:samplingrules-average_match

• 🟩 max_rss_usage [-1.710MB; -1.595MB] or [-5.901%; -5.506%]

scenario:samplingrules-low_match

• 🟩 max_rss_usage [-1.691MB; -1.567MB] or [-5.784%; -5.360%]

scenario:samplingrules-very_low_match

• 🟩 max_rss_usage [-1.852MB; -1.743MB] or [-4.186%; -3.939%]

scenario:sethttpmeta-all-enabled

• 🟩 max_rss_usage [-2.082MB; -1.921MB] or [-7.020%; -6.480%]

scenario:sethttpmeta-no-useragentvariant

• 🟩 max_rss_usage [-2.094MB; -1.940MB] or [-7.059%; -6.540%]

scenario:sethttpmeta-all-disabled

• 🟩 max_rss_usage [-1.993MB; -1.840MB] or [-6.731%; -6.214%]

scenario:sethttpmeta-useragentvariant_exists_1

• 🟩 max_rss_usage [-2.001MB; -1.836MB] or [-6.762%; -6.205%]

scenario:sethttpmeta-useragentvariant_exists_2

• 🟩 max_rss_usage [-2.034MB; -1.850MB] or [-6.866%; -6.244%]

scenario:sethttpmeta-useragentvariant_exists_3

• 🟩 max_rss_usage [-2.082MB; -1.935MB] or [-7.016%; -6.522%]

scenario:sethttpmeta-useragentvariant_not_exists_1

• 🟩 max_rss_usage [-2.046MB; -1.873MB] or [-6.906%; -6.321%]

scenario:sethttpmeta-useragentvariant_not_exists_2

• 🟩 max_rss_usage [-2.105MB; -1.939MB] or [-7.090%; -6.531%]

scenario:sethttpmeta-no-collectipvariant

• 🟩 max_rss_usage [-2.039MB; -1.900MB] or [-6.883%; -6.412%]

scenario:sethttpmeta-collectipvariant_exists

• 🟩 max_rss_usage [-2.028MB; -1.912MB] or [-6.832%; -6.441%]

scenario:sethttpmeta-obfuscation-worst-case-implicit-query

• 🟩 max_rss_usage [-1.482MB; -1.341MB] or [-5.049%; -4.571%]

scenario:sethttpmeta-obfuscation-worst-case-explicit-query

• 🟩 max_rss_usage [-2.074MB; -1.930MB] or [-6.952%; -6.467%]

scenario:sethttpmeta-obfuscation-regular-case-implicit-query

• 🟩 max_rss_usage [-2.075MB; -1.918MB] or [-6.958%; -6.430%]

scenario:sethttpmeta-obfuscation-regular-case-explicit-query

• 🟩 max_rss_usage [-1.982MB; -1.835MB] or [-6.652%; -6.159%]

scenario:sethttpmeta-obfuscation-no-query

• 🟩 max_rss_usage [-1.991MB; -1.828MB] or [-6.731%; -6.178%]

scenario:sethttpmeta-obfuscation-send-querystring-disabled

• 🟩 max_rss_usage [-1.558MB; -1.384MB] or [-5.310%; -4.717%]

scenario:sethttpmeta-obfuscation-disabled

• 🟩 max_rss_usage [-2.026MB; -1.858MB] or [-6.827%; -6.263%]

scenario:span-start

• 🟩 max_rss_usage [-1.301MB; -1.112MB] or [-2.678%; -2.289%]

scenario:span-start-traceid128

• 🟩 max_rss_usage [-1.178MB; -1.019MB] or [-2.424%; -2.096%]

scenario:span-add-tags

• 🟥 max_rss_usage [+2.846MB; +3.010MB] or [+6.820%; +7.211%]

scenario:span-add-metrics

• 🟩 max_rss_usage [-4.393MB; -4.230MB] or [-6.974%; -6.716%]

scenario:span-start-finish

• 🟩 max_rss_usage [-1.757MB; -1.600MB] or [-6.031%; -5.492%]

scenario:span-start-finish-traceid128

• 🟩 max_rss_usage [-1.822MB; -1.649MB] or [-6.241%; -5.647%]

scenario:tracer-small

• 🟩 max_rss_usage [-1.859MB; -1.699MB] or [-6.380%; -5.831%]

scenario:tracer-medium

• 🟩 max_rss_usage [-1.825MB; -1.669MB] or [-6.275%; -5.741%]

scenario:tracer-large

• 🟩 max_rss_usage [-1.945MB; -1.759MB] or [-6.494%; -5.873%]