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
| 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. |
There was a problem hiding this comment.
I think we can remove this from the README. We don't have any other content like this for other products. We usually just expect people to use our public docs site instead.
| and layer ARN/version together, plus runtime, architecture, configuration source, | ||
| service/environment, request IDs, and ingestion evidence. All are pending deployed | ||
| validation and release coordination. Release the tracer first, then the wrapper | ||
| and layer that contain it. Older tracer versions safely skip the capability. |
There was a problem hiding this comment.
This seems more like a PR description than something we'd want to commit to the repo. I think we should remove it.
| 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.
|
|
||
| flush_in_thread = _get_env("DD_FLUSH_IN_THREAD", "false", as_bool) | ||
| feature_flag_flush_timeout_ms = _get_env( | ||
| "DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS", 500.0, float |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] Rename the variable to DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS, to match the tracer's plural feature flag prefix.
The tracer's DD_FEATURE_FLAGS_ENABLED, DD_FEATURE_FLAGS_CONFIGURATION_SOURCE and DD_FEATURE_FLAGS_CONFIGURATION_SOURCE_AGENTLESS_* settings use the plural prefix (ddtrace/internal/settings/openfeature.py). Other settings there use DD_FFE_, DD_FLAGGING_ or DD_EXPERIMENTAL_FLAGGING_, so a singular DD_FEATURE_FLAG_ would add another spelling. We want one prefix for feature flag variables across repositories. A name is hard to change after it ships, because a rename usually needs an alias and a deprecation period.
Suggested change — config, README, runbook and tests
--- a/README.md
+++ b/README.md
@@ -85,7 +85,7 @@ 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
+`DD_FEATURE_FLAGS_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
--- a/datadog_lambda/config.py
+++ b/datadog_lambda/config.py
@@ -86,7 +86,7 @@ class Config:
flush_in_thread = _get_env("DD_FLUSH_IN_THREAD", "false", as_bool)
feature_flag_flush_timeout_ms = _get_env(
- "DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS", 500.0, float
+ "DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS", 500.0, float
)
flush_to_log = _get_env("DD_FLUSH_TO_LOG", "false", as_bool)
logs_injection = _get_env("DD_LOGS_INJECTION", "true", as_bool)
--- a/tests/integration/FEATURE_FLAG_FLUSH.md
+++ b/tests/integration/FEATURE_FLAG_FLUSH.md
@@ -30,7 +30,7 @@ Verify that the selected profile can access the approved account before deployin
from `serverless.yml`, with a unique `integration-tests-*` function name.
3. Configure `FFL_SMOKE_FLAG_KEY` with an existing string test flag that produces
exposure events for the test subjects. Set `DD_SERVICE` to a unique smoke-test
- service, `DD_ENV` to the test environment and `DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS=500`.
+ service, `DD_ENV` to the test environment and `DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS=500`.
Keep evaluation telemetry and exposure intake enabled.
4. For direct delivery, use `DD_FEATURE_FLAGS_CONFIGURATION_SOURCE=agentless`,
the test organization's `DD_API_KEY`/`DD_SITE`, and no local Agent endpoint.
--- a/tests/test_feature_flags.py
+++ b/tests/test_feature_flags.py
@@ -72,9 +72,9 @@ class FeatureFlagFlushTest(unittest.TestCase):
feature_flags, "config", Config()
), patch.object(feature_flags, "_telemetry") as telemetry:
if value is None:
- os.environ.pop("DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS", None)
+ os.environ.pop("DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS", None)
else:
- os.environ["DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS"] = value
+ os.environ["DD_FEATURE_FLAGS_FLUSH_TIMEOUT_MS"] = value
telemetry.FLUSH_API_VERSION = 1
telemetry.flush.return_value = "completed"
feature_flags.flush_feature_flag_telemetry(The full suite passes with the rename.
|
|
||
|
|
||
| class FeatureFlagTracerIntegrationTest(unittest.TestCase): | ||
| @unittest.skipUnless( |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] Add a contract test that runs in CI. CI skips this test today. The other tests mock either the decorator's internals or flush_feature_flag_telemetry, so no CI job checks that the real decorator reaches _telemetry.flush().
- CI installs
.[dev], which does not includeopenfeature-sdk._telemetryis new in fix(openfeature): add bounded invocation-end EVP flush dd-trace-py#21135. - The real-wrapper tests reach
flush_feature_flag_telemetrybut never call_telemetry.flush().tests/utils.get_mock_context()returns 100 ms of remaining time, so the budget is 0.
The suggested test replaces _telemetry with a small fake module and reloads feature_flags. It then calls the real decorator with a handler that returns and a handler that raises. It does not need a ddtrace release with the contract. When a release has it, also add openfeature-sdk to the dev extra, so this integration test runs on the Python versions whose ddtrace pin allows that release.
Suggested test — in TestDatadogLambdaWrapper, which supplies the mocks
--- a/tests/test_wrapper.py
+++ b/tests/test_wrapper.py
@@ -77,6 +77,36 @@ class TestDatadogLambdaWrapper(unittest.TestCase):
self.mock_dd_lambda_layer_tag = patcher.start()
self.addCleanup(patcher.stop)
+ def test_real_decorator_calls_flush_contract_v1(self):
+ import types
+ import ddtrace.internal.openfeature as openfeature_pkg
+ from datadog_lambda import feature_flags
+
+ calls = []
+ fake = types.ModuleType("ddtrace.internal.openfeature._telemetry")
+ fake.FLUSH_API_VERSION = 1
+ fake.flush = lambda timeout: calls.append(timeout) or "completed"
+ self.addCleanup(importlib.reload, feature_flags)
+ with patch.dict(sys.modules, {fake.__name__: fake}), patch.object(
+ openfeature_pkg, "_telemetry", fake, create=True
+ ):
+ importlib.reload(feature_flags) # runs the real import line
+ context = get_mock_context()
+ context.get_remaining_time_in_millis = lambda: 30000
+
+ ok = wrapper.datadog_lambda_wrapper(lambda event, context: {"ok": True})
+ self.assertEqual(ok({}, context), {"ok": True})
+
+ error = ValueError("handler error")
+
+ def raising(event, context):
+ raise error
+
+ with self.assertRaises(ValueError) as caught:
+ wrapper.datadog_lambda_wrapper(raising)({}, context)
+ self.assertIs(caught.exception, error)
+ self.assertEqual(calls, [0.5, 0.5])
+
@patch("datadog_lambda.config.Config.trace_enabled", False)
def test_datadog_lambda_wrapper(self):
@wrapper.datadog_lambda_wrapperThe test passes. It fails when I remove the wrapper's flush call, and when the import path is wrong.
|
|
||
| [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.
| cap_ms = config.feature_flag_flush_timeout_ms | ||
| if not math.isfinite(cap_ms) or cap_ms <= 0: | ||
| return | ||
| # No remaining-time information means no safe invocation budget. |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P3] Return early when context has no get_remaining_time_in_millis, and log the unsupported-tracer skip.
This comment says: "No remaining-time information means no safe invocation budget." The next line calls context.get_remaining_time_in_millis() without a check. A missing method raises AttributeError, and the outer except logs it at debug level as "Feature flag telemetry flush failed", with a traceback. The skip is correct, but the log reads like a failure.
Also, line 20 returns without a log when FLUSH_API_VERSION is not 1. If the tracer moves to contract version 2, the flush stops with no log.
Suggested change and tests
--- a/datadog_lambda/feature_flags.py
+++ b/datadog_lambda/feature_flags.py
@@ -17,14 +17,19 @@ SAFETY_MARGIN_MS = 100.0
def flush_feature_flag_telemetry(context):
try:
- if _telemetry is None or getattr(_telemetry, "FLUSH_API_VERSION", None) != 1:
+ if _telemetry is None:
+ return
+ if getattr(_telemetry, "FLUSH_API_VERSION", None) != 1:
+ logger.debug("Skipping feature flag telemetry flush: unsupported tracer")
return
cap_ms = config.feature_flag_flush_timeout_ms
if not math.isfinite(cap_ms) or cap_ms <= 0:
return
# No remaining-time information means no safe invocation budget.
- remaining_ms = context.get_remaining_time_in_millis()
- budget_ms = min(cap_ms, remaining_ms - SAFETY_MARGIN_MS)
+ get_remaining_ms = getattr(context, "get_remaining_time_in_millis", None)
+ if get_remaining_ms is None:
+ return
+ budget_ms = min(cap_ms, get_remaining_ms() - SAFETY_MARGIN_MS)
if budget_ms <= 0:
logger.debug(
"Skipping feature flag telemetry flush: no invocation time remains"
--- a/tests/test_feature_flags.py
+++ b/tests/test_feature_flags.py
@@ -92,6 +92,26 @@ class FeatureFlagFlushTest(unittest.TestCase):
)
telemetry.flush.assert_called_once()
+ def test_unsupported_tracer_is_logged(self):
+ with patch.object(
+ feature_flags, "_telemetry", Mock(FLUSH_API_VERSION=2)
+ ), patch.object(feature_flags, "logger") as log:
+ feature_flags.flush_feature_flag_telemetry(
+ Mock(get_remaining_time_in_millis=lambda: 1000)
+ )
+ log.debug.assert_called_once_with(
+ "Skipping feature flag telemetry flush: unsupported tracer"
+ )
+
+ def test_missing_context_skips_without_failure_log(self):
+ with patch.object(feature_flags, "_telemetry") as telemetry, patch.object(
+ feature_flags, "logger"
+ ) as log:
+ telemetry.FLUSH_API_VERSION = 1
+ feature_flags.flush_feature_flag_telemetry(None)
+ telemetry.flush.assert_not_called()
+ log.debug.assert_not_called()
+
def test_incomplete_diagnostic(self):
with patch.object(feature_flags, "_telemetry") as telemetry, patch.object(
feature_flags, "logger"The full suite passes. Each new test fails without its change.
| from importlib import import_module | ||
| from time import time_ns | ||
|
|
||
| from datadog_lambda.feature_flags import flush_feature_flag_telemetry |
There was a problem hiding this comment.
We should put this import alongside where we do all the other feature imports from ddtrace (ex llmobs below).
You'll want to put the import behind a config check, so that way we're not doing the import unless the user has configured to enable the feature.
Then, where you call the actual flush_feature_flag_telemetry, you'll do another check of the config value.
Then, add DD_FEATURE_FLAGS_ENABLED to the config, alongside the DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS you already added. This will allow users to turn this off and potentially avoid the import.
| 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.
| 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.
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
DD_FEATURE_FLAG_FLUSH_TIMEOUT_MS(default 500 ms) and remaining invocation time minus a 100 ms safety margin.Testing
main: 361 tests plus 15 subtests passed with the rebuilt candidate tracer; 360 plus 15 subtests passed withddtrace4.1.1 (candidate-only test skipped). 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