Span Buffer Multiprocess Enhancement with Health Monitoring#6
Conversation
WalkthroughThe changes introduce support for multiple flusher processes in the span consumer, allowing parallel span flushing by distributing shards across processes. This includes updates to the flusher implementation, its factory, configuration options, and related tests. Documentation was also updated to clarify type checking anti-patterns in backend code. Changes
Sequence Diagram(s)sequenceDiagram
participant CLI
participant ConsumerFactory
participant FlusherFactory
participant SpanFlusher
participant FlusherProcess1
participant FlusherProcessN
CLI->>ConsumerFactory: Start "process-spans" with --flusher-processes=N
ConsumerFactory->>FlusherFactory: Create with flusher_processes=N
FlusherFactory->>SpanFlusher: Instantiate with max_processes=N
SpanFlusher->>FlusherProcess1: Assign subset of shards
SpanFlusher->>FlusherProcessN: Assign subset of shards
Note over FlusherProcess1,FlusherProcessN: Each process flushes assigned shards in parallel
SpanFlusher->>ConsumerFactory: Report health/backpressure per process
Estimated code review effort🎯 4 (Complex) | ⏱️ ~40 minutes Poem
✨ Finishing Touches
🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🔭 Outside diff range comments (1)
src/sentry/spans/consumers/process/flusher.py (1)
337-348: Fix unused loop variable.The static analysis correctly identified that
process_indexis unused in the loop body.- for process_index, process in self.processes.items(): + for _, process in self.processes.items():
🧹 Nitpick comments (2)
src/sentry/spans/consumers/process/flusher.py (2)
31-32: Update docstring to reflect multi-process architecture.The docstring still mentions "A background multiprocessing manager" (singular) but now creates multiple processes. Update to reflect the new multi-process design.
- A background multiprocessing manager that polls Redis for new segments to flush and to produce to Kafka. - Creates one process per shard for parallel processing. + A background multiprocessing manager that polls Redis for new segments to flush and to produce to Kafka. + Creates multiple processes to handle shards in parallel, with the number of processes configurable + via max_processes parameter.
127-133: Remove unused legacy method.The
_create_process_for_shardmethod appears to be legacy code from the previous single-shard-per-process design and is no longer used.Consider removing this unused method to reduce code complexity.
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
CLAUDE.md(1 hunks)src/sentry/consumers/__init__.py(1 hunks)src/sentry/spans/consumers/process/factory.py(3 hunks)src/sentry/spans/consumers/process/flusher.py(8 hunks)tests/sentry/spans/consumers/process/test_consumer.py(4 hunks)tests/sentry/spans/consumers/process/test_flusher.py(1 hunks)
🧰 Additional context used
🪛 Ruff (0.12.2)
src/sentry/spans/consumers/process/flusher.py
337-337: Loop control variable process_index not used within loop body
Rename unused process_index to _process_index
(B007)
🔇 Additional comments (13)
tests/sentry/spans/consumers/process/test_flusher.py (1)
83-83: LGTM: Test correctly adapted for multi-process backpressure tracking.The assertion change properly reflects the new architecture where backpressure is tracked per flusher process instead of globally. Using
any()to check if at least one process is experiencing backpressure is the correct approach.CLAUDE.md (1)
449-458: LGTM: Excellent addition to the anti-patterns guide.This new example clearly demonstrates the proper way to handle union type checking. Using
isinstance()instead ofhasattr()is indeed the safer and more explicit approach, as it prevents false positives and makes the code's intent clearer.src/sentry/spans/consumers/process/factory.py (2)
41-42: LGTM: Clean parameter addition with proper defaults.The new
flusher_processesparameter is well-designed with appropriate type annotation and default value that maintains backward compatibility.
52-52: LGTM: Proper parameter flow implementation.The parameter is correctly stored as an instance attribute and passed to the SpanFlusher with the appropriate
max_processesparameter name. The implementation follows expected patterns.Also applies to: 74-74
src/sentry/consumers/__init__.py (1)
430-438: LGTM: Well-designed CLI option addition.The new
--flusher-processesoption is properly implemented with:
- Clear naming and parameter mapping
- Sensible default value of 1
- Descriptive help text
- Proper integration with existing multiprocessing options
tests/sentry/spans/consumers/process/test_consumer.py (4)
1-1: LGTM: Appropriate test setup changes.The
timeimport addition andtransaction=Trueparameter are needed to support the test improvements and proper database transaction handling.Also applies to: 12-12
60-63: LGTM: Good timing improvements for threaded test reliability.The additional
poll()call and short sleep properly account for the asynchronous nature of the flusher threads, ensuring they have time to process the drift change before the test continues.
84-123: LGTM: Comprehensive test coverage for multi-process functionality.This new test thoroughly validates the key aspects of the multi-process flusher:
- Process count limits are properly enforced
- Shard distribution works correctly across processes
- Configuration parameters are applied as expected
The test design is solid with clear assertions and realistic scenarios.
82-83: LGTM: Proper function separation formatting.The additional blank lines follow Python style guidelines for separating function definitions.
src/sentry/spans/consumers/process/flusher.py (4)
59-67: LGTM! Clean shard distribution implementation.The modulo-based shard distribution ensures even load balancing across processes. This approach effectively distributes shards in a round-robin fashion.
144-146: Well-implemented shard tagging for observability.The comma-separated shard tag provides good visibility into which shards each process handles, making debugging and monitoring easier.
218-260: Robust multi-process health monitoring implementation.The health check properly monitors all processes independently, tracks restart counts per process, and provides detailed metrics for each affected shard when issues occur.
277-285: Conservative but safe backpressure handling.The implementation correctly checks all processes for backpressure and pauses the entire consumer if any process is under pressure. While this is conservative, it ensures data consistency and prevents partial segment writes.
| self.max_processes = max_processes or len(buffer.assigned_shards) | ||
|
|
There was a problem hiding this comment.
Add validation for max_processes parameter.
The code doesn't validate if max_processes is positive. A zero or negative value would cause issues in the modulo operation for shard distribution.
- self.max_processes = max_processes or len(buffer.assigned_shards)
+ self.max_processes = max_processes or len(buffer.assigned_shards)
+ if self.max_processes <= 0:
+ raise ValueError(f"max_processes must be positive, got {self.max_processes}")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| self.max_processes = max_processes or len(buffer.assigned_shards) | |
| self.max_processes = max_processes or len(buffer.assigned_shards) | |
| if self.max_processes <= 0: | |
| raise ValueError(f"max_processes must be positive, got {self.max_processes}") |
🤖 Prompt for AI Agents
In src/sentry/spans/consumers/process/flusher.py around lines 51 to 52, the
max_processes parameter is assigned without validation, which can cause errors
if it is zero or negative. Add a check to ensure max_processes is a positive
integer before using it; if it is not positive, raise a ValueError or set it to
a default positive value like the length of buffer.assigned_shards to prevent
issues in the modulo operation.
| self.max_processes = max_processes or len(buffer.assigned_shards) | ||
|
|
||
| self.mp_context = mp_context = multiprocessing.get_context("spawn") | ||
| self.stopped = mp_context.Value("i", 0) | ||
| self.redis_was_full = False | ||
| self.current_drift = mp_context.Value("i", 0) | ||
| self.backpressure_since = mp_context.Value("i", 0) | ||
| self.healthy_since = mp_context.Value("i", 0) | ||
| self.process_restarts = 0 | ||
| self.produce_to_pipe = produce_to_pipe | ||
|
|
||
| self._create_process() | ||
|
|
||
| def _create_process(self): | ||
| # Determine which shards get their own processes vs shared processes | ||
| self.num_processes = min(self.max_processes, len(buffer.assigned_shards)) |
There was a problem hiding this comment.
💡 Verification agent
🧩 Analysis chain
Handle edge case of empty assigned shards.
If buffer.assigned_shards is empty and max_processes is not provided, this would result in 0 processes being created, which could cause issues.
🏁 Script executed:
#!/bin/bash
# Check if there are any safeguards against empty assigned_shards in SpansBuffer initialization
ast-grep --pattern 'class SpansBuffer {
$$$
__init__($$$) {
$$$
}
$$$
}'
# Also check for any validation in the buffer implementation
rg -A 5 "assigned_shards" --type pyLength of output: 8173
Ensure SpanFlusher never spins up zero worker processes
The constructor in src/sentry/spans/consumers/process/flusher.py computes:
self.max_processes = max_processes or len(buffer.assigned_shards)
…
self.num_processes = min(self.max_processes, len(buffer.assigned_shards))If buffer.assigned_shards is empty (and max_processes isn’t set), this yields num_processes == 0, creating an empty process_to_shards_map and no workers—which will break downstream logic.
Proposed fixes (choose one):
- Guard against empty shards at the top of
SpanFlusher.__init__:if not buffer.assigned_shards: raise ValueError("SpanFlusher requires at least one assigned_shard")
- Enforce a minimum of one process when computing
num_processes:- self.num_processes = min(self.max_processes, len(buffer.assigned_shards)) + self.num_processes = max(1, min(self.max_processes, len(buffer.assigned_shards)))
Either approach will prevent a zero‐process edge case in production.
🤖 Prompt for AI Agents
In src/sentry/spans/consumers/process/flusher.py around lines 51 to 60, the
calculation of self.num_processes can result in zero if buffer.assigned_shards
is empty and max_processes is not set, causing no worker processes to be created
and breaking downstream logic. To fix this, add a guard at the start of the
SpanFlusher.__init__ method that raises a ValueError if buffer.assigned_shards
is empty, ensuring there is at least one assigned shard before proceeding.
Alternatively, enforce a minimum of one process by adjusting the calculation of
self.num_processes to never be less than one.
2 participants
Test 6
Summary by CodeRabbit