ref(span-buffer): Introduce multiprocessed flusher - #1
Open
linxia0415 wants to merge 25 commits into
Open
Conversation
📝 WalkthroughWalkthroughThis PR refactors the Span Flusher from managing a single background worker to distributing span flushing across multiple configurable background workers (one per shard-group). A new ChangesMulti-Process Span Flusher Architecture
Documentation Update
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/sentry/spans/consumers/process/flusher.py`:
- Around line 47-66: Validate and reject non-positive max_processes at the start
of the constructor: if max_processes is not None and max_processes <= 0 raise a
ValueError (or similar) before assigning self.max_processes or computing
self.num_processes; leave None handling unchanged so it still defaults to
len(buffer.assigned_shards). Update the logic near the assignment to
self.max_processes and before computing self.num_processes and building
self.process_to_shards_map to ensure only positive integers are accepted.
- Around line 336-347: The shutdown loop in the flusher that iterates over
self.processes may break when remaining_time <= 0 and skip terminating/joining
the current and remaining worker processes; update the loop in the flusher (the
code that checks process.is_alive() and uses multiprocessing.Process) so that if
the deadline has passed you do not simply break out — instead, for the current
and each remaining process call process.terminate() (for multiprocessing.Process
instances) and then attempt a short join (e.g., process.join(0.1)) to avoid
leaking background processes; ensure you still honor the existing wait loop when
time remains and only fall back to terminate/join when time has expired.
In `@tests/sentry/spans/consumers/process/test_consumer.py`:
- Line 15: The global monkeypatch of time.sleep at the top of the test file
makes the post-drift wait a no-op and causes test_basic to race the background
flusher; change the patch so it only overrides sleep where the consumer imports
it (or apply/restore the patch only around the specific sections that need a
no-op). Specifically, remove or replace the top-level
monkeypatch.setattr("time.sleep", ...) and instead monkeypatch the sleep used by
the consumer module (the time.sleep symbol imported/used by the code under test)
or scope the monkeypatch to the exact block in test_basic that must skip
sleeping so the post-drift wait (lines ~60-62 in test_basic) actually executes a
real wait. Ensure you reference the consumer module's time symbol (the import
used by the background flusher) when patching so only the intended sleep is
stubbed.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: 43881555-0a22-4f77-aac6-1ffb0f6ec467
📒 Files selected for processing (7)
.github/dependabot.ymlCLAUDE.mdsrc/sentry/consumers/__init__.pysrc/sentry/spans/consumers/process/factory.pysrc/sentry/spans/consumers/process/flusher.pytests/sentry/spans/consumers/process/test_consumer.pytests/sentry/spans/consumers/process/test_flusher.py
💤 Files with no reviewable changes (1)
- .github/dependabot.yml
Comment on lines
+47
to
+66
| max_processes: int | None = None, | ||
| produce_to_pipe: Callable[[KafkaPayload], None] | None = None, | ||
| ): | ||
| self.buffer = buffer | ||
| self.next_step = next_step | ||
| 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)) | ||
| self.process_to_shards_map: dict[int, list[int]] = { | ||
| i: [] for i in range(self.num_processes) | ||
| } | ||
| for i, shard in enumerate(buffer.assigned_shards): | ||
| process_index = i % self.num_processes | ||
| self.process_to_shards_map[process_index].append(shard) |
There was a problem hiding this comment.
Reject non-positive max_processes up front.
0 is silently converted to “all shards” by Line 51, and negative values make Lines 60-66 build an invalid process map before failing at runtime. Guard this before deriving self.num_processes.
Suggested fix
def __init__(
self,
buffer: SpansBuffer,
next_step: ProcessingStrategy[FilteredPayload | int],
max_processes: int | None = None,
produce_to_pipe: Callable[[KafkaPayload], None] | None = None,
):
self.next_step = next_step
+ if max_processes is not None and max_processes < 1:
+ raise ValueError("max_processes must be >= 1")
self.max_processes = max_processes or len(buffer.assigned_shards)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sentry/spans/consumers/process/flusher.py` around lines 47 - 66, Validate
and reject non-positive max_processes at the start of the constructor: if
max_processes is not None and max_processes <= 0 raise a ValueError (or similar)
before assigning self.max_processes or computing self.num_processes; leave None
handling unchanged so it still defaults to len(buffer.assigned_shards). Update
the logic near the assignment to self.max_processes and before computing
self.num_processes and building self.process_to_shards_map to ensure only
positive integers are accepted.
Comment on lines
+336
to
+347
| # Wait for all processes to finish | ||
| for process_index, process in self.processes.items(): | ||
| if deadline is not None: | ||
| remaining_time = deadline - time.time() | ||
| if remaining_time <= 0: | ||
| break | ||
|
|
||
| while process.is_alive() and (deadline is None or deadline > time.time()): | ||
| time.sleep(0.1) | ||
|
|
||
| if isinstance(self.process, multiprocessing.Process): | ||
| self.process.terminate() | ||
| if isinstance(process, multiprocessing.Process): | ||
| process.terminate() |
There was a problem hiding this comment.
Timed joins can return with flusher workers still running.
When remaining_time <= 0, the loop breaks before terminating the current or remaining workers. join(timeout=...) can therefore leak background processes past shutdown.
Suggested fix
- for process_index, process in self.processes.items():
- if deadline is not None:
- remaining_time = deadline - time.time()
- if remaining_time <= 0:
- break
-
- while process.is_alive() and (deadline is None or deadline > time.time()):
- time.sleep(0.1)
-
- if isinstance(process, multiprocessing.Process):
- process.terminate()
+ timed_out = False
+ for process in self.processes.values():
+ while process.is_alive():
+ if deadline is not None and deadline <= time.time():
+ timed_out = True
+ break
+ time.sleep(0.1)
+
+ if timed_out:
+ for process in self.processes.values():
+ if isinstance(process, multiprocessing.Process) and process.is_alive():
+ process.terminate()🧰 Tools
🪛 Ruff (0.15.15)
[warning] 337-337: Loop control variable process_index not used within loop body
Rename unused process_index to _process_index
(B007)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/sentry/spans/consumers/process/flusher.py` around lines 336 - 347, The
shutdown loop in the flusher that iterates over self.processes may break when
remaining_time <= 0 and skip terminating/joining the current and remaining
worker processes; update the loop in the flusher (the code that checks
process.is_alive() and uses multiprocessing.Process) so that if the deadline has
passed you do not simply break out — instead, for the current and each remaining
process call process.terminate() (for multiprocessing.Process instances) and
then attempt a short join (e.g., process.join(0.1)) to avoid leaking background
processes; ensure you still honor the existing wait loop when time remains and
only fall back to terminate/join when time has expired.
| @pytest.mark.django_db(transaction=True) | ||
| def test_basic(monkeypatch): | ||
| # Flush very aggressively to make test pass instantly | ||
| monkeypatch.setattr("time.sleep", lambda _: None) |
There was a problem hiding this comment.
The post-drift wait is currently a no-op.
Line 15 monkeypatches time.sleep globally, so Line 62 never actually waits. This keeps test_basic racing the background flusher.
Suggested fix
`@pytest.mark.django_db`(transaction=True)
def test_basic(monkeypatch):
+ real_sleep = time.sleep
# Flush very aggressively to make test pass instantly
monkeypatch.setattr("time.sleep", lambda _: None)
@@
step.poll()
# Give flusher threads time to process after drift change
- time.sleep(0.1)
+ real_sleep(0.1)Also applies to: 60-62
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/sentry/spans/consumers/process/test_consumer.py` at line 15, The global
monkeypatch of time.sleep at the top of the test file makes the post-drift wait
a no-op and causes test_basic to race the background flusher; change the patch
so it only overrides sleep where the consumer imports it (or apply/restore the
patch only around the specific sections that need a no-op). Specifically, remove
or replace the top-level monkeypatch.setattr("time.sleep", ...) and instead
monkeypatch the sleep used by the consumer module (the time.sleep symbol
imported/used by the code under test) or scope the monkeypatch to the exact
block in test_basic that must skip sleeping so the post-drift wait (lines ~60-62
in test_basic) actually executes a real wait. Ensure you reference the consumer
module's time symbol (the import used by the background flusher) when patching
so only the intended sleep is stubbed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Refs STREAM-266
Compression helped a lot bring the time on the main thread down. We now think we can scale down the consumers. However, we also need to ensure the flusher can keep up. So let's make the flusher spawn a process per redis shard.
We think this is easier to do than making the process_spans/insert_spans multiprocessed. Over time we can offload mroe things like statistics/metrics to the flusher, and keep the main thread minimal.
Summary by CodeRabbit
New Features
--flusher-processesconfiguration option to control the number of parallel span flusher worker processes for the process-spans consumer.Documentation
isinstance()overhasattr()for type checking.Chores