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).
| 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
| 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.
| 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.
| @@ -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.
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.