add more logging on celery task retry (#3695)

# What this PR does

This is a follow up to https://github.com/grafana/oncall/pull/3677.

It appears that when a task uses the [`autoretry_for`
kwarg](https://docs.celeryq.dev/en/stable/userguide/tasks.html#automatic-retry-for-known-exceptions)
in the task decorator, it doesn't log the exception in `on_failure` as
would be expected. Now when retrying, we log out a message + any
exception/stack trace information.

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [x] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)

engine/common/custom_celery_tasks/dedicated_queue_retry_task.py

@@ -1,9 +1,12 @@
 from celery import shared_task
+from celery.utils.log import get_task_logger
 
 from common.custom_celery_tasks.log_exception_on_failure_task import LogExceptionOnFailureTask
 
 RETRY_QUEUE = "retry"
 
+logger = logger = get_task_logger(__name__)
+
 
 class DedicatedQueueRetryTask(LogExceptionOnFailureTask):
     """
@@ -14,6 +17,8 @@ class DedicatedQueueRetryTask(LogExceptionOnFailureTask):
     def retry(
         self, args=None, kwargs=None, exc=None, throw=True, eta=None, countdown=None, max_retries=None, **options
     ):
+        logger.warn("Retrying celery task", exc_info=exc)
+
         # Just call retry with queue argument
         return super().retry(
             args=args,

engine/common/custom_celery_tasks/tests/test_dedicated_queue_retry_task.py

@@ -0,0 +1,19 @@
+import pytest
+
+from common.custom_celery_tasks.dedicated_queue_retry_task import shared_dedicated_queue_retry_task
+
+EXCEPTION_MSG = "my exception"
+
+
+def test_dedicated_queue_retry_task_logs_stack_trace_on_task_retry_when_using_autoretry_for(caplog):
+    @shared_dedicated_queue_retry_task(autoretry_for=(Exception,), retry_backoff=False, max_retries=1)
+    def my_task():
+        raise ValueError(EXCEPTION_MSG)
+
+    with pytest.raises(ValueError):
+        my_task.apply().get()
+
+    assert "Traceback (most recent call last):" in caplog.text
+    assert "Retrying celery task" in caplog.text
+    assert f"raise ValueError(EXCEPTION_MSG)\nValueError: {EXCEPTION_MSG}" in caplog.text
+    caplog.clear()

engine/common/tests/test_task_queue_assignment.py

@@ -9,6 +9,7 @@ we should avoid @shared_dedicated_queue_retry_task or @shared_task and
 remove entirely if it is not needed.
 """
 COMMON_IGNORED_TASKS = {
+    "common.custom_celery_tasks.tests.test_dedicated_queue_retry_task.my_task",
     "common.custom_celery_tasks.tests.test_log_exception_on_failure_task.my_task",
     "common.custom_celery_tasks.tests.test_log_exception_on_failure_task.my_task_two",
 }