[python] implement /flush endpoint in python weblogs

utils/_context/_scenarios/endtoend.py

@@ -314,13 +314,11 @@ def configure(self, config: pytest.Config):
                 self.library_interface_timeout = 25
             elif library in ("golang",):
                 self.library_interface_timeout = 10
-            elif library in ("nodejs", "ruby"):
+            elif library in ("nodejs", "python", "ruby"):
                 self.library_interface_timeout = 0
             elif library in ("php",):
                 # possibly something weird on obfuscator, let increase the delay for now
                 self.library_interface_timeout = 10
-            elif library in ("python",):
-                self.library_interface_timeout = 5
             else:
                 self.library_interface_timeout = 40
         else:

utils/_context/containers.py

@@ -1089,9 +1089,9 @@ def warmup_request(self, timeout: int = 10):
     def flush(self) -> None:
         if self.library.name not in (
             "nodejs",
+            "python",
             "ruby",
         ):
-            # only nodejs and ruby supports it
             return
 
         try:

utils/build/docker/python/django/app/urls.py

@@ -5,8 +5,8 @@
 import random
 import shlex
 import subprocess
-import xmltodict
 import sys
+import xmltodict
 import boto3
 import django
 import httpx
@@ -34,6 +34,7 @@
 import ddtrace
 
 from ddtrace.appsec import trace_utils as ato_user_sdk_v1
+from ddtrace.internal import telemetry
 
 try:
     from ddtrace.appsec import track_user_sdk
@@ -1192,6 +1193,18 @@ def stripe_webhook(request):
         return JsonResponse({"error": str(e)}, status=403)
 
 
+def flush(request):
+    # NOTE: If anything needs to be flushed here before the test suite ends,
+    #       this is the place to do it.
+    #       See https://github.com/DataDog/system-tests/blob/main/docs/edit/flushing.md
+    tracer.flush()
+    # app_shutdown() sends a force flush with an app-closing event so the agent
+    # finalises the telemetry batch, then disables the writer (safe: /flush is
+    # called only once, at the end of the test suite).
+    telemetry.telemetry_writer.app_shutdown()
+    return HttpResponse("OK")
+
+
 urlpatterns = [
     path("", hello_world),
     path("api_security/sampling/<int:status_code>", api_security_sampling_status),
@@ -1294,4 +1307,5 @@ def stripe_webhook(request):
     path("stripe/create_checkout_session", stripe_create_checkout_session),
     path("stripe/create_payment_intent", stripe_create_payment_intent),
     path("stripe/webhook", stripe_webhook),
+    path("flush", flush),
 ]

utils/build/docker/python/fastapi/main.py

@@ -37,6 +37,7 @@
 
 import ddtrace
 from ddtrace.appsec import trace_utils as appsec_trace_utils
+from ddtrace.internal import telemetry
 from openfeature import api
 from ddtrace.openfeature import DataDogProvider
 from openfeature.evaluation_context import EvaluationContext
@@ -1400,3 +1401,16 @@ async def stripe_webhook(request: Request):
         return JSONResponse(event.data.object)
     except Exception as e:
         return JSONResponse({"error": str(e)}, status_code=403)
+
+
+@app.get("/flush", response_class=PlainTextResponse)
+def flush():
+    # NOTE: If anything needs to be flushed here before the test suite ends,
+    #       this is the place to do it.
+    #       See https://github.com/DataDog/system-tests/blob/main/docs/edit/flushing.md
+    tracer.flush()
+    # app_shutdown() sends a force flush with an app-closing event so the agent
+    # finalises the telemetry batch, then disables the writer (safe: /flush is
+    # called only once, at the end of the test suite).
+    telemetry.telemetry_writer.app_shutdown()
+    return "OK"

utils/build/docker/python/flask/app.py

@@ -97,6 +97,7 @@
     config._logs_injection = True
 
 from ddtrace.appsec import trace_utils as appsec_trace_utils
+from ddtrace.internal import telemetry
 from ddtrace.internal.datastreams import data_streams_processor
 from ddtrace.internal.datastreams.processor import DsmPathwayCodec
 from ddtrace.data_streams import set_consume_checkpoint
@@ -2234,3 +2235,16 @@ def stripe_webhook():
         return jsonify(event.data.object)
     except Exception as e:
         return jsonify({"error": str(e)}), 403
+
+
+@app.route("/flush")
+def flush():
+    # NOTE: If anything needs to be flushed here before the test suite ends,
+    #       this is the place to do it.
+    #       See https://github.com/DataDog/system-tests/blob/main/docs/edit/flushing.md
+    tracer.flush()
+    # app_shutdown() sends a force flush with an app-closing event so the agent
+    # finalises the telemetry batch, then disables the writer (safe: /flush is
+    # called only once, at the end of the test suite).
+    telemetry.telemetry_writer.app_shutdown()
+    return Response("OK")

utils/build/docker/python/tornado/main.py

@@ -21,6 +21,7 @@
 from ddtrace.appsec import trace_utils as appsec_trace_utils
 from ddtrace.appsec import track_user_sdk
 from ddtrace.contrib.trace_utils import set_user
+from ddtrace.internal import telemetry
 from ddtrace.openfeature import DataDogProvider
 from ddtrace.trace import tracer
 from iast import (
@@ -867,6 +868,19 @@ async def post(self) -> None:
             self.write(json.dumps({"error": str(e)}))
 
 
+class FlushHandler(BaseHandler):
+    def get(self) -> None:
+        # NOTE: If anything needs to be flushed here before the test suite ends,
+        #       this is the place to do it.
+        #       See https://github.com/DataDog/system-tests/blob/main/docs/edit/flushing.md
+        tracer.flush()
+        # app_shutdown() sends a force flush with an app-closing event so the agent
+        # finalises the telemetry batch, then disables the writer (safe: /flush is
+        # called only once, at the end of the test suite).
+        telemetry.telemetry_writer.app_shutdown()
+        self.write("OK")
+
+
 class ExternalRequestHandler(BaseHandler):
     SUPPORTED_METHODS = ("GET", "POST", "PUT", "TRACE")
 
@@ -1115,6 +1129,8 @@ def make_app() -> Application:
             (r"/stripe/create_checkout_session", StripeCreateCheckoutSessionHandler),
             (r"/stripe/create_payment_intent", StripeCreatePaymentIntentHandler),
             (r"/stripe/webhook", StripeWebhookHandler),
+            # Flush endpoint
+            (r"/flush", FlushHandler),
         ],
         debug=False,
         cookie_secret="just_for_tests",  # noqa: S106