feat(uptime): Add ability to use queues to manage parallelism#12
feat(uptime): Add ability to use queues to manage parallelism#12ShashankFC wants to merge 1 commit into
Conversation
One potential problem we have with batch processing is that any one slow item will clog up the whole batch. This pr implements a queueing method instead, where we keep N queues that each have their own workers. There's still a chance of individual items backlogging a queue, but we can try increased concurrency here to reduce the chances of that happening <!-- Describe your PR here. -->
There was a problem hiding this comment.
Pull request overview
This PR introduces a new "thread-queue-parallel" execution mode for uptime monitoring that enables ordered queue-based processing of remote subscriptions with improved offset tracking and safe commit management.
Changes:
- Added thread-queue-parallel processing mode with ordered queue-based message handling
- Implemented offset tracking system with gap detection to ensure safe Kafka offset commits
- Updated CLI options to support the new execution mode
Reviewed changes
Copilot reviewed 5 out of 7 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/sentry/uptime/consumers/test_results_consumer.py | Added comprehensive tests for thread-queue-parallel mode including ordering, concurrency, offset commit behavior, error handling, and Kafka integration tests |
| tests/sentry/remote_subscriptions/consumers/test_queue_consumer.py | New test file covering OffsetTracker, FixedQueuePool, and SimpleQueueProcessingStrategy components |
| src/sentry/remote_subscriptions/consumers/result_consumer.py | Extended ResultsStrategyFactory to support thread-queue-parallel mode with queue pool initialization and worker creation |
| src/sentry/remote_subscriptions/consumers/queue_consumer.py | New implementation of ordered queue processing with offset tracking, worker threads, and safe commit logic |
| src/sentry/consumers/init.py | Updated CLI options to include thread-queue-parallel mode and revised help text for max-workers parameter |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| while not self.shutdown: | ||
| try: | ||
| work_item = self.work_queue.get() | ||
| except queue.ShutDown: |
There was a problem hiding this comment.
The exception name queue.ShutDown is incorrect. The Python queue module raises queue.Empty when get() times out, but there is no built-in ShutDown exception. This will cause an unhandled exception if the queue is shut down.
| for q in self.queues: | ||
| try: | ||
| q.shutdown(immediate=False) | ||
| except Exception: | ||
| logger.exception("Error shutting down queue") | ||
|
|
There was a problem hiding this comment.
Python's queue.Queue does not have a shutdown() method. This will raise an AttributeError at runtime during graceful shutdown.
| for q in self.queues: | |
| try: | |
| q.shutdown(immediate=False) | |
| except Exception: | |
| logger.exception("Error shutting down queue") |
|
|
||
| with mock.patch.object(type(factory.result_processor), "__call__") as mock_processor_call: | ||
| subscription_2 = self.create_uptime_subscription( | ||
| subscription_id=uuid.uuid4().hex, interval_seconds=300, url="http://santry.io" |
There was a problem hiding this comment.
Corrected spelling of 'santry' to 'sentry'.
| """Process items from the queue in order.""" | ||
| while not self.shutdown: | ||
| try: | ||
| work_item = self.work_queue.get() |
There was a problem hiding this comment.
The get() call blocks indefinitely without a timeout. This makes it difficult to perform graceful shutdown since the thread will block waiting for items even after self.shutdown is set to True. Consider using get(timeout=1.0) and checking self.shutdown between attempts.
| work_item = self.work_queue.get() | |
| work_item = self.work_queue.get(timeout=1.0) | |
| except queue.Empty: | |
| # Periodically wake up to check for shutdown. | |
| continue |
Test 9
Summary by CodeRabbit
New Features
Improvements
✏️ Tip: You can customize this high-level summary in your review settings.
Replicated from ai-code-review-evaluation/sentry-coderabbit#9