chore(asm): use sample delay env var for api sec sampling (DataDog#8578)

This PR implements last addition to RFC "API Security Sampling Algorithm
RFC", allowing sampling delay to be set by a private, non documented
environment variable.

## 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

- [x] Title is accurate
- [x] All changes are related to the pull request's stated goal
- [x] Description motivates each change
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes
- [x] Testing strategy adequately addresses listed risks
- [x] Change is maintainable (easy to change, telemetry, documentation)
- [x] Release note makes sense to a user of the library
- [x] Author has 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)

---------

Co-authored-by: Federico Mon <[email protected]>

ddtrace/appsec/_api_security/api_manager.py

@@ -23,8 +23,6 @@
 metrics = Metrics(namespace="datadog.api_security")
 _sentinel = object()
 
-# Delay in seconds to avoid sampling the same route too often
-DELAY = 30.0
 
 # Max number of endpoint hashes to keep in the hashtable
 MAX_HASHTABLE_SIZE = 4096
@@ -107,7 +105,7 @@ def _should_collect_schema(self, env, priority: int) -> bool:
         end_point_hash = hash((route, method, status))
         current_time = time.monotonic()
         previous_time = self._hashtable.get(end_point_hash, M_INFINITY)
-        if previous_time >= current_time - DELAY:
+        if previous_time >= current_time - asm_config._api_security_sample_delay:
             return False
         if previous_time is M_INFINITY:
             if len(self._hashtable) >= MAX_HASHTABLE_SIZE:

ddtrace/appsec/_constants.py

@@ -144,6 +144,7 @@ class API_SECURITY(metaclass=Constant_Class):
     RESPONSE_HEADERS_NO_COOKIES = "_dd.appsec.s.res.headers"
     RESPONSE_BODY = "_dd.appsec.s.res.body"
     SAMPLE_RATE = "DD_API_SECURITY_REQUEST_SAMPLE_RATE"
+    SAMPLE_DELAY = "DD_API_SECURITY_SAMPLE_DELAY"
     MAX_PAYLOAD_SIZE = 0x1000000  # 16MB maximum size
 
 

ddtrace/settings/asm.py

@@ -49,6 +49,7 @@ class ASMConfig(Env):
     _user_model_name_field = Env.var(str, APPSEC.USER_MODEL_NAME_FIELD, default="")
     _api_security_enabled = Env.var(bool, API_SECURITY.ENV_VAR_ENABLED, default=True)
     _api_security_sample_rate = 0.0
+    _api_security_sample_delay = Env.var(float, API_SECURITY.SAMPLE_DELAY, default=30.0)
     _api_security_parse_response_body = Env.var(bool, API_SECURITY.PARSE_RESPONSE_BODY, default=True)
     _asm_libddwaf = build_libddwaf_filename()
     _asm_libddwaf_available = os.path.exists(_asm_libddwaf)
@@ -90,6 +91,7 @@ class ASMConfig(Env):
         "_user_model_name_field",
         "_api_security_enabled",
         "_api_security_sample_rate",
+        "_api_security_sample_delay",
         "_api_security_parse_response_body",
         "_waf_timeout",
         "_iast_redaction_enabled",

tests/appsec/contrib_appsec/utils.py

@@ -949,9 +949,7 @@ def test_api_security_scanners(self, interface: Interface, get_tag, apisec_enabl
 
         from ddtrace.ext import http
 
-        with override_global_config(
-            dict(_asm_enabled=True, _api_security_enabled=apisec_enabled, _api_security_sample_rate=1.0)
-        ):
+        with override_global_config(dict(_asm_enabled=True, _api_security_enabled=apisec_enabled)):
             self.update_tracer(interface)
             response = interface.client.post(
                 "/",
@@ -961,7 +959,6 @@ def test_api_security_scanners(self, interface: Interface, get_tag, apisec_enabl
             assert self.status(response) == 200
             assert get_tag(http.STATUS_CODE) == "200"
             assert asm_config._api_security_enabled == apisec_enabled
-            assert asm_config._api_security_sample_rate == 1.0
 
             value = get_tag("_dd.appsec.s.req.body")
             if apisec_enabled:
@@ -973,11 +970,14 @@ def test_api_security_scanners(self, interface: Interface, get_tag, apisec_enabl
 
     @pytest.mark.parametrize("apisec_enabled", [True, False])
     @pytest.mark.parametrize("priority", ["keep", "drop"])
-    def test_api_security_sampling(self, interface: Interface, get_tag, apisec_enabled, priority):
+    @pytest.mark.parametrize("delay", [0.0, 120.0])
+    def test_api_security_sampling(self, interface: Interface, get_tag, apisec_enabled, priority, delay):
         from ddtrace.ext import http
 
         payload = {"mastercard": "5123456789123456"}
-        with override_global_config(dict(_asm_enabled=True, _api_security_enabled=apisec_enabled)):
+        with override_global_config(
+            dict(_asm_enabled=True, _api_security_enabled=apisec_enabled, _api_security_sample_delay=delay)
+        ):
             self.update_tracer(interface)
             response = interface.client.post(
                 f"/asm/?priority={priority}",
@@ -1005,7 +1005,10 @@ def test_api_security_sampling(self, interface: Interface, get_tag, apisec_enabl
             assert asm_config._api_security_enabled == apisec_enabled
 
             value = get_tag("_dd.appsec.s.req.body")
-            assert value is None
+            if apisec_enabled and priority == "keep" and delay == 0.0:
+                assert value
+            else:
+                assert value is None
 
     def test_request_invalid_rule_file(self, interface):
         """