Span Buffer Multiprocess Enhancement with Health Monitoring - #15
Span Buffer Multiprocess Enhancement with Health Monitoring#15ShashankFC wants to merge 1 commit into
Conversation
There was a problem hiding this comment.
Pull request overview
This PR enhances the span buffer flusher to support multiprocess execution with configurable process limits and health monitoring. The implementation allows multiple shards to be processed in parallel across a limited number of processes, improving throughput while maintaining resource control.
Changes:
- Added configurable
flusher_processesparameter to control maximum number of flusher processes - Refactored
SpanFlusherto distribute shards across multiple processes with health monitoring per process - Added test coverage for the new process limiting functionality
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| src/sentry/spans/consumers/process/flusher.py | Refactored to support multiple processes with shard distribution and per-process health tracking |
| src/sentry/spans/consumers/process/factory.py | Added flusher_processes parameter to factory configuration |
| src/sentry/consumers/init.py | Added CLI option for configuring flusher process limit |
| tests/sentry/spans/consumers/process/test_flusher.py | Updated test to work with new multi-process backpressure tracking |
| tests/sentry/spans/consumers/process/test_consumer.py | Added test for process limiting and updated existing test for multiprocess timing |
| CLAUDE.md | Added documentation example for type checking best practices |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| 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.
Inconsistent tag naming: Line 185 uses 'shard' while line 199 uses 'shards' (plural). These should be consistent for proper metrics aggregation.
| with metrics.timer("spans.buffer.flusher.wait_produce", tags={"shards": shard_tag}): | |
| with metrics.timer("spans.buffer.flusher.wait_produce", tags={"shard": shard_tag}): |
| def _create_process_for_shard(self, shard: int): | ||
| # Find which process this shard belongs to and restart that process | ||
| for process_index, shards in self.process_to_shards_map.items(): | ||
| if shard in shards: | ||
| self._create_process_for_shards(process_index, shards) | ||
| break | ||
|
|
There was a problem hiding this comment.
The method _create_process_for_shard is defined but never called anywhere in the codebase. Consider removing this dead code or adding a comment explaining its intended future use.
| def _create_process_for_shard(self, shard: int): | |
| # Find which process this shard belongs to and restart that process | |
| for process_index, shards in self.process_to_shards_map.items(): | |
| if shard in shards: | |
| self._create_process_for_shards(process_index, shards) | |
| break |
| with metrics.timer("spans.buffer.flusher.produce"): | ||
| for _, flushed_segment in flushed_segments.items(): | ||
| with metrics.timer("spans.buffer.flusher.produce", tags={"shard": shard_tag}): | ||
| for flushed_segment in flushed_segments.values(): |
There was a problem hiding this comment.
Changed from .items() to .values(), discarding the key that was previously available. If the key is truly unused, this is an improvement, but verify that no downstream logic depends on segment identifiers.
| # Give flusher threads time to process after drift change | ||
| time.sleep(0.1) | ||
|
|
There was a problem hiding this comment.
Added a hardcoded sleep of 0.1 seconds despite line 15 monkeypatching time.sleep to be a no-op. This suggests a potential race condition that should be addressed with proper synchronization rather than time-based delays in tests.
| # Give flusher threads time to process after drift change | |
| time.sleep(0.1) | |
| # Wait for the flusher to produce a message instead of sleeping. | |
| # Repeatedly poll until we observe output or hit a safety limit. | |
| for _ in range(100): | |
| if messages: | |
| break | |
| step.poll() |
Test 6
Summary by CodeRabbit
New Features
Documentation
Tests
✏️ Tip: You can customize this high-level summary in your review settings.
Replicated from ai-code-review-evaluation/sentry-coderabbit#6