Repository navigation
Conversation
Use the tracer flush contract within the remaining Lambda budget while preserving handler outcomes and warm providers. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Keep invalid timeout defaults numeric, initialize the smoke provider before invocation, and assert older tracer compatibility. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Incorporate upstream Lambda v130 and instrumentation updates for FFL-3412. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
![datadog-official[bot]](https://avatars.githubusercontent.com/in/1358509?s=60&v=4){kind=small}
There was a problem hiding this comment.
More details
Static review identified no actionable defect in cleanup placement, timeout budgeting, or older-tracer compatibility. Runtime behavior and deployed AWS freeze/thaw delivery remain unverified.
🤖 Bits Code Review · Commit f6f08ba
| except Exception as e: | ||
| logger.error(format_err_with_traceback(e)) | ||
| finally: | ||
| flush_feature_flag_telemetry(context) |
There was a problem hiding this comment.
_after already catches and logs everything, so the outer try/except is a no-op. Can we call flush_feature_flag_telemetry(context) inside _after alongside LLMObs.flush(), before flush_stats/flush_extension? That matches the other flushes. If an early _after failure skipping flushes is a concern, it applies to all of them and belongs in a separate change. The BaseException re-raise and the submit_errors_metric hardening also look unrelated to this PR.
The other changes here are also unnecessary and can be reverted.
| "Skipping feature flag telemetry flush: no invocation time remains" | ||
| ) | ||
| return | ||
| outcome = _telemetry.flush(budget_ms / 1000.0) |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P1] Back off after repeated timeouts that use the full cap. Without a backoff, a slow or unreachable intake adds up to the whole cap, 500 ms by default, to every invocation that has unsent flag events.
- For a positive budget,
flush()in fix(openfeature): add bounded invocation-end EVP flush dd-trace-py#21135 returnstimed_outonly after its deadline passes, so each timeout costs the full budget. - The tracer's direct-intake route has no cooldown. Only agentless-local routes have one (
FeatureFlagEVPRouteSelector.send()inddtrace/internal/openfeature/_evp_transport.py). - Before this change, the wrapper did not wait for feature flag intake at invocation end. During an intake incident, functions with unsent flag events can pay up to the cap on each invocation, and Lambda bills that time.
Datadog's serverless libraries already back off in this situation. datadogpy, which this wrapper uses to send metrics without the extension, stops sending for 300 s after 3 timeouts. datadog-lambda-go opens a circuit breaker for 60 s after more than 4 failures in 30 s. The suggested fix takes the threshold from datadogpy and the window from datadog-lambda-go: after 3 full-cap timeouts with no completed or failed flush between them, it skips the wait for 60 s. A timeout from a short remaining-time budget neither counts nor resets the streak. During the window, invocations skip the invocation-end flush, and the background writers keep sending on their own schedule. After the window, the next invocation flushes again.
Keep the 500 ms default and the 100 ms margin. The industry standard for serverless SDKs is to flush before the invocation returns, with caps of 1 s or longer, so 500 ms is already conservative. A reserve of about 100 ms of remaining time matches published guidance.
Precedent — Datadog libraries that already back off
- datadogpy:
_max_timeouts = 3and_backoff_period = 300indatadog/api/__init__.py.api_client.pylogs "Max number of datadog timeouts exceeded, backing off for %s seconds" when the timeout counter reaches the maximum. - datadog-lambda-go:
defaultCircuitBreakerInterval = 30s,defaultCircuitBreakerTimeout = 60sanddefaultCircuitBreakerTotalFailures = 4ininternal/metrics/constants.go. The breaker trips whenTotalFailures > 4.
A timeout caused by a short remaining-time budget does not count, because it says nothing about the intake. Any outcome other than timed_out resets the count.
Suggested fix and tests — each test fails when its branch is removed
--- a/datadog_lambda/feature_flags.py
+++ b/datadog_lambda/feature_flags.py
@@ -2,6 +2,7 @@
import logging
import math
+import time
from datadog_lambda.config import config
@@ -13,9 +14,15 @@ except ImportError:
logger = logging.getLogger(__name__)
SAFETY_MARGIN_MS = 100.0
+# After repeated full-cap timeouts, stop adding the wait to every invocation for a while.
+MAX_CONSECUTIVE_TIMEOUTS = 3
+TIMEOUT_BACKOFF_S = 60.0
+_consecutive_timeouts = 0
+_skip_until = 0.0
def flush_feature_flag_telemetry(context):
+ global _consecutive_timeouts, _skip_until
try:
if _telemetry is None or getattr(_telemetry, "FLUSH_API_VERSION", None) != 1:
return
@@ -30,7 +37,17 @@ def flush_feature_flag_telemetry(context):
"Skipping feature flag telemetry flush: no invocation time remains"
)
return
+ if time.time() < _skip_until:
+ logger.debug("Skipping feature flag telemetry flush: backing off")
+ return
outcome = _telemetry.flush(budget_ms / 1000.0)
+ if outcome != "timed_out":
+ _consecutive_timeouts = 0
+ elif budget_ms >= cap_ms:
+ _consecutive_timeouts += 1
+ if _consecutive_timeouts >= MAX_CONSECUTIVE_TIMEOUTS:
+ _consecutive_timeouts = 0
+ _skip_until = time.time() + TIMEOUT_BACKOFF_S
if outcome != "completed":
logger.debug("Feature flag telemetry flush incomplete: %s", outcome)
except Exception:
--- a/tests/test_feature_flags.py
+++ b/tests/test_feature_flags.py
@@ -9,6 +9,37 @@ from datadog_lambda.config import Config
class FeatureFlagFlushTest(unittest.TestCase):
+ def setUp(self):
+ for name, value in [("_consecutive_timeouts", 0), ("_skip_until", 0.0)]:
+ patcher = patch.object(feature_flags, name, value)
+ patcher.start()
+ self.addCleanup(patcher.stop)
+
+ def _flush_with_outcomes(self, outcomes, remaining_ms=30000):
+ context = Mock(get_remaining_time_in_millis=lambda: remaining_ms)
+ with patch.object(feature_flags, "_telemetry") as telemetry:
+ telemetry.FLUSH_API_VERSION = 1
+ telemetry.flush.side_effect = outcomes
+ for _ in outcomes:
+ feature_flags.flush_feature_flag_telemetry(context)
+ return telemetry.flush.call_count
+
+ def test_repeated_full_budget_timeouts_back_off(self):
+ with patch.object(feature_flags.time, "time", return_value=1000.0) as now:
+ self.assertEqual(self._flush_with_outcomes(["timed_out"] * 4), 3)
+ now.return_value = 1000.0 + feature_flags.TIMEOUT_BACKOFF_S
+ self.assertEqual(self._flush_with_outcomes(["completed"]), 1)
+
+ def test_success_resets_timeout_count(self):
+ outcomes = ["timed_out", "timed_out", "completed", "timed_out", "timed_out"]
+ self.assertEqual(self._flush_with_outcomes(outcomes), 5)
+ self.assertEqual(feature_flags._skip_until, 0.0)
+
+ def test_short_budget_timeouts_do_not_back_off(self):
+ self.assertEqual(
+ self._flush_with_outcomes(["timed_out"] * 4, remaining_ms=200), 4
+ )
+
def test_budget(self):
for remaining, cap, expected in [
(5000, 500, 0.5),The change adds module state, so FeatureFlagFlushTest needs the setUp shown, which resets it between tests. The full suite passes (1000 tests). Each new test fails when I remove its branch: the backoff, the full-cap check, or the reset on success.
| except BaseException as e: | ||
| if "BlockingException" in type(e).__name__: | ||
| self.blocking_response = get_asm_blocked_response(self.event_source) | ||
| else: |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] Add a test for this branch, and mention the change in the PR description. A handler's SystemExit or KeyboardInterrupt now propagates.
Since #835 (released in v8.126.0), this branch swallowed SystemExit, KeyboardInterrupt and any other BaseException that is not an Exception and whose class name does not contain BlockingException. The wrapper then returned self.response, which _before resets to None. Unless AppSec replaced the response, sys.exit(1) in a handler produced a successful null response. This else: raise restores the behaviour from before #835. That is the right behaviour, but handlers that call sys.exit() will now fail the invocation, and no test in tests/ raises SystemExit or KeyboardInterrupt.
Suggested test — fails with main's behaviour
--- a/tests/test_feature_flags.py
+++ b/tests/test_feature_flags.py
@@ -105,6 +105,25 @@ class FeatureFlagFlushTest(unittest.TestCase):
"Feature flag telemetry flush incomplete: %s", "timed_out"
)
+ def test_non_blocking_base_exception_propagates_and_flushes(self):
+ for raised in [SystemExit(3), KeyboardInterrupt()]:
+ decorated = object.__new__(wrapper._LambdaDecorator)
+ decorated.func = Mock(side_effect=raised)
+ decorated.response = None
+ decorated.blocking_response = None
+ decorated.span = None
+ decorated._before = Mock()
+ decorated._after = Mock()
+ context = Mock(get_remaining_time_in_millis=lambda: 30000)
+ with self.subTest(raised=type(raised).__name__), patch.object(
+ wrapper, "flush_feature_flag_telemetry"
+ ) as flush:
+ with self.assertRaises(type(raised)) as caught:
+ decorated({}, context)
+ self.assertIs(caught.exception, raised)
+ decorated._after.assert_called_once()
+ flush.assert_called_once_with(context)
+
def test_wrapper_preserves_result_and_exception_when_cleanup_fails(self):
for raises in [False, True]:
original = ValueError("handler error")With else: raise removed, which is main's behaviour, both subtests fail with SystemExit not raised and KeyboardInterrupt not raised. The full suite passes with the test added.
|
|
||
| [1]: https://www.datadoghq.com/blog/troubleshoot-lambda-function-request-response-payloads/ | ||
|
|
||
| ## Feature flag telemetry at invocation end |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] Move this section above "Opening Issues", add the variable to the configuration table, and drop the release-status paragraph. The README ships to PyPI.
- The section is after
## Licenseand the link definitions. - The variable is missing from the "Configuration" table.
pyproject.tomlsetsreadme = "README.md", so the "pending coordinated release" paragraph goes to PyPI and becomes stale.- With the current pins, Python 3.8 and 3.9 do not get this behavior:
pyproject.tomlpins them toddtrace>=3.19.1,<4, and the tracer PR targets ddtrace 4.x. The README does not say so.
Suggested README — assumes the rename and backoff comments are taken
--- a/README.md
+++ b/README.md
@@ -31,8 +31,17 @@ Besides the environment variables supported by dd-trace-py, the datadog-lambda-p
| DD_CAPTURE_LAMBDA_PAYLOAD_MAX_DEPTH | Determines the level of detail captured from AWS Lambda payloads, which are then assigned as tags for the `aws.lambda` span. It specifies the nesting depth of the JSON payload structure to process. Once the specified maximum depth is reached, the tag's value is set to the stringified value of any nested elements beyond this level. <br> For example, given the input payload: <pre>{<br> "lv1" : {<br> "lv2": {<br> "lv3": "val"<br> }<br> }<br>}</pre> If the depth is set to `2`, the resulting tag's key is set to `function.request.lv1.lv2` and the value is `{\"lv3\": \"val\"}`. <br> If the depth is set to `0`, the resulting tag's key is set to `function.request` and value is `{\"lv1\":{\"lv2\":{\"lv3\": \"val\"}}}` | `10` |
| DD_EXCEPTION_REPLAY_ENABLED | When set to `true`, the Lambda will run with Error Tracking Exception Replay enabled, capturing local variables. | `false` |
| DD_DSM_EXCHANGE_NAME | If you are using Datadog Data Streams Monitoring and propagating context through Amazon Event Bridge, this variable specifies the name of the EventBridge EventBus. The name of the bus is *not* passed through in the event payload and therefore cannot be automatically inferred at consume time. | `false` |
+| DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS | Maximum time, in milliseconds, that the wrapper waits at the end of an invocation to send pending feature flag evaluation and exposure events. `0` disables the wait. See [Feature flag telemetry at invocation end](#feature-flag-telemetry-at-invocation-end). | `500` |
+## Feature flag telemetry at invocation end
+
+The wrapper tries to send pending feature flag evaluation and exposure events before it returns a response or rethrows a handler exception. No handler code or provider shutdown is required, and providers stay available for warm invocations.
+
+`DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS` limits the added wait. The wait is also limited to the remaining invocation time minus 100 ms. After three timeouts in a row that use the full limit, the wrapper skips the wait for 60 seconds.
+
+Delivery is best effort. A timeout stops the wait but does not cancel a request in flight, which can complete after a warm thaw. Abrupt termination can still lose buffered events. This requires ddtrace `<version>` or later, so it is available on Python 3.10 and later.
+
## Opening Issues
If you encounter a bug with this package, we want to hear about it. Before opening a new issue, search the existing issues to avoid duplicates.
@@ -76,27 +85,3 @@ Unless explicitly stated otherwise all files in this repository are licensed und
This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2019 Datadog, Inc.
[1]: https://www.datadoghq.com/blog/troubleshoot-lambda-function-request-response-payloads/
-
-## Feature flag telemetry at invocation end
-
-With a tracer and Lambda layer that support automatic feature flag flushing,
-the wrapper attempts to send pending evaluation and exposure events before
-returning a response or rethrowing a handler exception. No additional handler
-code or provider shutdown is required; providers and polling remain available
-for warm invocations.
-
-`DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS` sets the maximum added wait in milliseconds
-(default: `500`; `0` disables waiting). The actual budget is the smaller of that
-cap and the invocation's remaining time minus a 100 ms safety margin. The wrapper
-skips waiting when that budget is exhausted or unavailable. Incomplete flushes
-are reported in debug logs without changing the handler's result.
-
-Delivery is best effort. A timeout stops waiting but does not cancel or retry
-in-flight requests; they may complete after a warm thaw. Abrupt termination can
-still lose buffered events. Endpoint acceptance does not guarantee immediate
-visibility in the UI. Older tracers safely skip the integration.
-
-Release status: this capability requires the tracer's internal flush contract
-version 1. The first supported tracer/library/layer combination is pending
-coordinated release and deployed validation; an older published layer does not
-provide this behavior merely by setting the timeout variable.If you don't take those, keep the current name and drop the backoff sentence. Replace <version> with the first ddtrace release that contains DataDog/dd-trace-py#21135.
| return | ||
| outcome = _telemetry.flush(budget_ms / 1000.0) | ||
| if outcome != "completed": | ||
| logger.debug("Feature flag telemetry flush incomplete: %s", outcome) |
There was a problem hiding this comment.
I checked your other PR, do we even need to return something from the _telemetry.flush? If all we're doing is logging, then why not just log from dd-trace-py?
| logger.debug("Feature flag telemetry flush incomplete: %s", outcome) | ||
| except Exception: | ||
| # Never replace the response or the original handler exception. | ||
| logger.debug("Feature flag telemetry flush failed", exc_info=True) |
There was a problem hiding this comment.
To be very honest, this method just seems really over engineered as it is right now. Why the check of FLUSH_API_VERSION? And if we route the flushed telemetry data through the lambda extension (instead of agentless), the outbound flush http request will be very fast and wouldn't need all this extra overhead.
Instead, this PR could simply be these two additions to datadog_lambda/wrapper.py and the config addition:
# added to datadog_lambda/config.py
feature_flags_enabled = _get_env("DD_FEATURE_FLAGS_ENABLED", "true", as_bool)# at top of file
_feature_flags = None
if config.feature_flags_enabled:
try:
from ddtrace.internal.openfeature import _telemetry as _feature_flags
except ImportError:
pass# at end of self._after
if _feature_flags is not None:
_feature_flags.flush()There was a problem hiding this comment.
Actually, if we just call periodic on the writer directly, this PR could become just 5d3d3e3 and you wouldn't even need the ddtrace PR.
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
Calling periodic() directly is much simpler, and exception replay is good precedent. Before we switch, two gaps in 5d3d3e3 need a decision: it has no wait budget, and it does not wait for a send that is already in flight. I ran it against dd-trace-py main to check both.
- No wait budget. Each
periodic()sends synchronously, with a 2 s socket timeout per writer, and nothing limits the total wait to the remaining invocation time. With a local endpoint that takes 1.5 s per request, the wrapper took about 3 s with 1000 ms remaining. With only that much time left, Lambda would end the invocation with a timeout. The tracer's direct-intake route has no cooldown, so a slow intake can affect every invocation that still has unsent events. - An in-flight send is not awaited. When a writer's own periodic tick has already taken the batch,
periodic()finds an empty buffer and returns. In the test, the wrapper returned in 1–2 ms while the server still held that batch's request, so the sandbox can freeze in the middle of the send.
On routing: with the default agentless configuration source, the tracer uses a local route only if /info advertises an EVP proxy route and evp_proxy_allowed_headers. The extension's /info lists neither, so these events go directly to the intake over HTTPS with DD_API_KEY, not through the extension.
A possible middle path: keep your wiring, and bound the wait in the wrapper, for example by running the two periodic() calls on a helper thread and joining it with a timeout (the smaller of a cap and the remaining time minus 100 ms, with the backoff from this comment). Then add a flush lock to the tracer writers, so that a second periodic() waits for an in-flight send. The flush lock needs care in ExposureWriter.enqueue, which calls periodic() while it holds _lock on main. I have not built this combination yet. I can prototype it if that direction sounds right.
Results — 5d3d3e3 with dd-trace-py main 9b07368c88, Python 3.12, three runs
Both writers pointed at a local HTTP server through a remote-config route selector, which uses the local route without discovery. The handler enqueued one evaluation and one exposure. For the in-flight case, a background periodic() call stands in for the writer's own tick, and the server holds that request until the wrapper returns.
| Case | Result |
|---|---|
| Endpoint answers at once, 30 s remaining | wrapper took 4–5 ms; the server received both batches |
| Endpoint takes 1.5 s per request, 1000 ms remaining | wrapper took 3014–3026 ms; the server received both batches |
| Background send in progress, server holding it | wrapper returned in 1–2 ms; that batch was still unacknowledged |
Probe test — runs inside this repo's TestDatadogLambdaWrapper setup
"""Probes for 5d3d3e3. Run: pytest -s tests/test_review_probe_rey.py -k test_probe"""
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from unittest.mock import Mock
import datadog_lambda.wrapper as wrapper
from ddtrace.internal.openfeature import _config as ffe_config
from ddtrace.internal.openfeature import writer as ffe_writer_mod
from ddtrace.internal.openfeature._evp_transport import EVP_ORIGIN_HEADERS
from ddtrace.internal.openfeature._evp_transport import FeatureFlagEVPRouteSelector
from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter
from ddtrace.internal.openfeature._flagevaluation_writer import _EvalEvent
from ddtrace.internal.openfeature.writer import ExposureWriter
from ddtrace.internal.settings.openfeature import REMOTE_CONFIG
from tests.test_wrapper import TestDatadogLambdaWrapper
from tests.utils import get_mock_context
class ReyProbe(TestDatadogLambdaWrapper):
def _server(self, delay, gate=None, arrived=None):
posts = []
class Handler(BaseHTTPRequestHandler):
def do_POST(self):
self.rfile.read(int(self.headers["Content-Length"]))
if arrived is not None:
arrived.set()
if gate is not None:
gate.wait(5)
time.sleep(delay)
posts.append((self.path, time.monotonic()))
self.send_response(202)
self.end_headers()
def log_message(self, *args):
pass
srv = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=srv.serve_forever, daemon=True).start()
self.addCleanup(srv.shutdown)
return srv, posts
def _writers(self, srv):
sel = FeatureFlagEVPRouteSelector(
configuration_source=REMOTE_CONFIG,
agent_url="http://127.0.0.1:%d" % srv.server_port,
api_key=None,
site="datadoghq.com",
info_provider=lambda _: {
"endpoints": ("/evp_proxy/v2/",),
"evp_proxy_allowed_headers": tuple(EVP_ORIGIN_HEADERS),
},
)
ev = FlagEvaluationWriter(interval=3600.0, route_selector=sel)
ex = ExposureWriter(enabled=True, interval=3600.0, route_selector=sel)
previous = ffe_writer_mod._EXPOSURE_WRITER
ffe_writer_mod._EXPOSURE_WRITER = ex
self.addCleanup(setattr, ffe_writer_mod, "_EXPOSURE_WRITER", previous)
provider = Mock(_flag_eval_evp_writer=ev)
ffe_config._provider_instances.append(provider)
self.addCleanup(ffe_config._provider_instances.remove, provider)
return ev, ex
def _invoke(self, handler, remaining_ms):
ctx = get_mock_context()
ctx.get_remaining_time_in_millis = lambda: remaining_ms
decorated = wrapper.datadog_lambda_wrapper(handler)
started = time.monotonic()
result = decorated({}, ctx)
return result, (time.monotonic() - started) * 1000
def _handler(self, ev, ex):
def handler(event, context):
ev.enqueue(
_EvalEvent("flag", "on", "allocation", "subject", {}, False, "", 1)
)
ex.enqueue({"timestamp": 1})
return {"ok": True}
return handler
def test_probe_fast_intake(self):
srv, posts = self._server(0.0)
ev, ex = self._writers(srv)
_, ms = self._invoke(self._handler(ev, ex), 30000)
print(
"RESULT fast: wrapper %.0f ms, posts=%d %s"
% (ms, len(posts), [p for p, _ in posts])
)
def test_probe_slow_intake_ignores_remaining_time(self):
srv, posts = self._server(1.5)
ev, ex = self._writers(srv)
_, ms = self._invoke(self._handler(ev, ex), 1000)
print(
"RESULT slow: wrapper %.0f ms with 1000 ms remaining, posts=%d"
% (ms, len(posts))
)
def test_probe_inflight_send_not_awaited(self):
gate, arrived = threading.Event(), threading.Event()
srv, posts = self._server(0.0, gate=gate, arrived=arrived)
ev, ex = self._writers(srv)
ex.enqueue({"timestamp": 1})
# Stands in for the writer's own 1 s tick, which has taken the batch.
background = threading.Thread(target=ex.periodic)
background.start()
assert arrived.wait(2) # the server now holds that request
_, ms = self._invoke(lambda e, c: {"ok": True}, 30000)
pending = background.is_alive() and not posts
gate.set()
background.join(5)
print(
"RESULT inflight: wrapper returned in %.0f ms while the batch was unacknowledged: %s"
% (ms, pending)
)Routing — why the events skip the extension
- The extension's
info()handler lists its endpoints inbottlecap/src/traces/trace_agent.rs. The list has no/evp_proxy/v2or/evp_proxy/v4base path, and the response has noevp_proxy_allowed_headers. - dd-trace-py's
FeatureFlagEVPRouteSelector._discover_local_route()(ddtrace/internal/openfeature/_evp_transport.py) returns no local route unless/infoadvertises one of those base paths and allows theDD-EVP-ORIGINheaders. The selector then uses direct intake withDD_API_KEY.
There was a problem hiding this comment.
This hasn't been an issue for any of the other flushes we do at the end of invocations, but none of those are agentless. I'd like to test out adding the EVP proxy to /info and then seeing if calling periodic() directly causes any issues.
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
Your plan works with today's tracer. I tested it from a real dd-trace-py process through the patched extension's router to its relay queue. One change in bottlecap/src/traces/trace_agent.rs (patch below) is enough, and both periodic() calls returned in 0.8–1.9 ms across three runs. I did not run the extension's flush to the intake.
With dd-trace-py's default agentless configuration source, the tracer needs four things from the extension:
- List the bare base path
/evp_proxy/v2/in/infoendpoints. Full routes alone are not enough: dd-trace-py compares each endpoint with the base path (a trailing slash is ignored) before it uses an EVP route. - Add
evp_proxy_allowed_headerswithDD-EVP-ORIGINandDD-EVP-ORIGIN-VERSION. The match ignores case. Without these headers, the tracer sends directly to the intake whenDD_API_KEYis set, and has no route otherwise. - Route
/evp_proxy/v2/api/v2/flagevaluationand/evp_proxy/v2/api/v2/exposurestoevent-platform-intake. Usehandle_proxy, as the LLM Obs routes do, withevent-platform-intakeas the backend domain.handle_proxyalready forwards the request headers, and the proxy flusher addsDD-API-KEY. - Do not list
/evp_proxy/v4/unless you also route it. When the tracer sees v4, it sends to/evp_proxy/v4/...instead.
What I ran, with dd-trace-py main 9b07368c88 and bottlecap main 8a2cc02:
- Patched bottlecap: the real tracer chose the local
/evp_proxy/v2route from the real/info. Bottlecap queued both batches forevent-platform-intakewith the origin headers intact. I passed no API key to the tracer's route selector. The extension adds its ownDD-API-KEYwhen it flushes (proxy_flusher.rs), so this route should also serve functions that keep the key inDD_API_KEY_SECRET_ARN. I did not run that flush. - Unpatched bottlecap: with no API key, the tracer found no route, and nothing reached bottlecap. With
DD_API_KEYset, the tracer uses direct HTTPS. - Fallback: if an advertised route answers 404 or 405, the tracer replays that batch directly to the intake (when
DD_API_KEYis set) and uses direct for later batches. I saw the replay attempt. My stand-in intake was plain HTTP, which the tracer refuses, so I did not see the replay succeed.
Three side effects to plan for:
- Other runtimes move too. Java and Go look for the exact
/evp_proxy/v2/endpoint when they choose agent mode for LLM Observability. WithDD_LLMOBS_AGENTLESS_ENABLEDunset, they would switch from agentless to the extension in Lambda. The extension already relays those LLM Obs routes, but they are worth a smoke test. dd-trace-py (substring match) and Node (prefix match) already send LLM Obs to the extension today. - Loading flags still needs
DD_API_KEY. dd-trace-py's default agentless configuration source does not start without it (create_agentless_source), so a function that only hasDD_API_KEY_SECRET_ARNgets its events relayed but loads no flags. That is a separate gap. - The in-flight case from my earlier comment remains. In my runs the two local sends took 0.8–1.9 ms together; I did not measure the race itself. Functions without the extension still send directly and still need a wait cap.
Patch — cargo fmt clean, no clippy findings in trace_agent.rs, and the 11 trace_agent tests pass, including the 2 new ones
--- a/bottlecap/src/traces/trace_agent.rs
+++ b/bottlecap/src/traces/trace_agent.rs
@@ -64,6 +64,10 @@ const LLM_OBS_EVAL_METRIC_ENDPOINT_PATH: &str = "/evp_proxy/v2/api/intake/llm-ob
const LLM_OBS_EVAL_METRIC_ENDPOINT_PATH_V2: &str =
"/evp_proxy/v2/api/intake/llm-obs/v2/eval-metric";
const LLM_OBS_SPANS_ENDPOINT_PATH: &str = "/evp_proxy/v2/api/v2/llmobs";
+// Some tracers need this base path in /info before they use EVP proxy routes.
+const EVP_PROXY_V2_BASE_PATH: &str = "/evp_proxy/v2/";
+const FFE_FLAG_EVALUATION_ENDPOINT_PATH: &str = "/evp_proxy/v2/api/v2/flagevaluation";
+const FFE_EXPOSURES_ENDPOINT_PATH: &str = "/evp_proxy/v2/api/v2/exposures";
const INFO_ENDPOINT_PATH: &str = "/info";
const V1_DEBUGGER_ENDPOINT_PATH: &str = "/debugger/v1/input";
const V2_DEBUGGER_ENDPOINT_PATH: &str = "/debugger/v2/input";
@@ -73,6 +77,8 @@ const INSTRUMENTATION_ENDPOINT_PATH: &str = "/telemetry/proxy/api/v2/apmtelemetr
// Intake endpoints
const DSM_INTAKE_PATH: &str = "/api/v0.1/pipeline_stats";
const LLM_OBS_SPANS_INTAKE_PATH: &str = "/api/v2/llmobs";
+const FFE_FLAG_EVALUATION_INTAKE_PATH: &str = "/api/v2/flagevaluation";
+const FFE_EXPOSURES_INTAKE_PATH: &str = "/api/v2/exposures";
const LLM_OBS_EVAL_METRIC_INTAKE_PATH: &str = "/api/intake/llm-obs/v1/eval-metric";
const LLM_OBS_EVAL_METRIC_INTAKE_PATH_V2: &str = "/api/intake/llm-obs/v2/eval-metric";
const PROFILING_INTAKE_PATH: &str = "/api/v2/profile";
@@ -452,6 +458,11 @@ impl TraceAgent {
post(Self::llm_obs_eval_metric_proxy_v2),
)
.route(LLM_OBS_SPANS_ENDPOINT_PATH, post(Self::llm_obs_spans_proxy))
+ .route(
+ FFE_FLAG_EVALUATION_ENDPOINT_PATH,
+ post(Self::ffe_flag_evaluation_proxy),
+ )
+ .route(FFE_EXPOSURES_ENDPOINT_PATH, post(Self::ffe_exposures_proxy))
.route(V1_DEBUGGER_ENDPOINT_PATH, post(Self::debugger_logs_proxy))
.route(V2_DEBUGGER_ENDPOINT_PATH, post(Self::debugger_intake_proxy))
.route(
@@ -594,6 +605,33 @@ impl TraceAgent {
.await
}
+ async fn ffe_flag_evaluation_proxy(
+ State(state): State<ProxyState>,
+ request: Request,
+ ) -> Response {
+ Self::handle_proxy(
+ state.config,
+ state.proxy_aggregator,
+ request,
+ "event-platform-intake",
+ FFE_FLAG_EVALUATION_INTAKE_PATH,
+ "ffe_flag_evaluation",
+ )
+ .await
+ }
+
+ async fn ffe_exposures_proxy(State(state): State<ProxyState>, request: Request) -> Response {
+ Self::handle_proxy(
+ state.config,
+ state.proxy_aggregator,
+ request,
+ "event-platform-intake",
+ FFE_EXPOSURES_INTAKE_PATH,
+ "ffe_exposures",
+ )
+ .await
+ }
+
// Used for `/debugger/v1/input` in Exception Replay
async fn debugger_logs_proxy(State(state): State<ProxyState>, request: Request) -> Response {
Self::handle_proxy(
@@ -646,10 +684,15 @@ impl TraceAgent {
LLM_OBS_EVAL_METRIC_ENDPOINT_PATH,
LLM_OBS_EVAL_METRIC_ENDPOINT_PATH_V2,
LLM_OBS_SPANS_ENDPOINT_PATH,
+ EVP_PROXY_V2_BASE_PATH,
+ FFE_FLAG_EVALUATION_ENDPOINT_PATH,
+ FFE_EXPOSURES_ENDPOINT_PATH,
V1_DEBUGGER_ENDPOINT_PATH,
V2_DEBUGGER_ENDPOINT_PATH,
DEBUGGER_DIAGNOSTICS_ENDPOINT_PATH,
],
+ // handle_proxy forwards request headers; tracers check these before using EVP.
+ "evp_proxy_allowed_headers": ["DD-EVP-ORIGIN", "DD-EVP-ORIGIN-VERSION"],
"client_drop_p0s": false,
}
);
@@ -1286,4 +1329,66 @@ mod tests {
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
+
+ #[tokio::test]
+ async fn info_advertises_evp_proxy_for_feature_flags() {
+ let response = TraceAgent::info().await;
+ let body = axum::body::to_bytes(response.into_body(), usize::MAX)
+ .await
+ .expect("info body");
+ let info: serde_json::Value = serde_json::from_slice(&body).expect("info json");
+ let endpoints = info["endpoints"].as_array().expect("endpoints");
+ for path in [
+ EVP_PROXY_V2_BASE_PATH,
+ FFE_FLAG_EVALUATION_ENDPOINT_PATH,
+ FFE_EXPOSURES_ENDPOINT_PATH,
+ ] {
+ assert!(endpoints.iter().any(|e| e == path), "missing {path}");
+ }
+ let headers = info["evp_proxy_allowed_headers"]
+ .as_array()
+ .expect("evp_proxy_allowed_headers");
+ for header in ["DD-EVP-ORIGIN", "DD-EVP-ORIGIN-VERSION"] {
+ assert!(headers.iter().any(|h| h == header), "missing {header}");
+ }
+ }
+
+ #[tokio::test]
+ async fn feature_flag_events_are_relayed_to_event_platform_intake() {
+ let agent = build_test_agent();
+ let site = agent.config.site.clone();
+ let router = agent.make_router().expect("make_router");
+ for (path, intake) in [
+ (
+ FFE_FLAG_EVALUATION_ENDPOINT_PATH,
+ FFE_FLAG_EVALUATION_INTAKE_PATH,
+ ),
+ (FFE_EXPOSURES_ENDPOINT_PATH, FFE_EXPOSURES_INTAKE_PATH),
+ ] {
+ let response = router
+ .clone()
+ .oneshot(
+ Request::builder()
+ .method("POST")
+ .uri(path)
+ .header("Content-Type", "application/json")
+ .header("DD-EVP-ORIGIN", "dd-trace-py")
+ .header("DD-EVP-ORIGIN-VERSION", "4.17.0")
+ .header("X-Datadog-EVP-Subdomain", "event-platform-intake")
+ .body(Body::from(r#"{"events":[]}"#))
+ .expect("build request"),
+ )
+ .await
+ .expect("route response");
+ assert_eq!(response.status(), StatusCode::OK, "{path}");
+ let batch = agent.proxy_aggregator.lock().await.get_batch();
+ assert_eq!(batch.len(), 1, "{path}");
+ assert_eq!(
+ batch[0].target_url,
+ format!("https://event-platform-intake.{site}{intake}")
+ );
+ assert_eq!(batch[0].headers["dd-evp-origin"], "dd-trace-py");
+ assert_eq!(batch[0].headers["dd-evp-origin-version"], "4.17.0");
+ }
+ }
}Tracer-to-extension check — a real dd-trace-py process against the real router, patched and unpatched
I added a temporary ignored test that serves make_router() on a port, ran the tracer against it, and then printed what bottlecap queued for relay. The harness stops at that queue; it does not run the proxy flusher.
/// Serves the real router on 127.0.0.1:18126 for an out-of-process tracer, then prints what it
/// would relay. Run with `cargo test serve_router_for_tracer -- --ignored --nocapture`.
#[tokio::test]
#[ignore = "manual end-to-end harness"]
async fn serve_router_for_tracer() {
let agent = build_test_agent();
let router = agent.make_router().expect("make_router");
let listener = tokio::net::TcpListener::bind("127.0.0.1:18126")
.await
.expect("bind");
println!("E2E listening");
let server = tokio::spawn(async move { axum::serve(listener, router).await });
tokio::time::sleep(std::time::Duration::from_secs(25)).await;
server.abort();
for request in agent.proxy_aggregator.lock().await.get_batch() {
println!(
"E2E relay target={} origin={:?} version={:?} subdomain={:?} bytes={}",
request.target_url,
request.headers.get("dd-evp-origin"),
request.headers.get("dd-evp-origin-version"),
request.headers.get("x-datadog-evp-subdomain"),
request.body.len()
);
}
}"""Real dd-trace-py writers -> real bottlecap router on 127.0.0.1:18126. No API key passed to the tracer."""
import time
from ddtrace.internal.openfeature._evp_transport import FeatureFlagEVPRouteSelector
from ddtrace.internal.openfeature._flagevaluation_writer import FlagEvaluationWriter, _EvalEvent
from ddtrace.internal.openfeature.writer import ExposureWriter
from ddtrace.internal.settings.openfeature import AGENTLESS
sel = FeatureFlagEVPRouteSelector(
configuration_source=AGENTLESS, agent_url="http://127.0.0.1:18126", api_key=None, site="datadoghq.com"
)
route = sel.select()
print("TRACER route:", None if route is None else (route.intake, route.base_path))
ev = FlagEvaluationWriter(interval=3600.0, route_selector=sel)
ex = ExposureWriter(enabled=True, interval=3600.0, route_selector=sel)
ev.enqueue(_EvalEvent("flag", "on", "allocation", "subject", {}, False, "", 1))
ex.enqueue(
{
"timestamp": int(time.time() * 1000),
"allocation": {"key": "allocation"},
"flag": {"key": "flag"},
"variant": {"key": "on"},
"subject": {"id": "subject", "attributes": {}},
}
)
started = time.monotonic()
ev.periodic()
ex.periodic()
print("TRACER periodic() x2 took %.1f ms" % ((time.monotonic() - started) * 1000))| Bottlecap | Tracer route | Queued for relay |
|---|---|---|
| patched | local /evp_proxy/v2; both periodic() calls took 1.9, 0.9 and 0.8 ms in three runs |
https://event-platform-intake.<site>/api/v2/flagevaluation (312 bytes) and /api/v2/exposures (216 bytes), each with DD-EVP-ORIGIN: dd-trace-py, DD-EVP-ORIGIN-VERSION and X-Datadog-EVP-Subdomain: event-platform-intake |
| unpatched | none; both periodic() calls returned in 0.1 ms |
nothing |
The test config has an empty site, so the printed host was event-platform-intake. with no domain.
What the tracer checks — real dd-trace-py discovery against each /info shape
FeatureFlagEVPRouteSelector with the default agentless configuration source, pointed at a local /info:
/info shape |
DD_API_KEY set |
DD_API_KEY unset |
|---|---|---|
| bottlecap today | direct | no route |
| plus the two flag routes | direct | no route |
plus /evp_proxy/v2/, no allowed headers |
direct | no route |
plus /evp_proxy/v2/ and the two allowed headers |
local /evp_proxy/v2, direct fallback |
local /evp_proxy/v2 |
| same, header names in lower case | local /evp_proxy/v2, direct fallback |
local /evp_proxy/v2 |
same, plus /evp_proxy/v4/ |
local /evp_proxy/v4 |
local /evp_proxy/v4 |
With the remote-config source, the tracer uses the local v2 route without discovery. On the local route, the tracer sends Content-Type: application/json, DD-EVP-ORIGIN, DD-EVP-ORIGIN-VERSION and X-Datadog-EVP-Subdomain: event-platform-intake, and no DD-API-KEY. The checks live in FeatureFlagEVPRouteSelector._discover_local_route() in ddtrace/internal/openfeature/_evp_transport.py.
| try: | ||
| from ddtrace.internal.openfeature import _telemetry | ||
| except ImportError: | ||
| _telemetry = None |
There was a problem hiding this comment.
Once we get this PR more ready to merge, I'd like to run a before/after test on the cold start duration. We've found in the past that adding imports, when the code ultimately isn't being used here, can add significant amounts of time to the beginning of the first lambda invocation. From what I can tell, I don't think that will be the case here, but it will be good to know for certain. I'll run that testing once we get to a good point.
Use the shared DD_FEATURE_FLAGS_ prefix for the timeout setting and remove PR-specific README and smoke-runbook text. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Respect DD_FEATURE_FLAGS_ENABLED during import and cleanup, test the real wrapper contract with released dependencies, and distinguish expected skips in debug logs. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Skip invocation-end flush attempts for 60 seconds after three full-budget timeouts, keep failures isolated, and restore existing wrapper error handling. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Motivation
A customer asked about flushing feature-flag telemetry from Python Lambda functions (FFL-3412). Short invocations can return before background writers send evaluation and exposure events, leaving telemetry delayed by Lambda freezing or lost on termination.
This change automatically attempts a flush before the invocation finishes, without customers adding handler code. Providers remain available for warm invocations.
Changes
_after, including after handler errors. Earlier cleanup errors may skip the flush, following the existing cleanup behavior.DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS(default 500 ms; 0 disables waiting) and remaining invocation time minus a 100 ms safety margin.DD_FEATURE_FLAGS_ENABLED(defaulttrue); disabling it skips the flush helper import and invocation-end call.Testing
ddtrace4.1.1 (candidate-only test skipped). CI's pinned Black formatting checks and source Flake8 passed.Release coordination
Companion: dd-trace-py #21135. Either PR can merge first; ship the tracer capability before the Lambda library/layer containing both changes. Older tracers safely skip the flush. Delivery remains best effort; timeout stops waiting, not the underlying send.
PR checklist
Types of Changes
Check all that apply