-
[OPIK-6444] [PY-SDK] fix: improve streamer reliability (#6648)
frostbyte_neo 发布于
2026-05-08 19:02:46 +00:00 - [OPIK-6444] [PY-SDK] fix: serialize flush with BatchManager lock to stop silent trace drops
FlushingThread.run() called batcher.flush() without holding
BatchManager._lock, while BatchManager.process_message holds it for
add(). Messages appended during the post-iteration tail of
_create_batches_from_accumulated_messages() (notably the recursive
payload-size walk in sequence_splitter.split_into_batches) were not in
the returned batches, but were wiped by self._accumulated_messages = [].Pass BatchManager._lock into FlushingThread and acquire it for the full
probe iteration. Also wrap the loop body in try/except so a single bad
flush no longer silently kills the daemon thread.Reproduced ~1.2% drops at production cadence (2s flush interval, steady
traffic) and 22-60% under bursty / aggressive flushing - both go to 0
with the fix. Added two regression tests covering the race and the
thread-death-on-exception path.- fix(sdk): isolate try/except per batcher in FlushingThread
Addresses PR #6648 review: a failing flush on one batcher no longer
skips the remaining batchers in the same probe iteration. Each batcher
gets its own try/except so a transient encoder/anonymizer failure on
one type doesn't starve the others until the next 100ms tick.- refactor(sdk): hide locking behind BatchManager.flush_ready
FlushingThread no longer needs to know about batchers or the BatchManager
lock. It now just invokes a caller-supplied callable on its probe
interval. BatchManager owns the locking + iteration logic in a new
flush_ready() method, which is what FlushingThread calls.Same correctness as before for OPIK-6444, cleaner separation of
concerns. Adds a unit test asserting that a failing batcher in
flush_ready() does not skip the rest in the same tick.- Fix lint errors
下载附件