Repository navigation
Conversation
Track pending, dequeued, and in-flight feature flag telemetry and expose a versioned internal boundary for Lambda cleanup. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Codeowners resolved asResolved from the full PR diff against |
Dependency direction analysis
|
BenchmarksBenchmark execution time: 2026-10-23 02:48:14 Comparing candidate commit 2391a8e in PR branch Found 0 performance improvements and 3 performance regressions! Performance is the same for 17 metrics, 0 unstable metrics.
|
Account for every consumed event, report aggregation failures on their batch, and allow subsequent healthy flushes to complete. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
🎉 All green!🧪 All tests passed 🔗 Commit SHA: 2391a8e | Docs | View more details | Give us feedback! |
Incorporate upstream tracer updates and validate OpenFeature flushing with the rebuilt native periodic threads. Environment: Datadog workspace Co-Authored-By: Codex GPT-6 <noreply@localhost>
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2391a8e342
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| failed |= progress.last_failure >= started | ||
| if progress.accepted > progress.settled: | ||
| pending.append((writer, progress.accepted, progress.failures)) |
There was a problem hiding this comment.
When a periodic writer sends and rejects an invocation's batch before the invocation-end call starts, accepted == settled and last_failure < started, so this path returns completed even though the snapshotted work was not accepted by an endpoint. This is especially reachable for invocations longer than the writers' 1- or 10-second intervals. Track unreported failures across flush calls rather than considering only failures timestamped after this call began.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] I confirmed this with a test. Count failures per writer instead of comparing failure times with the call's start time.
When the endpoint rejects a background send before flush() starts, accepted == settled and last_failure < started, so flush() returns completed. This reproduces for both writers. With a count of failures already reported, the next call no longer returns completed after that failure, and later calls do not count it again.
Suggested fix and test — the test fails at this head and passes with the fix
--- a/ddtrace/internal/openfeature/_telemetry.py
+++ b/ddtrace/internal/openfeature/_telemetry.py
@@ -25,7 +25,7 @@ class Progress:
self.accepted = 0
self.settled = 0
self.failures = 0
- self.last_failure = float("-inf")
+ self.reported_failures = 0
def accept(self) -> None:
with self.lock:
@@ -36,7 +36,6 @@ class Progress:
self.settled = target
if not success:
self.failures += 1
- self.last_failure = time.monotonic()
class Writer(Protocol):
@@ -107,7 +106,9 @@ def flush(timeout: float) -> str:
if not progress.lock.acquire(timeout=max(0.0, deadline - time.monotonic())):
return "timed_out"
try:
- failed |= progress.last_failure >= started
+ # Count a failure in the next call, even if it settled before that call started.
+ failed |= progress.failures != progress.reported_failures
+ progress.reported_failures = progress.failures
if progress.accepted > progress.settled:
pending.append((writer, progress.accepted, progress.failures))
finally:
@@ -149,6 +150,7 @@ def flush(timeout: float) -> str:
return "timed_out"
try:
failed |= progress.failures != failures
+ progress.reported_failures = progress.failures
if progress.settled < target:
waiting.append((writer, target, failures))
finally:
--- a/tests/openfeature/test_telemetry_flush.py
+++ b/tests/openfeature/test_telemetry_flush.py
@@ -294,3 +294,13 @@ def test_ambiguous_send_failure_is_not_retried(kind):
assert _telemetry.flush(1) == "failed"
assert _telemetry.flush(1) == "completed"
assert connection.request.call_count == 1
+
+
+@pytest.mark.parametrize("kind", ["evaluation", "exposure"])
+def test_rejection_before_flush_is_reported_once(kind):
+ writer, connection = make_writer(kind, status=500)
+ enqueue(writer)
+ writer.periodic() # The background send is rejected before the invocation-end flush starts.
+ assert _telemetry.flush(1) == "failed"
+ assert _telemetry.flush(1) == "completed"
+ assert connection.request.call_count == 1test_failure_does_not_poison_next_invocation and test_ambiguous_send_failure_is_not_retried still pass, along with the rest of the writer suites and tests/internal/test_serverless.py (266 tests).
There was a problem hiding this comment.
🤖 Comment from Codex working with @leoromanovsky:
[P1] I independently reproduced this for both writers on 2391a8e. Aaron's failure-watermark change addresses the tested sequence: report an earlier rejection once, then allow the next flush to complete without another send.
Please preserve that recovery behavior when replacing the timestamp comparison. This corrects the delivery result; it does not retry or recover the rejected batch.
Rejection and successful-delivery controls
The tests run writer.periodic() before calling flush(). With HTTP 500, the reviewed code incorrectly returns completed for both evaluation and exposure writers. HTTP 202 controls return completed as expected.
With the watermark change in the combined candidate, each rejection returns failed once, the next call returns completed, and the request count stays at one. Removing the initial unreported-failure comparison makes both rejection cases fail again while both HTTP 202 controls pass.
This validates sequential invocation boundaries. It does not establish semantics for simultaneous independent flush() callers.
| coordinator.worker.start() | ||
| except Exception: | ||
| coordinator.worker = None | ||
| raise |
There was a problem hiding this comment.
On the first pending flush, PeriodicThread.start() can raise an OSError when the runtime cannot allocate a thread, and the explicit re-raise lets that exception escape the invocation-end boundary instead of returning one of its documented results. In a resource-constrained Lambda this can turn best-effort telemetry cleanup into an application-visible failure; return failed (or timed_out when the budget is exhausted) after resetting the worker.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] I confirmed this with a test. Return "failed" when the coordinator thread cannot start, and clear the pending request.
flush() documents three return values, but an OSError from PeriodicThread.start() escapes it. The Lambda wrapper catches the exception and logs a traceback, so the invocation is safe but gets no result. The fix still logs the error at debug level, then returns "failed". Clearing requested drops the writer references that the failed start leaves behind, and the test checks that too.
Suggested fix and test — the test fails at this head and passes with the fix
--- a/ddtrace/internal/openfeature/_telemetry.py
+++ b/ddtrace/internal/openfeature/_telemetry.py
@@ -133,7 +133,9 @@ def flush(timeout: float) -> str:
coordinator.worker.start()
except Exception:
coordinator.worker = None
- raise
+ coordinator.requested.clear()
+ log.debug("Feature flag telemetry flush worker failed to start", exc_info=True)
+ return "failed"
coordinator.signal.set()
finally:
coordinator.lock.release()
--- a/tests/openfeature/test_telemetry_flush.py
+++ b/tests/openfeature/test_telemetry_flush.py
@@ -294,3 +294,14 @@ def test_ambiguous_send_failure_is_not_retried(kind):
assert _telemetry.flush(1) == "failed"
assert _telemetry.flush(1) == "completed"
assert connection.request.call_count == 1
+
+
+def test_worker_start_failure_returns_failed_and_recovers():
+ writer, connection = make_writer()
+ enqueue(writer)
+ with mock.patch.object(_telemetry.PeriodicThread, "start", side_effect=OSError("can't start new thread")):
+ assert _telemetry.flush(1) == "failed"
+ assert _telemetry._coordinator.worker is None
+ assert not _telemetry._coordinator.requested
+ assert _telemetry.flush(1) == "completed"
+ assert connection.request.call_count == 1The writer suites and tests/internal/test_serverless.py pass with this change (265 tests).
![datadog-datadog-prod-us1[bot]](https://avatars.githubusercontent.com/u/365230?s=60&v=4){kind=small}
There was a problem hiding this comment.
The first flush can exceed its advertised timeout because coordinator startup waits without a deadline on thread-start synchronization. Contention during concurrent startup or post-fork restarts can therefore overrun the invocation’s remaining execution budget.
🤖 Bits Code Review · Commit 2391a8e
| # inherited worker against the parent's requested work. | ||
| coordinator.worker.__autorestart__ = False | ||
| try: | ||
| coordinator.worker.start() |
There was a problem hiding this comment.
On the first flush with pending work, concurrent periodic-thread startup or post-fork restarts can block this call beyond the supplied timeout. PeriodicThread.start() acquires threads._forking_lock without a timeout and then waits indefinitely for native startup acknowledgement. The flush deadline is checked only after start() returns, so invocation-end flushing can exceed its remaining Lambda execution budget. Startup must respect the remaining deadline or occur outside the bounded caller path.
React: 👍 useful · 👎 wrong · 😕 unclear · 👀 right, but not acting on it
What each reaction means
- 👍 Useful: correct, and you're fixing it or will.
- 👎 Wrong: the finding is incorrect, for example the bug can't happen or the code was misread.
- 😕 Unclear: you can't tell what it means or why it matters.
- 👀 Right, not acting: correct, but intended, out of scope for this PR, a known risk, or too minor.
🤖 Bits Code Review · Open Bits AI session
There was a problem hiding this comment.
🤖 Comment from Codex working with @leoromanovsky:
[P1] Confirmed on 2391a8e: coordinator startup can exceed the caller's total budget. Please move fallback startup outside flush() or bound its lock acquisition and native startup acknowledgement.
Aaron's suggestion to wake existing writer threads avoids coordinator startup for running writers. The fallback for unstarted writers still needs a separate fix. Catching startup exceptions does not bound a successful but delayed start.
Startup-contention reproduction
On Python 3.13, another thread held the real threads._forking_lock for 300 ms. flush(0.02) returned timed_out after 300 ms, despite its 20 ms budget. The regression assertion allowed 150 ms and still failed.
The tested parallel-worker candidate retains synchronous fallback startup, so it does not fully address this finding. I would retain this contention test alongside the running-writer concurrency test. The fallback's initialization and fork lifecycle still need an implementation decision.
| self.failures = 0 | ||
| self.last_failure = float("-inf") | ||
|
|
||
| def accept(self) -> None: |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P1] Take Progress.accept() off the ResetObject proxy and the extra lock. It runs for every evaluation the writer accepts, in every service, not only in Lambda.
accept() is the only new statement on the evaluation enqueue path. It resolves through a ResetObject proxy and then enters a second proxied lock. Both callers already hold an intake lock (_lifecycle_lock in FlagEvaluationWriter.enqueue, _lock in ExposureWriter.enqueue), which serializes the increments. flush() only reads the count, and an event accepted after its snapshot is outside that flush's target, so a stale read is harmless and the second lock adds nothing. The benchmark bot on this PR reports +10–20% on three hook-enqueue scenarios, and every accepted evaluation pays that cost.
Evidence — the regression reproduces locally, and the suggested fix measures within 1% of main
I measured in one process, interleaved: 15 rounds of 50,000 FlagEvalEVPHook.finally_after calls with the hook_enqueue benchmark setup, on Python 3.12. The table shows the median ns per call. "main" is this head with the accept() call removed.
| Variant | protected | full evaluation data |
|---|---|---|
| main | 3658 | 6231 |
| this PR | 4419 (+20.8%) | 6825 (+9.5%) |
| suggested fix | 3692 (+0.9%) | 6273 (+0.7%) |
In a separate run, removing only the lock recovered about 40% of the delta.
Suggested fix — a plain Progress, reset after fork by a forksafe hook
--- a/ddtrace/internal/openfeature/_flagevaluation_writer.py
+++ b/ddtrace/internal/openfeature/_flagevaluation_writer.py
@@ -812,7 +812,7 @@ class FlagEvaluationWriter(PeriodicService):
self._counter_lock = forksafe.Lock()
self._drain_lock = forksafe.RLock()
self._flush_lock = forksafe.Lock()
- self._progress = typing.cast(Progress, forksafe.ResetObject(Progress))
+ self._progress = Progress()
register(self)
self._drain_worker: typing.Optional[PeriodicThread] = None
--- a/ddtrace/internal/openfeature/_telemetry.py
+++ b/ddtrace/internal/openfeature/_telemetry.py
@@ -15,6 +15,7 @@ from ddtrace.internal.threads import PeriodicThread
FLUSH_API_VERSION = 1
log = get_logger(__name__)
+_progress: "WeakSet[Progress]" = WeakSet()
class Progress:
@@ -22,14 +23,18 @@ class Progress:
def __init__(self) -> None:
self.lock = forksafe.Lock()
+ self.reset()
+ _progress.add(self)
+
+ def reset(self) -> None:
self.accepted = 0
self.settled = 0
self.failures = 0
self.last_failure = float("-inf")
def accept(self) -> None:
- with self.lock:
- self.accepted += 1
+ # Hot path: no proxy, no lock. Owners call this under their intake lock.
+ self.accepted += 1
def finish(self, target: int, success: bool) -> None:
with self.lock:
@@ -39,6 +44,12 @@ class Progress:
self.last_failure = time.monotonic()
+@forksafe.register
+def _reset_progress() -> None:
+ for progress in list(_progress):
+ progress.reset()
+
+
class Writer(Protocol):
_progress: Progress
--- a/ddtrace/internal/openfeature/writer.py
+++ b/ddtrace/internal/openfeature/writer.py
@@ -128,7 +128,7 @@ class ExposureWriter(PeriodicService):
super().__init__(interval=interval)
self._lock = RLock()
self._flush_lock = forksafe.Lock()
- self._progress = cast(Progress, forksafe.ResetObject(Progress))
+ self._progress = Progress()
register(self)
self._process_state = cast(_ExposureProcessState, forksafe.ResetObject(_ExposureProcessState))
self._timeout: float = timeoutWith this change, tests/openfeature/test_telemetry_flush.py, test_exposure_writer.py, test_flagevaluation_writer.py and tests/internal/test_serverless.py pass (264 tests). When I disable _reset_progress, test_fork_child_does_not_flush_parent_work fails for both writers, so the existing fork test covers the hook.
There was a problem hiding this comment.
🤖 Comment from Codex working with @leoromanovsky:
[P1] When removing the progress proxy, please preserve child-process progress reset and rerun the enqueue benchmark. I checked a candidate that uses the existing intake locks for acceptance and retains a lock for settlement. Its semantic tests pass; I have not measured its speedup.
Fork controls and measurement scope
The candidate resets each live progress object through a fork hook. Both writer types pass tests that prevent parent-event replay and then send fresh child events across two flushes. Disabling that reset makes all four fork cases fail.
There is a separate useful control: clearing the writer registry in the child leaves the two original no-replay tests passing, but fails both fresh-child-delivery tests. Preserving registration as well as resetting progress matters.
The existing CI receipt measures roughly 2.5–2.6 µs added to the protected typical hook/enqueue path, or 19.55–20.28%. That is component overhead, not a measurement of total flag-evaluation or request latency. A new benchmark is still needed for the chosen fix.
| allows the native thread fork protocol to stop it; transport timeouts continue | ||
| to govern outstanding network operations during a fork. | ||
|
|
||
| ## Release coordination |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P3] Keep the compatibility rule here, and move the rest of "Release coordination" to the PR description.
Most of this section is release process: recording versions and layer ARNs, validation steps, and a release record. One sentence describes the branch, not the code: "No published combination is claimed by this development branch." Two sentences describe behavior and belong in this file: the Lambda side checks contract version 1, and older tracers are a no-op.
Suggested change — keep the two compatibility sentences, delete the rest
--- a/ddtrace/internal/openfeature/README.md
+++ b/ddtrace/internal/openfeature/README.md
@@ -35,18 +35,7 @@ coordinator does not restart inherited work in the child. Its bounded idle wait
allows the native thread fork protocol to stop it; transport timeouts continue
to govern outstanding network operations during a fork.
-## Release coordination
+## Compatibility
-The first published tracer version, Lambda library version, and compatible layer
-ARN/version must be recorded together after release artifacts are available.
-No published combination is claimed by this development branch. The Lambda side
-checks contract version 1 and safely skips older tracers. Release the tracer
-capability first, then the Lambda wrapper and a layer containing both changes.
-Python versions whose tracer release lacks contract version 1 remain a no-op.
-
-Before declaring support, validate a short successful invocation and a short
-raising invocation on the published layer, repeat them in a warm sandbox, and
-confirm ingestion of both evaluation and exposure events for a unique test
-subject. Cover local Agent EVP delivery separately from agentless direct fallback.
-Record the runtime, architecture, tracer version, Lambda library version, layer
-ARN, invocation IDs and ingestion evidence in the release record.
+The Lambda side checks contract version 1 and safely skips older tracers. Python
+versions whose tracer release lacks contract version 1 remain a no-op.| @@ -0,0 +1,6 @@ | |||
| fixes: | |||
| - | | |||
| OpenFeature: Enables compatible Datadog Lambda libraries to flush pending | |||
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P3] Match the release-note conventions, and add the fork fix to the note.
- All 17 other OpenFeature notes with a scope in
releasenotes/notes/use lowercaseopenfeature:. This one usesOpenFeature:. - 16 of those 17 start with
---. docs/releasenotes.rstasks a fix to state the user-visible symptom: "Fixes an issue where occurs when ."- This PR also stops a forked child from sending exposure events that its parent had buffered (
_ExposureProcessState). Before, those events could be sent twice. The description mentions resetting coordination state after a fork, but neither it nor the note mentions the duplicate exposures.
Suggested note — replaces the whole file
---
fixes:
- |
openfeature: Fixes an issue where feature flag evaluation and exposure events could be delayed
or lost when an AWS Lambda invocation ended shortly after evaluating flags. Requires a Datadog
Lambda library version that flushes these events at invocation end.
- |
openfeature: Fixes an issue where exposure events could be sent twice after a process forked.It parses as YAML with two fixes entries. I could not run reno lint in my environment.
| with self._lifecycle_lock: | ||
| self._accepting_events = False | ||
| self._request_drain_worker_stop() | ||
| # A stopped drain worker owns its dequeued event and final callback. |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P3] Clarify this comment. The lock check does not make on_shutdown fully bounded.
The comment says: "A stopped drain worker owns its dequeued event and final callback. Do not turn the application's bounded join into an unbounded wait."
- The non-blocking
_drain_lockcheck is correct. A drain worker that holds the lock runs the final drain throughon_shutdown=self.periodic. I traced the shutdown path and found no lost events. - The next line,
self.periodic(), still waits on_flush_lock. A coordinator or periodic flush holds that lock across network I/O, soon_shutdowncan still wait for a send in flight.
Suggested wording
--- a/ddtrace/internal/openfeature/_flagevaluation_writer.py
+++ b/ddtrace/internal/openfeature/_flagevaluation_writer.py
@@ -1174,8 +1174,8 @@ class FlagEvaluationWriter(PeriodicService):
with self._lifecycle_lock:
self._accepting_events = False
self._request_drain_worker_stop()
- # A stopped drain worker owns its dequeued event and final callback.
- # Do not turn the application's bounded join into an unbounded wait.
+ # A drain worker that holds the lock owns the final drain through its
+ # on_shutdown callback. periodic() can still wait for an in-flight send.
if not self._drain_lock.acquire(blocking=False):
return
self._drain_lock.release()| self.signal.clear() | ||
| for writer in writers: | ||
| try: | ||
| writer.periodic() |
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P2] Wake running writers with PeriodicThread.awake() so they flush in parallel on their own threads. Keep the coordinator only for writers whose worker is not running.
When the coordinator drives the flush for both writers, it runs their periodic() calls one after the other, so the flush waits for both sends in sequence. With a local endpoint that takes 300 ms per request, flush(0.5) returned timed_out on every call. The parallel version returned completed in about 305 ms.
Other Datadog tracers use the parallel pattern. In dd-trace-dotnet, FlagEvaluationWriter.FlushAsync wakes the writer's own consumer and waits with a cap. In dd-trace-java, the feature-flagging shutdown joins every writer thread against one shared deadline. Since #20716, awake() does not block, so the caller still waits outside the writers' flush and intake locks.
Evidence — measured wall time of flush() with both writers pending
Both writers were started. A local HTTP server slept before each 202. Each row is the median of 8 flushes.
| Delay per request | Budget | Current coordinator | Parallel |
|---|---|---|---|
| 50 ms | 2 s | 107 ms | 56 ms |
| 150 ms | 2 s | 308 ms | 154 ms |
| 300 ms | 2 s | 610 ms | 306 ms |
| 300 ms | 500 ms | 500 ms, timed_out |
305 ms, completed |
The provider starts FlagEvaluationWriter (_provider.py:284), and start_exposure_writer() starts the exposure writer (writer.py:349). Once both run, the coordinator thread starts only when a writer is not running. Until it starts, its 10 wakeups per second do not happen.
The check uses status, not only _worker. After stop(), awake() does nothing and _worker can remain set, so a stopped writer must still go to the coordinator.
Suggested fix and test — the test fails against the current coordinator
--- a/ddtrace/internal/openfeature/README.md
+++ b/ddtrace/internal/openfeature/README.md
@@ -15,12 +15,13 @@ drops after acceptance, and rejected or ambiguous deliveries are incomplete.
Failures observed while coordinating the snapshot are also included. A failed
invocation does not permanently poison subsequent flushes.
-One lazily created coordinator serves all invocation-end requests. Calls share a
-single monotonic deadline across registry/progress coordination and waiting for
-aggregation, route discovery, and delivery. The coordinator performs the writers'
-existing transport operations; the caller does not perform network I/O. Requests
-coalesce by writer, so repeated timeouts cannot accumulate helper workers or
-unbounded request queues. Providers, hooks, polling and ordinary writer workers
+Running writers flush on their own worker threads, in parallel, after a
+non-blocking wake. One lazily created coordinator serves writers whose worker is
+not running. Calls share a single monotonic deadline across registry/progress
+coordination and waiting for aggregation, route discovery, and delivery. The
+writer workers and the coordinator perform the existing transport operations;
+the caller does not perform network I/O. Requests coalesce by writer, so repeated
+timeouts cannot accumulate helper workers or unbounded request queues. Providers, hooks, polling and ordinary writer workers
remain active across warm invocations.
Timeout stops waiting, not delivery. Queued work and in-flight batches remain with
--- a/ddtrace/internal/openfeature/_telemetry.py
+++ b/ddtrace/internal/openfeature/_telemetry.py
@@ -10,6 +10,7 @@ from weakref import WeakSet
from ddtrace.internal import forksafe
from ddtrace.internal.logger import get_logger
+from ddtrace.internal.service import ServiceStatus
from ddtrace.internal.threads import PeriodicThread
@@ -117,26 +118,36 @@ def flush(timeout: float) -> str:
if not pending:
return "failed" if failed else "completed"
- coordinator = _coordinator
- if not coordinator.lock.acquire(timeout=max(0.0, deadline - time.monotonic())):
- return "timed_out"
- try:
- coordinator.requested.update(writer for writer, _, _ in pending)
- if coordinator.worker is None:
- coordinator.worker = PeriodicThread(
- 0.0, coordinator.run, name="ddtrace:feature-flag-flush", no_wait_at_start=True
- )
- # ResetObject replaces the coordinator in a child; don't restart its
- # inherited worker against the parent's requested work.
- coordinator.worker.__autorestart__ = False
- try:
- coordinator.worker.start()
- except Exception:
- coordinator.worker = None
- raise
- coordinator.signal.set()
- finally:
- coordinator.lock.release()
+ # Running writers flush on their own threads, in parallel. The coordinator
+ # serves writers whose worker is not running.
+ idle = []
+ for writer, _, _ in pending:
+ worker = getattr(writer, "_worker", None)
+ if worker is not None and getattr(writer, "status", None) is ServiceStatus.RUNNING:
+ worker.awake()
+ else:
+ idle.append(writer)
+ if idle:
+ coordinator = _coordinator
+ if not coordinator.lock.acquire(timeout=max(0.0, deadline - time.monotonic())):
+ return "timed_out"
+ try:
+ coordinator.requested.update(idle)
+ if coordinator.worker is None:
+ coordinator.worker = PeriodicThread(
+ 0.0, coordinator.run, name="ddtrace:feature-flag-flush", no_wait_at_start=True
+ )
+ # ResetObject replaces the coordinator in a child; don't restart its
+ # inherited worker against the parent's requested work.
+ coordinator.worker.__autorestart__ = False
+ try:
+ coordinator.worker.start()
+ except Exception:
+ coordinator.worker = None
+ raise
+ coordinator.signal.set()
+ finally:
+ coordinator.lock.release()
while pending:
remaining = deadline - time.monotonic()
--- a/tests/openfeature/test_telemetry_flush.py
+++ b/tests/openfeature/test_telemetry_flush.py
@@ -294,3 +294,21 @@ def test_ambiguous_send_failure_is_not_retried(kind):
assert _telemetry.flush(1) == "failed"
assert _telemetry.flush(1) == "completed"
assert connection.request.call_count == 1
+
+
+def test_running_writers_flush_in_parallel():
+ writers = [make_writer(kind) for kind in ("evaluation", "exposure")]
+ for writer, connection in writers:
+ connection.getresponse.side_effect = lambda: (time.sleep(0.2), mock.Mock(status=202))[1]
+ writer.start()
+ enqueue(writer)
+ try:
+ started = time.monotonic()
+ # One send at a time needs at least 0.4 s.
+ assert _telemetry.flush(0.35) == "completed"
+ assert time.monotonic() - started < 0.35
+ assert _telemetry._coordinator.worker is None
+ finally:
+ for writer, _ in writers:
+ writer.stop()
+ writer.join(2)The writer suites and tests/internal/test_serverless.py pass (265 tests). The tests in test_telemetry_flush.py build writers without starting them, so they still cover the coordinator path. Against the current coordinator, the new test fails with timed_out.
There was a problem hiding this comment.
🤖 Comment from Codex working with @leoromanovsky:
[P2] The parallel-worker approach also passes an independent concurrency test. Please retain the ServiceStatus.RUNNING guard: a retained worker reference alone does not establish that awake() will flush it.
Concurrency control without ordinary timer wakeups
I set both writer intervals to 3,600 seconds and blocked each mocked response until both requests had entered. With a 350 ms flush budget, the reviewed coordinator returns timed_out. The candidate that wakes running workers completes and sends each request once. Restoring serial coordination makes the test fail again.
This checks overlap directly, rather than relying only on elapsed-time improvement. It supports using each running writer's existing worker to make better use of the shared budget. It does not prove permanent event loss from the serial path.
The fallback for writers that are not running remains subject to the separate startup-deadline finding.
| @@ -0,0 +1,159 @@ | |||
| """Versioned internal boundary for Lambda's best-effort EVP flush. | |||
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
[P3] Consider renaming _telemetry now, before the module path becomes a cross-repo contract.
ddtrace.internal.openfeature._telemetry sits next to ddtrace.internal.telemetry, the instrumentation telemetry package, and _flagevaluation_writer.py imports both. After release, datadog-lambda-python imports this path and tests/internal/test_serverless.py pins it, so a later rename needs a compatibility alias or coordinated releases.
Suggested change: rename the module to _flush.py (or similar) and update its references: the imports in _flagevaluation_writer.py and writer.py; the allowlist entry and the new import test in tests/internal/test_serverless.py; tests/openfeature/test_telemetry_flush.py; this package's README.md; and the import in DataDog/datadog-lambda-python#872. The rename is mechanical, so I did not apply it.
| assert _telemetry._coordinator.worker is None | ||
| """ | ||
| _, _, status, _ = run_python_code_in_subprocess(code) | ||
| assert status == 0 |
There was a problem hiding this comment.
This should already be covered in the test_serverless_imports above, yeah? The only thing different are these asserts, which are already tested in your other unit tests below.
| @@ -1004,26 +1014,40 @@ def _stop_service(self, *args: typing.Any, **kwargs: typing.Any) -> None: | |||
| super()._stop_service(*args, **kwargs) | |||
|
|
|||
| def periodic(self) -> None: | |||
There was a problem hiding this comment.
What if..... we just call periodic directly from the lambda function? We already do that same thing for exception replay, so there's precident for it. That would mean this PR is completely unnecessary.....
There was a problem hiding this comment.
If that's true, then we could just do a single PR in datadog-lambda-python DataDog/datadog-lambda-python@5d3d3e3
That uses a bunch of code from ddtrace.internal which we don't love, but as long as it's well protected, I don't see why that smaller PR wouldn't work well.
There was a problem hiding this comment.
🤖 Comment from Claude working with Aaron Silverman:
I replied on the Lambda PR with results from running 5d3d3e3 against main (no wait budget, and an in-flight send is not awaited): DataDog/datadog-lambda-python#872 (comment)
leoromanovsky
left a comment
leoromanovsky
left a comment
There was a problem hiding this comment.
Commented in-line with some reproduction and more evidence.
FFE reviewer selectionReviewers: @leoromanovsky, @sameerank Why: Agentless and serverless delivery — The PR adds a bounded, versioned flush operation that a serverless (Lambda) runtime calls at the end of each invocation. Reviewers come from the matched pod. 🤖 Automated selection. Selection detailsInitiative: project / deployment-modes Why this match: The PR adds a bounded, versioned flush operation that a serverless (Lambda) runtime calls at the end of each invocation. It drains queued and in-flight feature-flag evaluation and exposure events across all writers, so short or frozen serverless invocations don't lose them. This is evaluation and exposure event egress for serverless server SDKs. The changes are a new flush coordinator in _telemetry.py, progress tracking and flush locks in the exposure and evaluation writers, and fork-reset buffer state. These change how events are delivered, not EVP consent or metadata semantics, and they add no onboarding diagnostics.
The PR author was excluded. Existing reviewers and teams were preserved. Debug or fix a problemWorkflow run · Bits Code session
Evidence and run details✅ Complete — reviewer coverage verified. The workflow checked pending requests or submitted reviews on the evaluated head. Generated asynchronously by the FFE reviewer-routing workflow. Changed paths: Evaluated head: Registry revision: Pinned skill source: Instructions: workflow-injected snapshot (no plugin loaded) Instruction snapshot SHA-256: |
Motivation
A customer asked about flushing feature-flag telemetry from Python Lambda functions (FFL-3412). Short invocations can finish before background writers send evaluation and exposure events; Lambda freezing can delay delivery, and termination can lose buffered events.
This PR provides the tracer capability that lets the Lambda wrapper flush at invocation end, without customers adding handler code or shutting down providers needed by warm invocations.
Changes
Delivery remains best effort. Timeout stops waiting, not an outstanding send; the implementation adds writer synchronization and one reusable background coordinator.
Testing
main: 269 tracer tests and 361 Lambda integration/wrapper tests plus 15 subtests passed against the rebuilt tracer; full lint checks passed.Release coordination
Companion: datadog-lambda-python #872. Either PR can merge first; release the tracer capability before the Lambda library/layer containing both changes. Record the supported versions before announcing support.
FFE reviewer selection
Reviewers: @sameerank, @leoromanovsky
Why: Agentless and serverless delivery — Adds a versioned, bounded invocation-end flush (_telemetry.flush) that the AWS Lambda wrapper calls to deliver queued and in-flight feature-flag evaluation and exposure EVP… Global reviewers help cover a shortage in the matched pod.
🤖 Automated selection.
Selection details
Initiative: project / deployment-modes
Why this match: Adds a versioned, bounded invocation-end flush (_telemetry.flush) that the AWS Lambda wrapper calls to deliver queued and in-flight feature-flag evaluation and exposure EVP events before a serverless invocation freezes or terminates. It coordinates periodic and flush-triggered sends, resets state after a fork, and keeps providers reusable across warm invocations. This is evaluation/exposure event egress for a serverless runtime without a persistent Agent, covering Agent EVP and agentless direct fallback. It does not change consent or payload semantics.
Why these owners: The matched initiative has fewer than two approved, non-author candidates. The approved global pool supplies the remaining review owners; proposed memberships were not activated.
The PR author was excluded. Existing reviewers and teams were preserved.
Debug or fix a problem
Workflow run · Bits Code session
dry_run=false. Unchanged completed runs are skipped. If the receipt saysneeds_reconciliationor a write outcome is uncertain, ask the workflow owner to reconcile GitHub and the receipt before retrying; do not delete the lock blindly.Evidence and run details
✅ Complete — reviewer coverage verified. The workflow checked pending requests or submitted reviews on the evaluated head.
Generated asynchronously by the FFE reviewer-routing workflow.
Changed paths:
ddtrace/internal/openfeature/_telemetry.py,ddtrace/internal/openfeature/writer.py,ddtrace/internal/openfeature/_flagevaluation_writer.py,ddtrace/internal/openfeature/README.md,tests/openfeature/test_telemetry_flush.py,tests/internal/test_serverless.pyEvaluated head:
2391a8e342b49b0fa0065edc7591d68b1daa2ab7Registry revision:
revision:1946aa2a-530e-487a-84a7-83aff44a1089Pinned skill source:
d214cfc65ae154b97d39af5cf32caf48aaec9bf0Instructions: workflow-injected snapshot (no plugin loaded)
Instruction snapshot SHA-256:
f23b2a88c13c0da90e2dd58246a14f257af91e2195c555039bd555b52e6f3516