ref(lw-deletes): add project_id killswitch and some logging

snuba/lw_deletions/strategy.py

@@ -18,9 +18,11 @@
 from snuba.datasets.storage import WritableTableStorage
 from snuba.lw_deletions.batching import BatchStepCustom, ValuesBatch
 from snuba.lw_deletions.formatters import Formatter
+from snuba.query.allocation_policies import AllocationPolicyViolations
 from snuba.query.query_settings import HTTPQuerySettings
 from snuba.state import get_int_config
 from snuba.utils.metrics import MetricsBackend
+from snuba.web import QueryException
 from snuba.web.bulk_delete_query import construct_or_conditions, construct_query
 from snuba.web.delete_query import (
     ConditionsType,
@@ -46,6 +48,7 @@ def __init__(
     ) -> None:
         self.__next_step = next_step
         self.__storage = storage
+        self.__storage_name = storage.get_storage_key().value
         self.__cluster_name = self.__storage.get_cluster().get_clickhouse_cluster_name()
         self.__tables = storage.get_deletion_settings().tables
         self.__formatter: Formatter = formatter
@@ -62,11 +65,17 @@ def submit(self, message: Message[ValuesBatch[KafkaPayload]]) -> None:
 
         try:
             self._execute_delete(conditions)
-        except TooManyOngoingMutationsError:
+        except TooManyOngoingMutationsError as err:
             # backpressure is applied while we wait for the
             # currently ongoing mutations to finish
             self.__metrics.increment("too_many_ongoing_mutations")
+            logger.warning(str(err), exc_info=True)
             raise MessageRejected
+        except QueryException as err:
+            cause = err.__cause__
+            if isinstance(cause, AllocationPolicyViolations):
+                self.__metrics.increment("allocation_policy_violation")
+                raise MessageRejected
 
         self.__next_step.submit(message)
 

snuba/web/bulk_delete_query.py

@@ -20,6 +20,7 @@
 from snuba.query.exceptions import InvalidQueryException, NoRowsToDeleteException
 from snuba.query.expressions import Expression
 from snuba.reader import Result
+from snuba.state import get_str_config
 from snuba.utils.metrics.util import with_span
 from snuba.utils.metrics.wrapper import MetricsWrapper
 from snuba.utils.schemas import ColumnValidator, InvalidColumnType
@@ -208,9 +209,14 @@ def delete_from_tables(
     if highest_rows_to_delete == 0:
         return result
 
+    storage_name = storage.get_storage_key().value
+    project_id = attribution_info.tenant_ids.get("project_id")
+    if project_id and should_use_killswitch(storage_name, str(project_id)):
+        return result
+
     delete_query: DeleteQueryMessage = {
         "rows_to_delete": highest_rows_to_delete,
-        "storage_name": storage.get_storage_key().value,
+        "storage_name": storage_name,
         "conditions": conditions,
         "tenant_ids": attribution_info.tenant_ids,
     }
@@ -224,3 +230,10 @@ def construct_or_conditions(conditions: Sequence[ConditionsType]) -> Expression:
     into OR conditions for a bulk delete
     """
     return combine_or_conditions([_construct_condition(cond) for cond in conditions])
+
+
+def should_use_killswitch(storage_name: str, project_id: str) -> bool:
+    killswitch_config = get_str_config(
+        f"lw_deletes_killswitch_{storage_name}", default=""
+    )
+    return project_id in killswitch_config if killswitch_config else False

tests/web/test_bulk_delete_query.py

@@ -108,6 +108,19 @@ def test_deletes_not_enabled_runtime_config() -> None:
         delete_from_storage(storage, conditions, attr_info)
 
 
+@pytest.mark.redis_db
+@patch("snuba.web.bulk_delete_query._enforce_max_rows", return_value=10)
+@patch("snuba.web.bulk_delete_query.produce_delete_query")
+def test_deletes_killswitch(mock_produce_query: Mock, mock_enforce_rows: Mock) -> None:
+    storage = get_writable_storage(StorageKey("search_issues"))
+    conditions = {"project_id": [1], "group_id": [1, 2, 3, 4]}
+    attr_info = get_attribution_info()
+
+    set_config("lw_deletes_killswitch_search_issues", "[1]")
+    delete_from_storage(storage, conditions, attr_info)
+    mock_produce_query.assert_not_called()
+
+
 @pytest.mark.redis_db
 def test_delete_invalid_column_type() -> None:
     storage = get_writable_storage(StorageKey("search_issues"))