Span Buffer Multiprocess Enhancement with Health Monitoring - #4
Span Buffer Multiprocess Enhancement with Health Monitoring#4ron-x5labs wants to merge 1 commit into
Conversation
|
Warning Review limit reached
Next review available in: 54 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (40)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
ron-x5labs
left a comment
ron-x5labs
left a comment
There was a problem hiding this comment.
Code Review: Span Buffer Multiprocess Enhancement with Health Monitoring
Problem
Refactor SpanFlusher from a single background process to N parallel processes (one per shard, or shared with a --flusher-processes cap), each with its own SpansBuffer and per-process health/backpressure tracking. The CI workflow deletions (~30 files) appear to be benchmark-PR diff noise and are not reviewed.
Solution Reviewed
SpanFlusher now builds a round-robin shard→process map (num_processes = min(max_processes, len(assigned_shards))), spawns one process per entry, each with its own SpansBuffer(shard_subset) and its own multiprocessing.Value for healthy_since/backpressure_since, and its own restart counter. The factory adds a flusher_processes param and a --flusher-processes CLI option (default 1, preserving old single-process behavior in production). submit() iterates all per-process buffers for record_stored_segments/get_memory_info and all per-process backpressure values; _ensure_processes_alive() loops all processes and restarts dead/hung ones in place.
Summary
Solid multiprocess refactor that preserves the single-process default for production rollout, but there are real issues: an unbounded/zero-validated CLI option that can silently disable flushing, metric tag inconsistencies that split observability at the healthy/unhealthy boundary, a join() that can skip terminating processes when the deadline elapses mid-wait, and redundant cluster-wide Redis queries scaling with process count. None are catastrophic at the default of 1 process, but several bite as soon as --flusher-processes > 1.
Files Reviewed
src/sentry/spans/consumers/process/flusher.py— deeply reviewed (main refactor)src/sentry/spans/consumers/process/factory.py— deeply reviewedsrc/sentry/consumers/__init__.py— deeply reviewed (CLI option)src/sentry/spans/buffer.py— deeply reviewed (context:flush_segments,get_memory_infocontract — not modified by this PR)tests/sentry/spans/consumers/process/test_consumer.py— deeply reviewedtests/sentry/spans/consumers/process/test_flusher.py— deeply reviewedCLAUDE.md— lightly reviewed (unrelatedhasattr/isinstancestyle note).github/workflows/*.yml(30 files) — skipped: benchmark-PR diff noise, not part of the feature
Verification
python3 -m py_compileon all 5 changed.pyfiles — passed (syntax valid)- pytest — skipped: no venv/devenv available in this environment (
sentry/pytestmodules not importable); could not runtest_consumer.py/test_flusher.py
Issues Found
See inline comments for blocking and non-blocking findings. Additional non-blocking note (no inline anchor — buffer.py is not modified by this PR): flush budget amplification — flush_segments computes shard_factor = len(self.assigned_shards) and max_segments_per_shard = ceil(max_flush_segments / shard_factor) on the per-process buffer (a shard subset), not the full shard set. With --flusher-processes N each process independently caps at max_flush_segments for its own shards, so total flush rate becomes ~N × spans.buffer.max-flush-segments and the any_shard_at_limit backpressure threshold rises (ceil(500/2)=250 vs ceil(500/8)=63). Likely intentional (more processes = more parallel flush throughput), but the option's meaning shifts from a global cap to a per-process cap when multiprocess is enabled — operators tuning max-flush-segments may not realize enabling --flusher-processes 4 quadruples the effective flush rate. Document the per-process semantics or scale by the global shard count.
💡 Suggestions
- flusher.py:127 —
_create_process_for_shard(self, shard: int)(singular) is dead code —_ensure_processes_alivecalls_create_process_for_shards(process_index, shards)(plural). Remove the unused variant to avoid confusion about the restart entry point. - factory.py:41 / consumers/init.py:434 — The factory's Python default (
flusher_processes: int | None = None→ one-process-per-shard) diverges from the CLI default of 1 (single shared process). In production this is safe because the CLI always supplies an int, but the effective default depends on the entry point. Worth a comment documenting that programmatic construction enables multiprocess while the CLI stays single-process unless operators opt in.
Verdict
Recommend changes before merge — the --flusher-processes 0/negative silent-disable and the metric tag inconsistency are the must-fix items; the join() and redundant-memory-query issues should be addressed before enabling --flusher-processes > 1 in production.
| click.Option( | ||
| ["--flusher-processes", "flusher_processes"], | ||
| default=1, | ||
| type=int, |
There was a problem hiding this comment.
🟡 Non-blocking — unbounded --flusher-processes can silently disable flushing. type=int with no lower bound means --flusher-processes 0 is silently coerced by max_processes = max_processes or len(buffer.assigned_shards) (flusher.py:51) into one-process-per-shard (0 is falsy — probably not what the operator meant), and a negative value (e.g. -1) is truthy so num_processes = min(-1, len(shards)) = -1, making range(-1) empty → zero flusher processes are spawned and self.processes/self.buffers stay empty. Segments are then never flushed (silent data stall) while the consumer keeps accepting spans. Use click.IntRange(min=1) (or validate >= 1) so a bad operator value fails loudly instead of silently disabling flushing.
| produce(kafka_payload) | ||
|
|
||
| with metrics.timer("spans.buffer.flusher.wait_produce"): | ||
| with metrics.timer("spans.buffer.flusher.wait_produce", tags={"shards": shard_tag}): |
There was a problem hiding this comment.
🔴 Blocking — metric tag-key typo splits the flush pipeline. spans.buffer.flusher.wait_produce uses tag key "shards" (plural) while spans.buffer.flusher.produce (line 185) and spans.buffer.segment_size_bytes (line 195) use "shard" (singular) for the same dimension (the comma-joined shard list). Same conceptual dimension under two tag keys means a dashboard scoped to shard=... silently omits wait_produce, and one scoped to shards=... omits produce/size. The plural is almost certainly a typo — use one key (singular "shard") for all three.
| # Report unhealthy for all shards handled by this process | ||
| for shard in shards: | ||
| metrics.incr( | ||
| "spans.buffer.flusher_unhealthy", tags={"cause": cause, "shard": shard} |
There was a problem hiding this comment.
🔴 Blocking — mixed value types under the same shard tag concept. produce/segment_size_bytes/wait_produce set shard to a comma-joined string like "0,4,8,12" (the per-process shard subset), but spans.buffer.flusher_unhealthy (here) sets shard to an individual int (0, then 4, …). For the same physical shard 4, healthy-path metrics carry shard="0,4,8,12" while the unhealthy metric carries shard=4. Filtering shard=4 returns only flusher_unhealthy and never produce/size/wait; filtering shard="0,4,8,12" returns produce metrics but not per-shard unhealthy. Observability is split exactly at the healthy/unhealthy boundary where correlation matters most. The comma-joined value also churns as the shard→process mapping changes with --flusher-processes and rebalances (cardinality drift). The codebase already has a stable per-shard int convention (shard_i in buffer.py record_stored_segments/num_segments_per_shard); reuse shard_i with individual ints instead of inventing comma-joined shard strings.
| except ValueError: | ||
| pass # Process already closed, ignore | ||
| if self.process_restarts[process_index] > MAX_PROCESS_RESTARTS: | ||
| raise RuntimeError( |
There was a problem hiding this comment.
🟡 Non-blocking — RuntimeError short-circuits the process loop before cleanup. The raise at line 248 is inside for process_index, process in self.processes.items(), placed before process.kill() (line 255) and _create_process_for_shards (line 259) for the current process, and before health-checking any remaining processes. Consequences: (a) the process that exceeded MAX_PROCESS_RESTARTS is left un-killed/uncleaned; (b) any other unhealthy processes later in the iteration are not reported via spans.buffer.flusher_unhealthy and not restarted this cycle. The consumer is being intentionally crashed so leftover cleanup mostly happens on exit, but the missing flusher_unhealthy metrics for the other sick processes is an observability gap at the moment of failure. Recommend: emit all unhealthy metrics across all processes first, then raise a single aggregated RuntimeError after the loop (or after restarting recoverable ones).
| self.process_restarts += 1 | ||
| self._create_process() | ||
| try: | ||
| if isinstance(process, multiprocessing.Process): |
There was a problem hiding this comment.
🟡 Non-blocking — process.kill() is a no-op for threads, orphaning hung threads on restart. The guard if isinstance(process, multiprocessing.Process): process.kill() skips threading.Thread entirely. When produce_to_pipe is set (tests, and any in-process mode), make_process = threading.Thread; a hung thread is never killed and the restart spawns a second thread racing the stuck one on the same backpressure_since/healthy_since Values. Add an explicit thread path (e.g. process.join(timeout=…) or set the daemon flag and document that thread-mode restart cannot forcibly stop a hung thread) so the thread case isn't silently leaked.
| if max_memory_percentage < 1.0: | ||
| memory_infos = list(self.buffer.get_memory_info()) | ||
| memory_infos: list[ServiceMemory] = [] | ||
| for buffer in self.buffers.values(): |
There was a problem hiding this comment.
🟡 Non-blocking — N× redundant cluster-wide Redis memory queries. SpansBuffer.get_memory_info() returns iter_cluster_memory_usage(self.client) (buffer.py:376) which is cluster-wide (all master nodes), not per-shard. The new submit() loops over all N buffers (lines 299–301) and sums: used = N × actual_cluster_used, available = N × actual_cluster_available. The ratio cancels N so the backpressure decision is still correct, but this performs N redundant Redis cluster INFO round-trips per submit() (was 1 in the single-process design) and the absolute used/available are inflated N-fold (wrong if ever logged/alerted). One get_memory_info() call suffices — the per-buffer loop implies per-shard granularity that does not exist.
| for process_index, process in self.processes.items(): | ||
| if deadline is not None: | ||
| remaining_time = deadline - time.time() | ||
| if remaining_time <= 0: |
There was a problem hiding this comment.
🟡 Non-blocking — join() can skip terminating later processes when the deadline elapses. When deadline is not None and remaining_time <= 0 mid-loop, the code breaks out of the for process_index, process in self.processes.items() loop. Processes later in iteration order are neither waited on nor terminate()d (the terminate() at line 346 is inside the loop body, skipped by the break). With N processes and a tight timeout, later processes can be left running after join() returns. Recommend: separate the wait loop from the terminate loop — always terminate() every multiprocessing.Process after the wait phase, regardless of deadline.
|
|
||
| step.poll() | ||
| # Give flusher threads time to process after drift change | ||
| time.sleep(0.1) |
There was a problem hiding this comment.
🟡 Non-blocking — time.sleep(0.1) is a no-op under the test's monkeypatch. test_basic patches time.sleep to lambda _: None (line 15), so the added time.sleep(0.1) at line 62 does nothing — it does not actually wait for the flusher thread to process after the drift change. The test passes only because step.join() later drains the thread. If the intent was to give the background thread time to observe the drift, use a real sleep via a non-patched clock or assert on the flushed output before join. As written, the added line is dead and misleading.
| } | ||
|
|
||
|
|
||
| @pytest.mark.django_db(transaction=True) |
There was a problem hiding this comment.
🟡 Non-blocking — test_flusher_processes_limit only asserts structure, not behavior. The test verifies len(flusher.processes) == 2 and shard distribution counts, but never submits spans or asserts that flushing actually occurs across the 2 processes. It would still pass if the flusher processes did nothing. Add a minimal flush assertion (submit a span, advance drift, assert a message is produced) to validate the multiprocess path end-to-end, not just the wiring.
1 participant
Benchmark PR recreated from getsentry#93824