fix(celery) Use model_name in process_incr (#87863)

Start using the model name instead of the model class when scheduling
buffer process tasks. By replacing model with model_name we remove the
last pickle parameters from celery. Support for `model` parameter has
been preserved as when these changes ship, we'll have tasks in queues
using `model`.

Refs #81913

Author: markstory

src/sentry/buffer/base.py

@@ -123,7 +123,7 @@ def incr(
         """
         process_incr.apply_async(
             kwargs={
-                "model": model,
+                "model_name": f"{model._meta.app_label}.{model._meta.model_name}",
                 "columns": columns,
                 "filters": filters,
                 "extra": extra,

src/sentry/celery.py

@@ -10,12 +10,8 @@
 
 from sentry.utils import metrics
 
-LEGACY_PICKLE_TASKS = frozenset(
-    [
-        # basic tasks that must be passed models still
-        "sentry.tasks.process_buffer.process_incr",
-    ]
-)
+# XXX: Pickle parameters are not allowed going forward
+LEGACY_PICKLE_TASKS: frozenset[str] = frozenset([])
 
 
 def holds_bad_pickle_object(value, memo=None):
@@ -44,7 +40,8 @@ def holds_bad_pickle_object(value, memo=None):
             "django database models are large and likely to be stale when your task is run. "
             "Instead pass primary key values to the task and load records from the database within your task.",
         )
-    if type(value).__module__.startswith(("sentry.", "getsentry.")):
+    app_module = type(value).__module__
+    if app_module.startswith(("sentry.", "getsentry.")):
         return value, "do not pickle custom classes"
 
     return None

src/sentry/event_manager.py

@@ -1800,6 +1800,10 @@ def _process_existing_aggregate(
         **incoming_metadata,
         "title": _get_updated_group_title(existing_metadata, incoming_metadata),
     }
+    initial_priority = updated_group_values["data"]["metadata"].get("initial_priority")
+    if initial_priority is not None:
+        # cast to an int, as we don't want to pickle enums into task args.
+        updated_group_values["data"]["metadata"]["initial_priority"] = int(initial_priority)
 
     # We pass `times_seen` separately from all of the other columns so that `buffer_inr` knows to
     # increment rather than overwrite the existing value

src/sentry/tasks/process_buffer.py

@@ -70,7 +70,21 @@ def process_incr(
     """
     from sentry import buffer
 
-    if not model and model_name:
+    if model:
+        # Using model parameter in the celery task is deprecated
+        # as we're trying to eliminate parameters that require pickle
+        logger.info(
+            "process_incr.model_kwarg",
+            extra={
+                "model": model,
+                "columns": columns,
+                "filters": filters,
+                "extra": extra,
+                "signal_only": signal_only,
+            },
+        )
+
+    if model_name:
         assert "." in model_name, "model_name must be in form `sentry.Group`"
         model = apps.get_model(model_name)
 

tests/sentry/buffer/test_base.py

@@ -22,11 +22,17 @@ def setUp(self):
 
     @mock.patch("sentry.buffer.base.process_incr")
     def test_incr_delays_task(self, process_incr):
-        model = mock.Mock()
+        model = Group
         columns = {"times_seen": 1}
         filters: dict[str, BufferField] = {"id": 1}
         self.buf.incr(model, columns, filters)
-        kwargs = dict(model=model, columns=columns, filters=filters, extra=None, signal_only=None)
+        kwargs = dict(
+            model_name="sentry.group",
+            columns=columns,
+            filters=filters,
+            extra=None,
+            signal_only=None,
+        )
         process_incr.apply_async.assert_called_once_with(kwargs=kwargs, headers=mock.ANY)
 
     def test_process_saves_data(self):

tests/sentry/tasks/test_process_buffer.py

@@ -19,10 +19,10 @@ class ProcessIncrTest(TestCase):
     def test_constraints_model_name(self, process):
         with pytest.raises(AssertionError) as err:
             process_incr(model_name="group", columns={"times_seen": 1}, filters={"pk": 1})
-        assert "model_name" in str(err)
+        assert "model_name must be in form" in str(err)
 
     @mock.patch("sentry.buffer.backend.process")
-    def test_calls_process(self, process):
+    def test_calls_process_with_model(self, process):
         columns = {"times_seen": 1}
         filters = {"pk": 1}
         process_incr(model=Group, columns=columns, filters=filters)