Add retry resilience to Agent365 exporter - #248
Add retry resilience to Agent365 exporter#248Nikhil Navakiran (nikhilNava) wants to merge 42 commits into
Conversation
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Use isolation_level=None + explicit BEGIN IMMEDIATE/COMMIT/ROLLBACK everywhere - Prune expired rows inside claim() transaction - Use os.mkdir(mode=0o700)+chmod for POSIX directory; enforce mode on existing dirs - Replace global Path.stat patch with module-local os.stat mock in ownership test - Add tests: autocommit mode assertion, claim-prunes-expired-rows Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
…op-pass, and new tests - Export ReplayIdentityError for identity/token-resolution failures; catching it releases the current record and gate probe then continues the batch. - Unexpected (general) exceptions now release the current record, release remaining leased records, and abort the pass. - Add test_replay_releases_record_when_identity_error_and_continues, test_general_exception_releases_current_and_remaining_and_stops, test_gate_blocked_releases_record_without_send, test_start_after_shutdown_is_safe_noop, and test_mid_batch_stop_releases_remaining_records. - Clarify in code comments why PERMANENT disposition calls record_success (identity is healthy; only the payload was rejected). - Document that start() after shutdown() is a safe no-op via docstring and debug log. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Replace the synchronous sleep/retry loop and global circuit breaker with a classified single-send (_post_once), a per-identity TransmissionGate, a durable SQLite queue, and background replay that rebuilds auth from each record's identity with a freshly resolved token. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Wire a365_exporter_disable_offline_storage and a365_exporter_storage_directory kwargs through _constants.py, use_microsoft_opentelemetry, _append_a365_components, Agent365ExporterOptions, and create_a365_components into _Agent365Exporter. Defaults: storage enabled, directory None. Document at-least-once semantics, 2-day/50-MB limits, and the security note that stored OTLP payloads may contain prompts when sensitive-data capture is enabled. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Address accumulated review findings without changing the fixed 50 MB / 2 day / 10-record behavior or adding dependencies: - Live export: release the per-identity gate probe and persist the payload if _post_once raises unexpectedly, so an identity is never permanently gated. - Replay: add a fixed periodic wake (30s background cadence) plus immediate continuation after a fully-drained maximal batch, so a startup backlog larger than one pass is not left at >10 records until an external wake. - Gate: add release_probe and record_success coverage tests. - Persistent storage: use non-symlink-following lstat for POSIX ownership, reject symlinked queue directories, and prefer creating ~/.local/state instead of falling back to a temp dir merely because it does not exist. - Options: reject an explicitly empty/whitespace storage_directory with ValueError, add a coerce_storage_directory helper, and strengthen end-to-end and no-leak coverage for the public option. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Address final-review findings in the A365 durable delivery persistent storage: - PersistentStorage.store capacity accounting used page_count * page_size, which counts freed (freelist) pages because SQLite does not shrink the file on delete. After a fill -> claim/delete -> refill cycle the queue stayed at its high-water mark and permanently rejected new records. Switch to live-page accounting ((page_count - freelist_count) * page_size) with defensive nonnegative handling, keeping the check inside the existing BEGIN IMMEDIATE transaction under the instance lock (atomicity/thread safety retained). Add a regression test that fills near a small cap, claims/deletes every record, and proves new records can be stored again. - Security hardening: create the leaf queue directory with mode=0o700 before the existing chmod, and after WAL journal initialization restrict the DB and any existing -wal/-shm sidecars to mode 0600 on POSIX (relying on the 0700 directory for future sidecar creation). Add targeted POSIX tests (skipped on Windows) plus a Windows-runnable unit test for the sidecar restriction, without weakening the existing ownership/symlink checks. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Fixes two defects identified in re-review:
1. Backoff overflow (durable_delivery.py)
- 2.0 ** failure_count raised OverflowError once failure_count
reached ~1024 (Python float / C double limit).
- Added _MAX_BACKOFF_EXPONENT = ceil(log2(cap / floor)) = 9,
derived directly from _RETRY_AFTER_CAP_SECONDS / _RETRY_AFTER_FLOOR_SECONDS.
- Clamp the exponent in _full_jitter_backoff before computing the power.
- Saturate state.failure_count at _MAX_BACKOFF_EXPONENT in
ecord_retryable_failure so the counter cannot grow unbounded.
2. Replay thread fragility (replay_coordinator.py)
- An unexpected exception escaping
un_once inside _run_loop
would terminate the daemon thread permanently and silently disable
all future replay.
- Wrapped the inner drain loop with �xcept Exception: _logger.exception(…)
so the outer loop continues on the next periodic wake.
- BaseException is not caught, preserving clean shutdown behaviour.
Regression tests added:
- test_record_retryable_failure_never_raises_beyond_exponent_1024
- test_backoff_stays_capped_at_3600_seconds_beyond_exponent_1024
- test_half_open_behavior_preserved_after_high_failure_count
- test_failure_count_does_not_grow_unbounded
- test_run_loop_survives_unexpected_exception_from_run_once
All 34 durable-delivery/replay tests pass; 396 A365 non-integration tests
pass; pylint 10.00/10; mypy clean on changed production files.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Address all four review findings against the durable-delivery batch
span processor hardening:
1. Wake the worker as soon as the queue reaches max_export_batch_size
instead of only on the periodic schedule/flush/shutdown, closing an
avoidable drop/stall window during bursts between schedule ticks.
Adds a deterministic long-delay-burst regression
(schedule_delay_millis=24h) proving threshold-wake alone drains
repeated bursts with zero drops.
2. Restore fork safety equivalent to upstream BatchSpanProcessor: a
forked child now reinitializes its condition/queue/counters/drop-log
state/worker thread from scratch via a new _init_state_after_fork(),
invoked both from __init__ and from a new _at_fork_reinit(). Adds an
os.register_at_fork(after_in_child=...) hook wrapped in a
weakref.WeakMethod (avoiding a permanent leak, since register_at_fork
has no unregister API), plus a PID guard
(_check_fork_reinit()/self._pid) checked on every _enqueue() call as
a fallback. Adds POSIX-only real os.fork() tests (via
multiprocessing.Process) proving the child exports independently and
that force_flush()/shutdown() do not hang in the child; these
correctly skip on Windows.
3. Preserve OTEL_BSP_MAX_QUEUE_SIZE, OTEL_BSP_SCHEDULE_DELAY,
OTEL_BSP_MAX_EXPORT_BATCH_SIZE, and OTEL_BSP_EXPORT_TIMEOUT
environment-variable defaults when the corresponding constructor arg
is None, via a new _int_env_default() helper mirroring upstream's
_default_* pattern: invalid (non-integer) env values are logged and
fall back to the hardcoded default instead of raising. Explicit
constructor args still always take precedence.
4. Move per-drop and post-shutdown-drop logging out of _enqueue()'s
critical section: rejection paths now only set a local drop_reason
string while self._condition is held, and the new
_log_dropped_span() throttled logger call happens strictly after the
lock is released. Throttling uses a separate _drop_log_lock (never
held during the actual log call) so a stuck/blocking log handler can
never stall the condition or any producer. Adds a test that attaches
a deliberately-blocking log handler and proves neither the condition
nor a second concurrent producer is blocked, plus a test that 100
rapid post-shutdown drops produce exactly 1 throttled log record.
Also redesigns the pre-existing
test_on_end_capacity_race_never_silently_evicts, whose static
queue-length snapshot assumption ("the worker cannot wake spontaneously
during the race window") is invalidated by the correct threshold-wake
fix, given max_export_batch_size <= max_queue_size is already enforced.
Now asserts conservation (accepted+dropped==total), no duplicate
exports, and a bounded accepted-count range. Adds a new deterministic
test_on_end_rejects_explicitly_once_queue_is_at_capacity that manually
pre-fills the queue under the lock (bypassing notify_all()) to restore
full race-free coverage of "reject at capacity, don't evict".
Verified via TDD (red then green) on Windows, plus real POSIX
execution in a Linux container (mounting the root venv's
site-packages read-only, no network access) which caught two genuine
bugs invisible to source-only review: the capacity-race test's
invalidated timing assumption, and a new fork test's incorrect
exact-batch-count assumption under real concurrent thread
interleaving (fixed to assert total delivered count instead).
Test results: 25 passed/3 skipped on Windows (POSIX-only fork tests
skip), 28 passed on Linux (all included), stable across repeated runs.
TestA365BatchProcessorKwargs: 8 passed. Broader distro -k "A365" and
full-repo-suite failures confirmed identical before/after via git
stash (pre-existing/unrelated). A365 non-integration suite: 462
passed/7 skipped, zero regressions (one unrelated, pre-existing flaky
test in test_persistent_storage.py observed and documented,
unaffected by this change).
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Close the gap where exporter.shutdown() would close durable storage and the HTTP session after only a fixed 5s join, even if the replay thread was still mid-send. shutdown() now waits (unbounded) for the replay thread to actually exit before closing storage/session, and concurrent shutdown() callers wait for that single owner's cleanup to finish instead of returning immediately. - ReplayCoordinator.shutdown(timeout_seconds=None) defaults to an unbounded join; a finite value still performs a bounded wait. Recording the stop request happens regardless of the timeout, and a defensive guard makes a (currently unreachable) self-join return False instead of raising. - _Agent365Exporter.shutdown() now has a single cleanup owner (the first caller under the lock); it joins replay unbounded, then closes storage and the session, then signals a completion event. Non-owner callers wait on that event instead of returning early, so every call only returns once cleanup has actually finished. - Added regression coverage: exporter shutdown blocking on an in-flight replay send before closing storage/session, concurrent shutdown callers closing storage/session exactly once, ReplayCoordinator's new unbounded default, and a durable-restart lifecycle test for the same shutdown ordering. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
- Fix Windows clock-race flake in test_expired_records_are_cleaned_up by inserting the pre-expired record via insert_raw_record (established convention) instead of two back-to-back time.time() calls that could tie on Windows' coarse clock. - Remove redundant local urlparse reimport in _post_once (already imported at module scope). - Apply black (line-length=120) to 11 files that had drifted from canonical formatting; no functional changes. - Strip pre-existing trailing whitespace / EOF blank lines in build/getting_started_with_kairo_exporter and docs/superpowers files so git diff --check origin/main...HEAD is clean. - Document replay endpoint/token reconstruction, poison-record discarding, and drain-safe shutdown in A365_DOCUMENTATION.md and CHANGELOG.md. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Performance comparisonThreshold: regressions >15.0% on gating scenarios fail the build. Higher ops/s is better; positive Δ means the PR is slower.
|
There was a problem hiding this comment.
Pull request overview
This PR adds retry-resilient, durable “store-and-forward” delivery for the Agent365 exporter, aligning Python behavior with the hardened durability/retry contract established in the .NET distro. It introduces a per-identity transmission gate with Retry-After support, a secure SQLite-backed queue, and a dedicated drain-safe batch worker to make replay and shutdown deterministic and fork-safe.
Changes:
- Implement durable delivery primitives:
DeliveryResult/DeliveryDisposition, per-identityTransmissionGate, HTTP-dateRetry-Afterparsing, and a backgroundReplayCoordinator. - Add secure SQLite persistence (
PersistentStorage) with schema migration, poisoning/validation behavior, leases, retention, and capacity accounting. - Replace the SDK
BatchSpanProcessorusage on the A365 path with an A365-owned batch worker and expose offline-storage configuration end-to-end (distro/options/helpers), with expanded tests and documentation.
Reviewed changes
Copilot reviewed 29 out of 29 changed files in this pull request and generated 1 comment.
Show a summary per file
| File | Description |
|---|---|
| tests/test_distro.py | Adds validation/forwarding coverage for batch kwargs and new offline-storage distro kwargs. |
| tests/a365/test_utils.py | Updates parse_retry_after tests to cover HTTP-date parsing and relative delta behavior. |
| tests/a365/test_replay_coordinator.py | Adds comprehensive unit tests for replay pass behavior, failure handling, wake/shutdown semantics. |
| tests/a365/test_persistent_storage.py | Adds extensive tests for SQLite queue behavior: migration, validation, retention, leases, capacity, and permissions. |
| tests/a365/test_payload_chunking.py | Updates exporter chunking tests to the new _post_once/DeliveryResult contract and durable-delivery gating behavior. |
| tests/a365/test_handler.py | Adds tests ensuring offline-storage options are surfaced and forwarded through the public handler APIs. |
| tests/a365/test_enriching_span_processor.py | Adds large suite validating the new A365-owned batch processor (capacity, lifecycle, env defaults, fork-safety). |
| tests/a365/test_durable_restart.py | Adds restart/durability tests validating replay with fresh tokens/endpoints and drain-safe shutdown. |
| tests/a365/test_durable_delivery.py | Adds tests for delivery dispositions and TransmissionGate retry/backoff behavior (including overflow bounds). |
| tests/a365/test_contextual_token_resolver.py | Updates contextual token resolver tests to new send contract and durable-delivery-disabled behavior. |
| tests/a365/test_circuit_breaker.py | Migrates legacy circuit-breaker tests to TransmissionGate and exporter integration semantics. |
| src/microsoft/opentelemetry/a365/core/exporters/utils.py | Adds HTTP-date Retry-After parsing, storage directory coercion, and forwards offline-storage options through helper construction. |
| src/microsoft/opentelemetry/a365/core/exporters/replay_coordinator.py | Introduces replay coordinator thread/pass logic and error handling model for durable replay. |
| src/microsoft/opentelemetry/a365/core/exporters/persistent_storage.py | Introduces secure SQLite-backed durable queue with schema migration and validation. |
| src/microsoft/opentelemetry/a365/core/exporters/enriching_span_processor.py | Reimplements A365 batch processing with atomic enqueue, threshold wake, deterministic shutdown, and fork-safety. |
| src/microsoft/opentelemetry/a365/core/exporters/durable_delivery.py | Adds delivery result types and per-identity gate with capped jittered exponential backoff and probe semantics. |
| src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter.py | Integrates durable delivery: gating, persistence, replay lifecycle, _post_once classification, and deterministic shutdown. |
| src/microsoft/opentelemetry/a365/core/exporters/agent365_exporter_options.py | Adds offline-storage options and validation to exporter options. |
| src/microsoft/opentelemetry/a365/core/exporters/init.py | Re-exports durable delivery helper types for package consumers. |
| src/microsoft/opentelemetry/_distro.py | Adds distro kwargs for offline storage and validates batch/storage options before component construction. |
| src/microsoft/opentelemetry/_constants.py | Adds constants for new distro kwargs. |
| README.md | Documents new offline-storage knobs for use_microsoft_opentelemetry. |
| docs/superpowers/specs/2026-08-14-a365-durable-delivery-hardening-design.md | Adds hardening design spec referencing parity goals and behavioral guarantees. |
| docs/superpowers/specs/2026-08-12-a365-durable-delivery-design.md | Adds initial durable-delivery design spec and contract description. |
| docs/superpowers/plans/2026-08-14-a365-durable-delivery-hardening.md | Adds implementation plan for hardening tasks and associated regression tests. |
| docs/superpowers/plans/2026-08-12-a365-durable-delivery.md | Adds implementation plan for initial durable delivery feature. |
| CHANGELOG.md | Documents new offline-storage options and durable delivery semantics. |
| A365_DOCUMENTATION.md | Adds durable delivery documentation and security considerations for on-disk payload storage. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| # Real (not simulated) fork-safety coverage needs the multiprocessing "fork" | ||
| # start method, which only exists on POSIX. Resolve this once, defensively, | ||
| # so a start method already configured by another plugin/module never turns | ||
| # into an import-time crash -- it just disables the POSIX-only tests below. | ||
| _FORK_AVAILABLE = hasattr(os, "fork") | ||
| if _FORK_AVAILABLE: | ||
| try: | ||
| if multiprocessing.get_start_method(allow_none=True) is None: | ||
| multiprocessing.set_start_method("fork") | ||
| _FORK_AVAILABLE = multiprocessing.get_start_method(allow_none=True) == "fork" | ||
| except RuntimeError: | ||
| _FORK_AVAILABLE = False |
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Summary
Retry-Aftersupport, jittered exponential backoff, and one half-open probeHardening parity
Retry-Afterhandling with a one-hour capOTEL_BSP_*environment variables, and defaultsSQLite is intentionally used instead of the .NET file-per-record storage implementation; the externally observable retry, replay, safety, and lifecycle guarantees are aligned.
Validation
469 passed, 8 skippedfor non-integrationtests/a365116 passed, 4 skipped, 8 subtests passedfor focused exporter, replay, processor, and distro coveragegit diff --check origin/main...HEAD: cleanpyproject.tomlanduv.lock: unchangedThe local full-suite environment has a pre-existing Azure Monitor exporter API mismatch; before the final hardening fixes, the branch and
origin/mainhad identical failing/erroring test IDs. The affected A365 suites were rerun after the final fixes.