Merge branch '2.17' into backport-11735-to-2.17

.gitlab-ci.yml

@@ -2,6 +2,7 @@ stages:
   - package
   - tests-gen
   - tests-trigger
+  - quality-gate
   - shared-pipeline
   - benchmarks
   - macrobenchmarks
@@ -84,3 +85,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**'

ddtrace/appsec/_common_module_patches.py

@@ -60,8 +60,10 @@ def wrapped_read_F3E51D71B4EC16EF(original_read_callable, instance, args, kwargs
     """
     wrapper for _io.BytesIO and _io.StringIO read function
     """
+    from ddtrace.appsec._iast._iast_request_context import is_iast_request_enabled
+
     result = original_read_callable(*args, **kwargs)
-    if asm_config._iast_enabled:
+    if asm_config._iast_enabled and is_iast_request_enabled():
         from ddtrace.appsec._iast._taint_tracking import OriginType
         from ddtrace.appsec._iast._taint_tracking import Source
         from ddtrace.appsec._iast._taint_tracking import get_tainted_ranges
@@ -87,7 +89,9 @@ def wrapped_open_CFDDB7ABBA9081B6(original_open_callable, instance, args, kwargs
     """
     wrapper for open file function
     """
-    if asm_config._iast_enabled:
+    from ddtrace.appsec._iast._iast_request_context import is_iast_request_enabled
+
+    if asm_config._iast_enabled and is_iast_request_enabled():
         try:
             from ddtrace.appsec._iast.taint_sinks.path_traversal import check_and_report_path_traversal
 
@@ -176,7 +180,9 @@ def wrapped_request_D8CB81E472AF98A2(original_request_callable, instance, args,
     wrapper for third party requests.request function
     https://requests.readthedocs.io
     """
-    if asm_config._iast_enabled:
+    from ddtrace.appsec._iast._iast_request_context import is_iast_request_enabled
+
+    if asm_config._iast_enabled and is_iast_request_enabled():
         from ddtrace.appsec._iast.taint_sinks.ssrf import _iast_report_ssrf
 
         _iast_report_ssrf(original_request_callable, *args, **kwargs)
@@ -216,7 +222,9 @@ def wrapped_system_5542593D237084A7(original_command_callable, instance, args, k
     """
     command = args[0] if args else kwargs.get("command", None)
     if command is not None:
-        if asm_config._iast_enabled:
+        from ddtrace.appsec._iast._iast_request_context import is_iast_request_enabled
+
+        if asm_config._iast_enabled and is_iast_request_enabled():
             from ddtrace.appsec._iast.taint_sinks.command_injection import _iast_report_cmdi
 
             _iast_report_cmdi(command)

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/langchain/patch.py

@@ -1,4 +1,3 @@
-import json
 import os
 import sys
 from typing import Any
@@ -954,17 +953,22 @@ def _on_span_started(span: Span):
                 span.set_tag_str("langchain.request.inputs.%d.%s" % (idx, k), integration.trunc(str(v)))
 
     def _on_span_finished(span: Span, streamed_chunks):
+        maybe_parser = instance.steps[-1] if instance.steps else None
         if (
             streamed_chunks
             and langchain_core
-            and isinstance(instance.steps[-1], langchain_core.output_parsers.JsonOutputParser)
+            and isinstance(maybe_parser, langchain_core.output_parsers.JsonOutputParser)
         ):
-            # it's possible that the chain has a json output parser
-            # this will have already concatenated the chunks into a json object
+            # it's possible that the chain has a json output parser type
+            # this will have already concatenated the chunks into an object
 
-            # it's also possible the json output parser isn't the last step,
+            # it's also possible the this parser type isn't the last step,
             # but one of the last steps, in which case we won't act on it here
-            content = json.dumps(streamed_chunks[-1])
+            result = streamed_chunks[-1]
+            if maybe_parser.__class__.__name__ == "JsonOutputParser":
+                content = safe_json(result)
+            else:
+                content = str(result)
         else:
             # best effort to join chunks together
             content = "".join([str(chunk) for chunk in streamed_chunks])

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/llmobs/_llmobs.py

@@ -23,6 +23,7 @@
 from ddtrace.internal.telemetry import telemetry_writer
 from ddtrace.internal.telemetry.constants import TELEMETRY_APM_PRODUCT
 from ddtrace.internal.utils.formats import asbool
+from ddtrace.internal.utils.formats import parse_tags_str
 from ddtrace.llmobs._constants import ANNOTATIONS_CONTEXT_ID
 from ddtrace.llmobs._constants import INPUT_DOCUMENTS
 from ddtrace.llmobs._constants import INPUT_MESSAGES
@@ -346,8 +347,20 @@ def flush(cls) -> None:
 
     @staticmethod
     def _patch_integrations() -> None:
-        """Patch LLM integrations."""
-        patch(**{integration: True for integration in SUPPORTED_LLMOBS_INTEGRATIONS.values()})  # type: ignore[arg-type]
+        """
+        Patch LLM integrations. Ensure that we do not ignore DD_TRACE_<MODULE>_ENABLED or DD_PATCH_MODULES settings.
+        """
+        integrations_to_patch = {integration: True for integration in SUPPORTED_LLMOBS_INTEGRATIONS.values()}
+        for module, _ in integrations_to_patch.items():
+            env_var = "DD_TRACE_%s_ENABLED" % module.upper()
+            if env_var in os.environ:
+                integrations_to_patch[module] = asbool(os.environ[env_var])
+        dd_patch_modules = os.getenv("DD_PATCH_MODULES")
+        dd_patch_modules_to_str = parse_tags_str(dd_patch_modules)
+        integrations_to_patch.update(
+            {k: asbool(v) for k, v in dd_patch_modules_to_str.items() if k in SUPPORTED_LLMOBS_INTEGRATIONS.values()}
+        )
+        patch(**integrations_to_patch)  # type: ignore[arg-type]
         log.debug("Patched LLM integrations: %s", list(SUPPORTED_LLMOBS_INTEGRATIONS.values()))
 
     @classmethod

releasenotes/notes/fix-appsec-use-unpatched-json-8d09aacad4808ef2.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    ASM: This fix resolves an issue where AppSec was using a patched JSON loads, creating telemetry errors.

releasenotes/notes/fix-llmobs-do-not-ignore-global-patch-configs-a2adc4803f55b142.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    LLM Observability: This fix resolves an issue where ``LLMObs.enable()`` ignored global patch configurations, specifically
+     the ``DD_TRACE_<INTEGRATION>_ENABLED`` and ``DD_PATCH_MODULES`` environment variables.

releasenotes/notes/iast-fix-iast-propagation-error-2-ba4a998133269a7c.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    ASM: This fix resolves an issue where AppSec was using a patched request and builtins functions, 
+    creating telemetry errors.

releasenotes/notes/langchain-pydantic-output-parsers-19bc162212ec051e.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    langchain: resolves a JSON decoding issue resulting from tagging streamed outputs from chains ending with a PydanticOutputParser.

tests/llmobs/test_llmobs_service.py

@@ -31,6 +31,7 @@
 from ddtrace.llmobs._constants import SPAN_KIND
 from ddtrace.llmobs._constants import SPAN_START_WHILE_DISABLED_WARNING
 from ddtrace.llmobs._constants import TAGS
+from ddtrace.llmobs._llmobs import SUPPORTED_LLMOBS_INTEGRATIONS
 from ddtrace.llmobs._llmobs import LLMObsTraceProcessor
 from ddtrace.llmobs.utils import Prompt
 from tests.llmobs._utils import _expected_llmobs_eval_metric_event
@@ -144,6 +145,65 @@ def test_service_enable_already_enabled(mock_logs):
         mock_logs.debug.assert_has_calls([mock.call("%s already enabled", "LLMObs")])
 
 
+@mock.patch("ddtrace.llmobs._llmobs.patch")
+def test_service_enable_patches_llmobs_integrations(mock_tracer_patch):
+    with override_global_config(dict(_dd_api_key="<not-a-real-api-key>", _llmobs_ml_app="<ml-app-name>")):
+        llmobs_service.enable()
+        mock_tracer_patch.assert_called_once()
+        kwargs = mock_tracer_patch.call_args[1]
+        for module in SUPPORTED_LLMOBS_INTEGRATIONS.values():
+            assert kwargs[module] is True
+        llmobs_service.disable()
+
+
+@mock.patch("ddtrace.llmobs._llmobs.patch")
+def test_service_enable_does_not_override_global_patch_modules(mock_tracer_patch, monkeypatch):
+    monkeypatch.setenv("DD_PATCH_MODULES", "openai:false")
+    with override_global_config(dict(_dd_api_key="<not-a-real-api-key>", _llmobs_ml_app="<ml-app-name>")):
+        llmobs_service.enable()
+        mock_tracer_patch.assert_called_once()
+        kwargs = mock_tracer_patch.call_args[1]
+        for module in SUPPORTED_LLMOBS_INTEGRATIONS.values():
+            if module == "openai":
+                assert kwargs[module] is False
+                continue
+            assert kwargs[module] is True
+        llmobs_service.disable()
+
+
+@mock.patch("ddtrace.llmobs._llmobs.patch")
+def test_service_enable_does_not_override_integration_enabled_env_vars(mock_tracer_patch, monkeypatch):
+    monkeypatch.setenv("DD_TRACE_OPENAI_ENABLED", "false")
+    with override_global_config(dict(_dd_api_key="<not-a-real-api-key>", _llmobs_ml_app="<ml-app-name>")):
+        llmobs_service.enable()
+        mock_tracer_patch.assert_called_once()
+        kwargs = mock_tracer_patch.call_args[1]
+        for module in SUPPORTED_LLMOBS_INTEGRATIONS.values():
+            if module == "openai":
+                assert kwargs[module] is False
+                continue
+            assert kwargs[module] is True
+        llmobs_service.disable()
+
+
+@mock.patch("ddtrace.llmobs._llmobs.patch")
+def test_service_enable_does_not_override_global_patch_config(mock_tracer_patch, monkeypatch):
+    """Test that _patch_integrations() ensures `DD_PATCH_MODULES` overrides `DD_TRACE_<MODULE>_ENABLED`."""
+    monkeypatch.setenv("DD_TRACE_OPENAI_ENABLED", "true")
+    monkeypatch.setenv("DD_TRACE_ANTHROPIC_ENABLED", "false")
+    monkeypatch.setenv("DD_PATCH_MODULES", "openai:false")
+    with override_global_config(dict(_dd_api_key="<not-a-real-api-key>", _llmobs_ml_app="<ml-app-name>")):
+        llmobs_service.enable()
+        mock_tracer_patch.assert_called_once()
+        kwargs = mock_tracer_patch.call_args[1]
+        for module in SUPPORTED_LLMOBS_INTEGRATIONS.values():
+            if module in ("openai", "anthropic"):
+                assert kwargs[module] is False
+                continue
+            assert kwargs[module] is True
+        llmobs_service.disable()
+
+
 def test_start_span_while_disabled_logs_warning(LLMObs, mock_logs):
     LLMObs.disable()
     _ = LLMObs.llm(model_name="test_model", name="test_llm_call", model_provider="test_provider")