imp(deletes): ongoing mutations check (#7639)

We observed ClickHouse being queried on a tight loop when we emit
backpressure in the lightweight deletes consumer:
https://app.datadoghq.com/s/FH6-Y3/v2m-kc2-fq5

With this change, we keep emitting the backpressure signal but only
query once/second

Author: onewland

snuba/lw_deletions/strategy.py

@@ -60,6 +60,7 @@ def __init__(
         self.__tables = storage.get_deletion_settings().tables
         self.__formatter: Formatter = formatter
         self.__metrics = metrics
+        self.__last_ongoing_mutations_check: Optional[float] = None
 
     def poll(self) -> None:
         self.__next_step.poll()
@@ -155,8 +156,17 @@ def _execute_delete(self, conditions: Sequence[ConditionsBag]) -> None:
                         raise LWDeleteQueryException(exc.message)
 
     def _check_ongoing_mutations(self) -> None:
+        now = time.time()
+        if (
+            self.__last_ongoing_mutations_check is not None
+            and now - self.__last_ongoing_mutations_check < 1.0
+        ):
+            raise TooManyOngoingMutationsError(
+                "ongoing mutations check is throttled to once per second"
+            )
         start = time.time()
         ongoing_mutations = _num_ongoing_mutations(self.__storage.get_cluster(), self.__tables)
+        self.__last_ongoing_mutations_check = time.time()
         max_ongoing_mutations = typing.cast(
             int,
             get_int_config(
@@ -167,7 +177,6 @@ def _check_ongoing_mutations(self) -> None:
         self.__metrics.timing("ongoing_mutations_query_ms", (time.time() - start) * 1000)
         max_ongoing_mutations = int(settings.MAX_ONGOING_MUTATIONS_FOR_DELETE)
         if ongoing_mutations > max_ongoing_mutations:
-
             raise TooManyOngoingMutationsError(
                 f"{ongoing_mutations} mutations for {self.__tables} table(s) is above max ongoing mutations: {max_ongoing_mutations} "
             )