Merge branch 'main' into peterg17/pytorch_profiling_integration2

.github/workflows/system-tests.yml

@@ -54,7 +54,7 @@ jobs:
       # system-tests requires an API_KEY, but it does not have to be a valid key, as long as we don't run a scenario
       # that make assertion on backend data. Using a fake key allow to run system tests on PR originating from forks.
       # If ever it's needed, a valid key exists in the repo, using ${{ secrets.DD_API_KEY }}
-      DD_API_KEY: 1234567890abcdef1234567890abcdef
+      DD_API_KEY: ${{ secrets.FAKE_DD_API_KEY }}
       CMAKE_BUILD_PARALLEL_LEVEL: 12
       SYSTEM_TESTS_AWS_ACCESS_KEY_ID: ${{ secrets.IDM_AWS_ACCESS_KEY_ID }}
       SYSTEM_TESTS_AWS_SECRET_ACCESS_KEY: ${{ secrets.IDM_AWS_SECRET_ACCESS_KEY }}
@@ -106,7 +106,7 @@ jobs:
       # system-tests requires an API_KEY, but it does not have to be a valid key, as long as we don't run a scenario
       # that make assertion on backend data. Using a fake key allow to run system tests on PR originating from forks.
       # If ever it's needed, a valid key exists in the repo, using ${{ secrets.DD_API_KEY }}
-      DD_API_KEY: 1234567890abcdef1234567890abcdef
+      DD_API_KEY: ${{ secrets.FAKE_DD_API_KEY }}
       CMAKE_BUILD_PARALLEL_LEVEL: 12
       SYSTEM_TESTS_AWS_ACCESS_KEY_ID: ${{ secrets.IDM_AWS_ACCESS_KEY_ID }}
       SYSTEM_TESTS_AWS_SECRET_ACCESS_KEY: ${{ secrets.IDM_AWS_SECRET_ACCESS_KEY }}

.gitlab-ci.yml

@@ -2,6 +2,7 @@ stages:
   - package
   - tests-gen
   - tests-trigger
+  - quality-gate
   - shared-pipeline
   - benchmarks
   - macrobenchmarks
@@ -87,3 +88,17 @@ deploy_to_di_backend:manual:
     UPSTREAM_COMMIT_AUTHOR: $CI_COMMIT_AUTHOR
     UPSTREAM_TAG: $CI_COMMIT_TAG
     UPSTREAM_PACKAGE_JOB: build
+
+check_new_flaky_tests:
+  stage: quality-gate
+  extends: .testrunner
+  script:
+    - curl -L --fail "https://github.com/DataDog/datadog-ci/releases/latest/download/datadog-ci_linux-x64" --output "/usr/local/bin/datadog-ci" && chmod +x /usr/local/bin/datadog-ci
+    - export DD_SITE=datadoghq.com
+    - export DD_API_KEY=$(aws ssm get-parameter --region us-east-1 --name ci.${CI_PROJECT_NAME}.dd-api-key-qualitygate --with-decryption --query "Parameter.Value" --out text)
+    - export DD_APP_KEY=$(aws ssm get-parameter --region us-east-1 --name ci.${CI_PROJECT_NAME}.dd-app-key-qualitygate --with-decryption --query "Parameter.Value" --out text)
+    - datadog-ci gate evaluate
+  except:
+    - main
+    - '[0-9].[0-9]*'
+    - 'mq-working-branch**'

.riot/requirements/151a249.txt

@@ -0,0 +1,20 @@
+#
+# This file is autogenerated by pip-compile with Python 3.13
+# by the following command:
+#
+#    pip-compile --allow-unsafe --no-annotate .riot/requirements/151a249.in
+#
+attrs==24.2.0
+coverage[toml]==7.6.9
+hypothesis==6.45.0
+iniconfig==2.0.0
+lxml==5.3.0
+mock==5.1.0
+opentracing==2.4.0
+packaging==24.2
+pluggy==1.5.0
+pytest==8.3.4
+pytest-cov==6.0.0
+pytest-mock==3.14.0
+ruamel-yaml==0.18.6
+sortedcontainers==2.4.0

.riot/requirements/4d1fa34.txt

@@ -0,0 +1,20 @@
+#
+# This file is autogenerated by pip-compile with Python 3.13
+# by the following command:
+#
+#    pip-compile --allow-unsafe --no-annotate .riot/requirements/4d1fa34.in
+#
+attrs==24.2.0
+coverage[toml]==7.6.9
+hypothesis==6.45.0
+iniconfig==2.0.0
+lxml==5.3.0
+mock==5.1.0
+opentracing==2.4.0
+packaging==24.2
+pluggy==1.5.0
+pytest==8.3.4
+pytest-cov==6.0.0
+pytest-mock==3.14.0
+ruamel-yaml==0.18.6
+sortedcontainers==2.4.0

ddtrace/appsec/_constants.py

@@ -129,6 +129,10 @@ class IAST(metaclass=Constant_Class):
     ENV_DEBUG: Literal["DD_IAST_DEBUG"] = "DD_IAST_DEBUG"
     ENV_PROPAGATION_DEBUG: Literal["DD_IAST_PROPAGATION_DEBUG"] = "DD_IAST_PROPAGATION_DEBUG"
     ENV_REQUEST_SAMPLING: Literal["DD_IAST_REQUEST_SAMPLING"] = "DD_IAST_REQUEST_SAMPLING"
+    DD_IAST_VULNERABILITIES_PER_REQUEST: Literal[
+        "DD_IAST_VULNERABILITIES_PER_REQUEST"
+    ] = "DD_IAST_VULNERABILITIES_PER_REQUEST"
+    DD_IAST_MAX_CONCURRENT_REQUESTS: Literal["DD_IAST_MAX_CONCURRENT_REQUESTS"] = "DD_IAST_MAX_CONCURRENT_REQUESTS"
     ENV_TELEMETRY_REPORT_LVL: Literal["DD_IAST_TELEMETRY_VERBOSITY"] = "DD_IAST_TELEMETRY_VERBOSITY"
     LAZY_TAINT: Literal["_DD_IAST_LAZY_TAINT"] = "_DD_IAST_LAZY_TAINT"
     JSON: Literal["_dd.iast.json"] = "_dd.iast.json"

ddtrace/appsec/_iast/__init__.py

@@ -1,4 +1,4 @@
-"""IAST (interactive application security testing) analyzes code for security vulnerabilities.
+"""IAST (Interactive Application Security Testing) analyzes code for security vulnerabilities.
 
 To add new vulnerabilities analyzers (Taint sink) we should update `IAST_PATCH` in
 `ddtrace/appsec/iast/_patch_modules.py`

ddtrace/appsec/_iast/_overhead_control_engine.py

@@ -3,7 +3,6 @@
 limit. It will measure operations being executed in a request and it will deactivate detection
 (and therefore reduce the overhead to nearly 0) if a certain threshold is reached.
 """
-import os
 from typing import Set
 from typing import Text
 from typing import Tuple
@@ -25,22 +24,18 @@ def get_request_sampling_value() -> float:
     return float(asm_config._iast_request_sampling)
 
 
-MAX_REQUESTS = int(os.environ.get("DD_IAST_MAX_CONCURRENT_REQUESTS", 2))
-MAX_VULNERABILITIES_PER_REQUEST = int(os.environ.get("DD_IAST_VULNERABILITIES_PER_REQUEST", 2))
-
-
 class Operation(object):
     """Common operation related to Overhead Control Engine (OCE). Every vulnerabilities/taint_sinks should inherit
     from this class. OCE instance calls these methods to control the overhead produced in each request.
     """
 
     _lock = threading.Lock()
-    _vulnerability_quota = MAX_VULNERABILITIES_PER_REQUEST
+    _vulnerability_quota = asm_config._iast_max_vulnerabilities_per_requests
     _reported_vulnerabilities: Set[Tuple[str, int]] = set()
 
     @classmethod
     def reset(cls):
-        cls._vulnerability_quota = MAX_VULNERABILITIES_PER_REQUEST
+        cls._vulnerability_quota = asm_config._iast_max_vulnerabilities_per_requests
         cls._reported_vulnerabilities = set()
 
     @classmethod
@@ -57,7 +52,7 @@ def acquire_quota(cls) -> bool:
     def increment_quota(cls) -> bool:
         cls._lock.acquire()
         result = False
-        if cls._vulnerability_quota < MAX_VULNERABILITIES_PER_REQUEST:
+        if cls._vulnerability_quota < asm_config._iast_max_vulnerabilities_per_requests:
             cls._vulnerability_quota += 1
             result = True
         cls._lock.release()
@@ -86,12 +81,13 @@ class OverheadControl(object):
     """
 
     _lock = threading.Lock()
-    _request_quota = MAX_REQUESTS
+    _request_quota = asm_config._iast_max_concurrent_requests
     _vulnerabilities: Set[Type[Operation]] = set()
     _sampler = RateSampler(sample_rate=get_request_sampling_value() / 100.0)
 
     def reconfigure(self):
         self._sampler = RateSampler(sample_rate=get_request_sampling_value() / 100.0)
+        self._request_quota = asm_config._iast_max_concurrent_requests
 
     def acquire_request(self, span: Span) -> bool:
         """Decide whether if IAST analysis will be done for this request.

ddtrace/appsec/_iast/taint_sinks/_base.py

@@ -61,9 +61,11 @@ def wrapper(wrapped: Callable, instance: Any, args: Any, kwargs: Any) -> Any:
             vulnerability and update the context with the report information.
             """
             if not is_iast_request_enabled():
-                log.debug(
-                    "[IAST] VulnerabilityBase.wrapper. No request quota or this vulnerability is outside the context"
-                )
+                if _is_iast_debug_enabled():
+                    log.debug(
+                        "[IAST] VulnerabilityBase.wrapper. No request quota or this vulnerability "
+                        "is outside the context"
+                    )
                 return wrapped(*args, **kwargs)
             elif cls.has_quota():
                 return func(wrapped, instance, args, kwargs)

ddtrace/appsec/_utils.py

@@ -5,6 +5,7 @@
 
 from ddtrace.appsec._constants import API_SECURITY
 from ddtrace.appsec._constants import APPSEC
+from ddtrace.internal._unpatched import unpatched_json_loads
 from ddtrace.internal.compat import to_unicode
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.utils.http import _get_blocked_template  # noqa:F401
@@ -17,8 +18,6 @@
 
 
 def parse_response_body(raw_body):
-    import json
-
     import xmltodict
 
     from ddtrace.appsec import _asm_request_context
@@ -54,7 +53,7 @@ def access_body(bd):
     try:
         # TODO handle charset
         if "json" in content_type:
-            req_body = json.loads(access_body(raw_body))
+            req_body = unpatched_json_loads(access_body(raw_body))
         elif "xml" in content_type:
             req_body = xmltodict.parse(access_body(raw_body))
         else:

ddtrace/contrib/internal/kafka/patch.py

@@ -247,7 +247,7 @@ def _instrument_message(messages, pin, start_ns, instance, err):
         name=schematize_messaging_operation(kafkax.CONSUME, provider="kafka", direction=SpanDirection.PROCESSING),
         service=trace_utils.ext_service(pin, config.kafka),
         span_type=SpanTypes.WORKER,
-        child_of=ctx if ctx is not None else pin.tracer.context_provider.active(),
+        child_of=ctx if ctx is not None and ctx.trace_id is not None else pin.tracer.context_provider.active(),
         activate=True,
     ) as span:
         # reset span start time to before function call

ddtrace/contrib/pytest/_plugin_v2.py

@@ -256,8 +256,16 @@ def _pytest_collection_finish(session) -> None:
         InternalTestSuite.discover(suite_id)
 
         item_path = Path(item.path if hasattr(item, "path") else item.fspath).absolute()
+        workspace_path = InternalTestSession.get_workspace_path()
+        if workspace_path:
+            try:
+                repo_relative_path = item_path.relative_to(workspace_path)
+            except ValueError:
+                repo_relative_path = item_path
+        else:
+            repo_relative_path = item_path
 
-        item_codeowners = InternalTestSession.get_path_codeowners(item_path)
+        item_codeowners = InternalTestSession.get_path_codeowners(repo_relative_path) if repo_relative_path else None
 
         source_file_info = _get_source_file_info(item, item_path)
 

ddtrace/internal/_unpatched.py

@@ -1,6 +1,7 @@
 # Acquire a reference to the open function from the builtins module. This is
 # necessary to ensure that the open function can be used unpatched when required.
 from builtins import open as unpatched_open  # noqa
+from json import loads as unpatched_json_loads  # noqa
 
 # Acquire a reference to the threading module. Some parts of the library (e.g.
 # the profiler) might be enabled programmatically and therefore might end up

ddtrace/internal/ci_visibility/api/_session.py

@@ -119,8 +119,8 @@ def set_efd_abort_reason(self, abort_reason: str):
         self._efd_abort_reason = abort_reason
 
     def efd_is_faulty_session(self):
-        """A session is considered "EFD faulty" if percentage of tests considered new is greater than the given
-        threshold
+        """A session is considered "EFD faulty" if the percentage of tests considered new is greater than the
+        given threshold, and the total number of news tests exceeds the threshold.
 
         NOTE: this behavior is cached on the assumption that this method will only be called once
         """
@@ -130,16 +130,19 @@ def efd_is_faulty_session(self):
         if self._session_settings.efd_settings.enabled is False:
             return False
 
-        total_tests = 0
-        new_tests = 0
+        total_tests_count = 0
+        new_tests_count = 0
         for _module in self._children.values():
             for _suite in _module._children.values():
                 for _test in _suite._children.values():
-                    total_tests += 1
+                    total_tests_count += 1
                     if _test.is_new():
-                        new_tests += 1
+                        new_tests_count += 1
 
-        new_tests_pct = 100 * (new_tests / total_tests)
+        if new_tests_count <= self._session_settings.efd_settings.faulty_session_threshold:
+            return False
+
+        new_tests_pct = 100 * (new_tests_count / total_tests_count)
 
         self._efd_is_faulty_session = new_tests_pct > self._session_settings.efd_settings.faulty_session_threshold
 

ddtrace/internal/ci_visibility/api/_test.py

@@ -101,8 +101,10 @@ def _set_efd_tags(self) -> None:
             self.set_tag(TEST_EFD_ABORT_REASON, self._efd_abort_reason)
 
         # NOTE: The is_new tag is currently only being set in the context of EFD (since that is the only context in
-        # which unique tests are fetched).
-        if self.is_new():
+        # which unique tests are fetched). Additionally, if a session is considered faulty, we do not want to tag the
+        # test as new.
+        session = self.get_session()
+        if self.is_new() and session is not None and not session.efd_is_faulty_session():
             self.set_tag(TEST_IS_NEW, self._is_new)
 
     def _set_atr_tags(self) -> None:

ddtrace/internal/ci_visibility/recorder.py

@@ -1031,7 +1031,7 @@ def _on_session_get_path_codeowners(path: Path) -> Optional[List[str]]:
     codeowners = CIVisibility.get_codeowners()
     if codeowners is None:
         return None
-    return codeowners.of(str(path.absolute()))
+    return codeowners.of(str(path))
 
 
 def _register_session_handlers():

ddtrace/internal/datadog/profiling/ddup/_ddup.pyi

@@ -3,6 +3,7 @@ from typing import Optional
 from typing import Union
 from .._types import StringType
 from ddtrace._trace.span import Span
+from ddtrace._trace.tracer import Tracer
 
 def config(
     env: StringType,
@@ -16,7 +17,7 @@ def config(
     enable_code_provenance: Optional[bool],
 ) -> None: ...
 def start() -> None: ...
-def upload() -> None: ...
+def upload(tracer: Optional[Tracer]) -> None: ...
 
 class SampleHandle:
     def flush_sample(self) -> None: ...

ddtrace/internal/datadog/profiling/ddup/_ddup.pyx

@@ -20,6 +20,7 @@ from ddtrace.internal.constants import DEFAULT_SERVICE_NAME
 from ddtrace.internal.packages import get_distributions
 from ddtrace.internal.runtime import get_runtime_id
 from ddtrace._trace.span import Span
+from ddtrace._trace.tracer import Tracer
 
 
 ctypedef void (*func_ptr_t)(string_view)
@@ -413,16 +414,16 @@ def _get_endpoint(tracer)-> str:
     return endpoint
 
 
-def upload() -> None:
+def upload(tracer: Optional[Tracer] = ddtrace.tracer) -> None:
     call_func_with_str(ddup_set_runtime_id, get_runtime_id())
 
-    processor = ddtrace.tracer._endpoint_call_counter_span_processor
+    processor = tracer._endpoint_call_counter_span_processor
     endpoint_counts, endpoint_to_span_ids = processor.reset()
 
     call_ddup_profile_set_endpoints(endpoint_to_span_ids)
     call_ddup_profile_add_endpoint_counts(endpoint_counts)
 
-    endpoint = _get_endpoint(ddtrace.tracer)
+    endpoint = _get_endpoint(tracer)
     call_func_with_str(ddup_config_url, endpoint)
 
     with nogil: