fix: subscription executors should retry when there are too many concurrent queries (#7547)

when there are too many concurrent queries, the subscription executors
shouldn't crash

https://www.notion.so/sentry/snuba-pods-crashlooping-2af8b10e4b5d80e0bd96fe9f3f2b5241?source=copy_link

---------

Co-authored-by: Rachel Chen <[email protected]>
Co-authored-by: Rachel Chen <[email protected]>

Author: 3 people

snuba/subscriptions/executor_consumer.py

@@ -93,9 +93,7 @@ def build_executor_consumer(
 ) -> StreamProcessor[KafkaPayload]:
     # Validate that a valid dataset/entity pair was passed in
     dataset = get_dataset(dataset_name)
-    dataset_entity_names = [
-        get_entity_name(e).value for e in dataset.get_all_entities()
-    ]
+    dataset_entity_names = [get_entity_name(e).value for e in dataset.get_all_entities()]
 
     # Only entities in the same dataset with the same scheduled and result topics
     # may be run together
@@ -197,9 +195,7 @@ def create_with_partitions(
             self.__total_partition_count,
             self.__total_concurrent_queries,
         )
-        self.__metrics.gauge(
-            "calculated_max_concurrent_queries", calculated_max_concurrent_queries
-        )
+        self.__metrics.gauge("calculated_max_concurrent_queries", calculated_max_concurrent_queries)
 
         strategy: ProcessingStrategy[KafkaPayload] = ExecuteQuery(
             self.__dataset,
@@ -242,15 +238,11 @@ def __init__(
         self.__encoder = SubscriptionScheduledTaskEncoder()
         self.__result_encoder = SubscriptionTaskResultEncoder()
 
-        self.__queue: Deque[
-            Tuple[Message[KafkaPayload], SubscriptionTaskResultFuture]
-        ] = deque()
+        self.__queue: Deque[Tuple[Message[KafkaPayload], SubscriptionTaskResultFuture]] = deque()
 
         self.__closed = False
 
-        self.__concurrent_gauge: Gauge = ThreadSafeGauge(
-            self.__metrics, "executor.concurrent"
-        )
+        self.__concurrent_gauge: Gauge = ThreadSafeGauge(self.__metrics, "executor.concurrent")
 
         self.__concurrent_clickhouse_gauge: Gauge = ThreadSafeGauge(
             self.__metrics, "executor.concurrent.clickhouse"
@@ -261,9 +253,7 @@ def __execute_query(
     ) -> Tuple[SubscriptionRequest, Result]:
         # Measure the amount of time that took between the task's scheduled
         # time and it beginning to execute.
-        self.__metrics.timing(
-            "executor.latency", (time.time() - task.timestamp.timestamp()) * 1000
-        )
+        self.__metrics.timing("executor.latency", (time.time() - task.timestamp.timestamp()) * 1000)
 
         timer = Timer("query")
 
@@ -297,16 +287,18 @@ def poll(self) -> None:
             try:
                 transformed_message = message.replace(
                     self.__result_encoder.encode(
-                        SubscriptionTaskResult(
-                            result_future.task, result_future.future.result()
-                        )
+                        SubscriptionTaskResult(result_future.task, result_future.future.result())
                     )
                 )
             except QueryException as exc:
                 cause = exc.__cause__
                 if isinstance(cause, ClickhouseError):
                     if cause.code in NON_RETRYABLE_CLICKHOUSE_ERROR_CODES:
                         logger.exception("Error running subscription query %r", exc)
+                        self.__metrics.increment(
+                            "subscription_executor_nonretryable_error",
+                            tags={"error_type": str(cause.code)},
+                        )
                     else:
                         raise SubscriptionQueryException(exc.message)
 
@@ -321,9 +313,7 @@ def submit(self, message: Message[KafkaPayload]) -> None:
         # we will start raising MessageRejected to slow down the consumer as
         # it means our executor cannot keep up
         queue_size_factor = state.get_config("executor_queue_size_factor", 10)
-        assert (
-            queue_size_factor is not None
-        ), "Invalid executor_queue_size_factor config"
+        assert queue_size_factor is not None, "Invalid executor_queue_size_factor config"
         max_queue_size = self.__max_concurrent_queries * queue_size_factor
 
         # Tell the consumer to pause until we have removed some futures from
@@ -343,8 +333,7 @@ def submit(self, message: Message[KafkaPayload]) -> None:
         # Don't execute stale subscriptions
         if (
             self.__stale_threshold_seconds is not None
-            and time.time() - datetime.timestamp(task.timestamp)
-            >= self.__stale_threshold_seconds
+            and time.time() - datetime.timestamp(task.timestamp) >= self.__stale_threshold_seconds
         ):
             should_execute = False
 
@@ -354,9 +343,7 @@ def submit(self, message: Message[KafkaPayload]) -> None:
                     message,
                     SubscriptionTaskResultFuture(
                         task,
-                        self.__executor.submit(
-                            self.__execute_query, task, tick_upper_offset
-                        ),
+                        self.__executor.submit(self.__execute_query, task, tick_upper_offset),
                     ),
                 )
             )
@@ -385,9 +372,7 @@ def join(self, timeout: Optional[float] = None) -> None:
             message, result_future = self.__queue.popleft()
 
             transformed_message = self.__result_encoder.encode(
-                SubscriptionTaskResult(
-                    result_future.task, result_future.future.result(remaining)
-                )
+                SubscriptionTaskResult(result_future.task, result_future.future.result(remaining))
             )
 
             self.__next_step.submit(message.replace(transformed_message))

snuba/web/constants.py

@@ -21,6 +21,7 @@
 NON_RETRYABLE_CLICKHOUSE_ERROR_CODES = {
     ErrorCodes.MEMORY_LIMIT_EXCEEDED,
     ErrorCodes.TOO_SLOW,
+    ErrorCodes.TOO_MANY_SIMULTANEOUS_QUERIES,
 }
 
 LW_DELETE_NON_RETRYABLE_CLICKHOUSE_ERROR_CODES = {