Span Buffer Multiprocess Enhancement with Health Monitoring

[everettbu]

Test 6

[greptile-apps]

Greptile Summary

This PR transforms the span processing system from single-process to multi-process architecture with configurable process limits. The core change refactors SpanFlusher to distribute shards across multiple processes using round-robin allocation, enabling parallel processing of spans from Redis to Kafka. Key components include:

Architecture Changes:

• Modified SpanFlusher to create multiple worker processes instead of a single process
• Added shard-to-process mapping logic that distributes shards evenly across available processes
• Implemented per-process health monitoring and backpressure tracking using multiprocessing.Value objects
• Each process group gets its own SpansBuffer instance for independent operation

Configuration Interface:

• Added --flusher-processes CLI option to the process-spans consumer (defaults to 1 for backward compatibility)
• Extended ProcessSpansStrategyFactory with flusher_processes parameter for programmatic control
• When max_processes < number of shards, shards are distributed across available processes

Monitoring Enhancements:

• Per-process health tracking with automatic restart capability for failed processes
• Enhanced metrics with shard-level tagging for better observability
• Backpressure detection now checks across all processes rather than a single global state

Documentation:

• Added anti-pattern guidance in CLAUDE.md about proper type checking with union types (isinstance() vs hasattr())

The integration fits into Sentry's existing span processing pipeline by maintaining the same external interface while parallelizing the internal work distribution, allowing better resource utilization on multi-core systems.

Confidence score: 2/5

• This PR introduces significant architectural complexity with potential for race conditions and resource management issues
• Multiple concerning implementation details including inconsistent metric naming, potential race conditions in process restart logic, and incomplete shutdown procedures
• Critical files src/sentry/spans/consumers/process/flusher.py needs thorough review for process synchronization, cleanup logic, and the shard distribution algorithm

_{6 files reviewed, 3 comments}

_{Edit Code Review Bot Settings | Greptile}

[greptile-apps]

syntax: Inconsistent metric tag naming: line 185 uses 'shard' but line 199 uses 'shards' for the same shard_tag value

Suggested change

	with metrics.timer("spans.buffer.flusher.produce", tags={"shard": shard_tag}):
	for flushed_segment in flushed_segments.values():
	if not flushed_segment.spans:
	continue
	spans = [span.payload for span in flushed_segment.spans]
	kafka_payload = KafkaPayload(None, orjson.dumps({"spans": spans}), [])
	metrics.timing("spans.buffer.segment_size_bytes", len(kafka_payload.value))
	metrics.timing(
	"spans.buffer.segment_size_bytes",
	len(kafka_payload.value),
	tags={"shard": shard_tag},
	)
	produce(kafka_payload)
	with metrics.timer("spans.buffer.flusher.wait_produce"):
	with metrics.timer("spans.buffer.flusher.wait_produce", tags={"shards": shard_tag}):
	with metrics.timer("spans.buffer.flusher.produce", tags={"shard": shard_tag}):
	for flushed_segment in flushed_segments.values():
	if not flushed_segment.spans:
	continue
	spans = [span.payload for span in flushed_segment.spans]
	kafka_payload = KafkaPayload(None, orjson.dumps({"spans": spans}), [])
	metrics.timing(
	"spans.buffer.segment_size_bytes",
	len(kafka_payload.value),
	tags={"shard": shard_tag},
	)
	produce(kafka_payload)
	with metrics.timer("spans.buffer.flusher.wait_produce", tags={"shard": shard_tag}):

[greptile-apps]

logic: Process termination only handles multiprocessing.Process but not threading.Thread cleanup

[greptile-apps]

style: These assertions test the same value (flusher.max_processes and flusher.num_processes both equal 2). Consider removing the redundant assertion or add a comment explaining why both are needed.

[GitHoobar]

Review Summary

🏷️ Draft Comments (6)

Skipped posting 6 draft comments that were valid but scored below your review threshold (>=13/15). Feel free to update them here.

src/sentry/spans/consumers/process/factory.py (1)

175-175: raise InvalidMessage(value.partition, value.offset) in the except block does not use raise ... from err, which can obscure the original exception and hinder debugging.

📊 Impact Scores:

• Production Impact: 2/5
• Fix Specificity: 2/5
• Urgency Impact: 2/5
• Total Score: 6/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/factory.py, line 175, the code raises InvalidMessage in an except block without using 'from None' or 'from err'. Update this line to 'raise InvalidMessage(value.partition, value.offset) from None' to ensure proper exception chaining and avoid masking the original error context.

---

src/sentry/spans/consumers/process/flusher.py (5)

166-183: SpanFlusher now creates one process per shard group, but each process flushes segments in a tight loop with time.sleep(1) when idle, causing unnecessary CPU usage and resource contention at scale.

📊 Impact Scores:

• Production Impact: 3/5
• Fix Specificity: 2/5
• Urgency Impact: 2/5
• Total Score: 7/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/flusher.py, lines 166-183, the flusher process uses a busy loop with `time.sleep(1)` when no segments are flushed, which can cause unnecessary CPU usage when scaled to many processes. Replace `time.sleep(1)` with `stopped.wait(timeout=1)` (using a multiprocessing.Event or similar) to allow the process to sleep efficiently and wake up promptly when stopped.

---

221-259: The SpanFlusher's _ensure_processes_alive method restarts unhealthy processes individually, but does not throttle restarts across all processes, risking cascading restarts and resource exhaustion under mass failure.

📊 Impact Scores:

• Production Impact: 4/5
• Fix Specificity: 3/5
• Urgency Impact: 3/5
• Total Score: 10/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/flusher.py, lines 221-259, the `_ensure_processes_alive` method restarts unhealthy processes without throttling, which can cause resource exhaustion if many processes fail at once. Add a simple restart throttle (e.g., 5 seconds between restarts) to prevent cascading restarts and protect system stability.

---

199-202: The SpanFlusher's main method synchronously waits for all Kafka producer futures after every flush, which can block the process and reduce throughput when producing to many partitions or under high load.

📊 Impact Scores:

• Production Impact: 3/5
• Fix Specificity: 2/5
• Urgency Impact: 2/5
• Total Score: 7/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/flusher.py, lines 199-202, the flusher process waits synchronously for all Kafka producer futures, which can block the process and reduce throughput under high load. Add a timeout (e.g., 2 seconds) to `future.result()` and capture exceptions to avoid indefinite blocking and improve resilience.

---

336-344: The SpanFlusher's join method waits for each process sequentially, which can cause unnecessary delays if one process hangs, impacting shutdown time for the entire system.

📊 Impact Scores:

• Production Impact: 3/5
• Fix Specificity: 2/5
• Urgency Impact: 2/5
• Total Score: 7/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/flusher.py, lines 336-344, the `join` method waits for each process sequentially, which can delay shutdown if one process hangs. Refactor to wait for all processes in parallel, checking their status in a loop, to ensure timely shutdown even if some processes are slow or unresponsive.

---

190-197: SpanFlusher.main does not validate or sanitize the contents of flushed_segment.spans before serializing and producing to Kafka, allowing malicious or malformed span data to be injected and propagated downstream, potentially leading to code execution or data corruption in downstream consumers.

📊 Impact Scores:

• Production Impact: 4/5
• Fix Specificity: 4/5
• Urgency Impact: 3/5
• Total Score: 11/15

🤖 AI Agent Prompt (Copy & Paste Ready):

In src/sentry/spans/consumers/process/flusher.py, lines 190-197, the code serializes and produces `flushed_segment.spans` directly to Kafka without validating the contents. This allows malicious or malformed data to be injected and propagated downstream, risking code execution or data corruption. Update this block to validate that each `span.payload` is a dict (and optionally restrict allowed keys/values), log and skip invalid spans, and only produce if at least one valid span remains. See the suggested code for reference.

---