feat(celery): Support Celery abstract tasks (#1287)

Prior to this change, the Celery integration always instruments
`task.run` and incorrectly instruments `task.__call__` (`task(...)` is
equivalent to `type(task).__call__(...)`, not `task.__call__(...)`).
After this change, we'll use the same logic as Celery to decide whether
to instrument `task.__call__` or `task.run`. That change allows abstract
tasks to catch/raise exceptions before the Sentry wrapper.

Author: johnzeringue

mypy.ini

@@ -59,3 +59,5 @@ ignore_missing_imports = True
 [mypy-sentry_sdk._queue]
 ignore_missing_imports = True
 disallow_untyped_defs = False
+[mypy-celery.app.trace]
+ignore_missing_imports = True

sentry_sdk/integrations/celery.py

@@ -30,6 +30,7 @@
         Ignore,
         Reject,
     )
+    from celery.app.trace import task_has_custom
 except ImportError:
     raise DidNotEnable("Celery not installed")
 
@@ -57,10 +58,12 @@ def setup_once():
         def sentry_build_tracer(name, task, *args, **kwargs):
             # type: (Any, Any, *Any, **Any) -> Any
             if not getattr(task, "_sentry_is_patched", False):
-                # Need to patch both methods because older celery sometimes
-                # short-circuits to task.run if it thinks it's safe.
-                task.__call__ = _wrap_task_call(task, task.__call__)
-                task.run = _wrap_task_call(task, task.run)
+                # determine whether Celery will use __call__ or run and patch
+                # accordingly
+                if task_has_custom(task, "__call__"):
+                    type(task).__call__ = _wrap_task_call(task, type(task).__call__)
+                else:
+                    task.run = _wrap_task_call(task, task.run)
 
                 # `build_tracer` is apparently called for every task
                 # invocation. Can't wrap every celery task for every invocation

tests/integrations/celery/test_celery.py

@@ -407,3 +407,25 @@ def walk_dogs(x, y):
         # passed as args or as kwargs, so make this generic
         DictionaryContaining({"celery_job": dict(task="dog_walk", **args_kwargs)})
     )
+
+
+def test_abstract_task(capture_events, celery, celery_invocation):
+    events = capture_events()
+
+    class AbstractTask(celery.Task):
+        abstract = True
+
+        def __call__(self, *args, **kwargs):
+            try:
+                return self.run(*args, **kwargs)
+            except ZeroDivisionError:
+                return None
+
+    @celery.task(name="dummy_task", base=AbstractTask)
+    def dummy_task(x, y):
+        return x / y
+
+    with start_transaction():
+        celery_invocation(dummy_task, 1, 0)
+
+    assert not events