Merge branch '3.x-staging' into munir/upgrade-min-aws-lambda

Author: mabdinur

.github/workflows/check_safe_main_merge.yml

@@ -5,6 +5,7 @@
 from ddtrace.internal import forksafe
 from ddtrace.internal.logger import get_logger
 from ddtrace.internal.periodic import PeriodicService
+from ddtrace.internal.service import ServiceStatus
 from ddtrace.internal.telemetry import telemetry_writer
 from ddtrace.internal.telemetry.constants import TELEMETRY_NAMESPACE
 from ddtrace.llmobs._evaluators.ragas.answer_relevancy import RagasAnswerRelevancyEvaluator
@@ -94,6 +95,8 @@ def recreate(self) -> "EvaluatorRunner":
         )
 
     def enqueue(self, span_event: Dict, span: Span) -> None:
+        if self.status == ServiceStatus.STOPPED:
+            return
         with self._lock:
             if len(self._buffer) >= self._buffer_limit:
                 logger.warning(

.riot/requirements/16562eb.txt

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    LLM Observability: This fix resolves an issue where spans were being enqueued to an inactive evaluator runner which caused noisy logs
+                  related to the evaluator runner buffer being full.

.riot/requirements/df60af6.txt

@@ -2492,13 +2492,6 @@ def select_pys(min_version: str = MIN_PYTHON_VERSION, max_version: str = MAX_PYT
                 "pytest-randomly": latest,
             },
             venvs=[
-                Venv(
-                    pys="3.7",
-                    pkgs={
-                        "openai[datalib]": "==1.30.1",
-                        "pillow": "==9.5.0",
-                    },
-                ),
                 Venv(
                     pys=select_pys(min_version="3.8", max_version="3.11"),
                     pkgs={
@@ -2927,11 +2920,13 @@ def select_pys(min_version: str = MIN_PYTHON_VERSION, max_version: str = MAX_PYT
         Venv(
             name="llmobs",
             command="pytest {cmdargs} tests/llmobs",
-            pkgs={"vcrpy": latest, "pytest-asyncio": "==0.21.1"},
-            venvs=[
-                Venv(pys="3.7"),
-                Venv(pys=select_pys(min_version="3.8"), pkgs={"ragas": "==0.1.21", "langchain": latest}),
-            ],
+            pkgs={
+                "vcrpy": latest,
+                "pytest-asyncio": "==0.21.1",
+                "ragas": "==0.1.21",
+                "langchain": latest,
+            },
+            pys=select_pys(min_version="3.8"),
         ),
         Venv(
             name="profile",

ddtrace/llmobs/_evaluators/runner.py

@@ -18,39 +18,44 @@
 DUMMY_SPAN = Span("dummy_span")
 
 
-def test_evaluator_runner_start(mock_evaluator_logs):
-    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=mock.MagicMock())
-    evaluator_runner.evaluators.append(DummyEvaluator(llmobs_service=mock.MagicMock()))
+@pytest.fixture
+def active_evaluator_runner(llmobs):
+    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=llmobs)
+    evaluator_runner.evaluators.append(DummyEvaluator(llmobs_service=llmobs))
     evaluator_runner.start()
+    yield evaluator_runner
+
+
+def test_evaluator_runner_start(mock_evaluator_logs, active_evaluator_runner):
     mock_evaluator_logs.debug.assert_has_calls([mock.call("started %r", "EvaluatorRunner")])
 
 
-def test_evaluator_runner_buffer_limit(mock_evaluator_logs):
-    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=mock.MagicMock())
+def test_evaluator_runner_buffer_limit(mock_evaluator_logs, active_evaluator_runner):
     for _ in range(1001):
-        evaluator_runner.enqueue({}, DUMMY_SPAN)
+        active_evaluator_runner.enqueue({}, DUMMY_SPAN)
     mock_evaluator_logs.warning.assert_called_with(
         "%r event buffer full (limit is %d), dropping event", "EvaluatorRunner", 1000
     )
 
 
-def test_evaluator_runner_periodic_enqueues_eval_metric(llmobs, mock_llmobs_eval_metric_writer):
-    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=llmobs)
-    evaluator_runner.evaluators.append(DummyEvaluator(llmobs_service=llmobs))
-    evaluator_runner.enqueue({"span_id": "123", "trace_id": "1234"}, DUMMY_SPAN)
-    evaluator_runner.periodic()
+def test_evaluator_runner_periodic_enqueues_eval_metric(mock_llmobs_eval_metric_writer, active_evaluator_runner):
+    active_evaluator_runner.enqueue({"span_id": "123", "trace_id": "1234"}, DUMMY_SPAN)
+    active_evaluator_runner.periodic()
     mock_llmobs_eval_metric_writer.enqueue.assert_called_once_with(
         _dummy_evaluator_eval_metric_event(span_id="123", trace_id="1234")
     )
 
 
-@pytest.mark.vcr_logs
-def test_evaluator_runner_timed_enqueues_eval_metric(llmobs, mock_llmobs_eval_metric_writer):
-    evaluator_runner = EvaluatorRunner(interval=0.01, llmobs_service=llmobs)
-    evaluator_runner.evaluators.append(DummyEvaluator(llmobs_service=llmobs))
+def test_evaluator_runner_stopped_does_not_enqueue_metric(llmobs, mock_llmobs_eval_metric_writer):
+    evaluator_runner = EvaluatorRunner(interval=0.1, llmobs_service=llmobs)
     evaluator_runner.start()
-
     evaluator_runner.enqueue({"span_id": "123", "trace_id": "1234"}, DUMMY_SPAN)
+    assert not evaluator_runner._buffer
+    assert mock_llmobs_eval_metric_writer.enqueue.call_count == 0
+
+
+def test_evaluator_runner_timed_enqueues_eval_metric(llmobs, mock_llmobs_eval_metric_writer, active_evaluator_runner):
+    active_evaluator_runner.enqueue({"span_id": "123", "trace_id": "1234"}, DUMMY_SPAN)
 
     time.sleep(0.1)
 

releasenotes/notes/fix-evaluator-runner-log-fcd0aa18602d4252.yaml

@@ -1383,6 +1383,7 @@ def test_llmobs_fork_recreates_and_restarts_eval_metric_writer():
 
 def test_llmobs_fork_recreates_and_restarts_evaluator_runner(mock_ragas_evaluator):
     """Test that forking a process correctly recreates and restarts the EvaluatorRunner."""
+    pytest.importorskip("ragas")
     with override_env(dict(_DD_LLMOBS_EVALUATORS="ragas_faithfulness")):
         with mock.patch("ddtrace.llmobs._evaluators.runner.EvaluatorRunner.periodic"):
             llmobs_service.enable(_tracer=DummyTracer(), ml_app="test_app")
@@ -1464,6 +1465,8 @@ def test_llmobs_fork_submit_evaluation(monkeypatch):
 def test_llmobs_fork_evaluator_runner_run(monkeypatch):
     """Test that forking a process correctly encodes new spans created in each process."""
     monkeypatch.setenv("_DD_LLMOBS_EVALUATOR_INTERVAL", 5.0)
+    pytest.importorskip("ragas")
+    monkeypatch.setenv("_DD_LLMOBS_EVALUATORS", "ragas_faithfulness")
     with mock.patch("ddtrace.llmobs._evaluators.runner.EvaluatorRunner.periodic"):
         llmobs_service.enable(_tracer=DummyTracer(), ml_app="test_app", api_key="test_api_key")
         pid = os.fork()