Enhancement: Add feature to manually retry failed task

flower/api/tasks.py

@@ -636,3 +636,74 @@ def get(self, taskid):
             response['worker'] = task.worker.hostname
 
         self.write(response)
+
+
+class TaskReapply(BaseTaskHandler):
+    @web.authenticated
+    async def post(self, taskid):
+        """
+Reapply a task with its original arguments
+
+**Example request**:
+
+.. sourcecode:: http
+
+  POST /api/task/reapply/91396550-c228-4111-9da4-9d88cfd5ddc6 HTTP/1.1
+  Content-Length: 0
+  Host: localhost:5555
+
+**Example response**:
+
+.. sourcecode:: http
+
+  HTTP/1.1 200 OK
+  Content-Length: 51
+  Content-Type: application/json; charset=UTF-8
+
+  {
+      "task-id": "aef8138a-4b2d-42fa-9c34-4ea23b1ffa9e"
+  }
+
+:reqheader Authorization: optional OAuth token to authenticate
+:statuscode 200: no error
+:statuscode 400: task has no name or its original arguments cannot be restored
+:statuscode 401: unauthorized request
+:statuscode 404: unknown task
+:statuscode 500: failed to reapply the task
+        """
+        task = tasks.get_task_by_id(self.application.events, taskid)
+        if not task:
+            raise HTTPError(404, f"Unknown task '{taskid}'")
+
+        taskname = task.name
+        if not taskname:
+            raise HTTPError(400, "Cannot reapply task with no name")
+
+        try:
+            args = tasks.make_json_serializable(tasks.parse_args(task.args))
+            kwargs = tasks.make_json_serializable(tasks.parse_kwargs(task.kwargs))
+        except (ValueError, TypeError) as exc:
+            logger.error("Cannot reapply task '%s': %s", taskid, exc)
+            raise HTTPError(400, f"Invalid task arguments: {exc}") from exc
+
+        # Preserve the original routing when the event carries it, so the
+        # retry runs on the same queue as the original task.
+        options = {}
+        if getattr(task, 'routing_key', None):
+            options['routing_key'] = task.routing_key
+        if getattr(task, 'exchange', None):
+            options['exchange'] = task.exchange
+
+        try:
+            # send_task works even when the task module is not importable
+            # by flower (same approach as TaskSendTask above)
+            result = self.capp.send_task(taskname, args=args, kwargs=kwargs, **options)
+        except Exception as exc:  # broker and serializer failures are heterogeneous
+            logger.exception("Error reapplying task '%s' with args=%s, kwargs=%s",
+                             taskid, args, kwargs)
+            raise HTTPError(500, f"Error reapplying task: {exc}") from exc
+
+        response = {'task-id': result.task_id}
+        if self.backend_configured(result):
+            response.update(state=result.state)
+        self.write(response)

flower/static/js/flower.js

@@ -690,4 +690,45 @@ var flower = (function () {
 
     });
 
+    $('#task-retry').click(function () {
+        const $button = $(this);
+        const $spinner = $button.find('.spinner-border');
+        const taskId = $('#taskid').text();
+
+        if (!taskId) {
+            show_alert('Task ID is missing. Cannot proceed.', 'danger');
+            return;
+        }
+
+        function resetButton() {
+            $button.prop('disabled', false);
+            $spinner.addClass('d-none');
+        }
+
+        // Show loading state while the task is reapplied
+        $button.prop('disabled', true);
+        $spinner.removeClass('d-none');
+
+        $.ajax({
+            type: 'POST',
+            url: url_prefix() + '/api/task/reapply/' + taskId,
+            dataType: 'json',
+            success: function (response) {
+                const newTaskId = response && response['task-id'];
+                if (!newTaskId) {
+                    show_alert('Task retry did not return a new task ID.', 'danger');
+                    resetButton();
+                    return;
+                }
+                show_alert(`Task ${taskId} has been retried (new task ID: ${newTaskId})`, 'success');
+                setTimeout(() => location.reload(), 1500);
+            },
+            error: function (data) {
+                console.error('Failed to retry task', taskId, data.status, data.responseText);
+                show_alert('Failed to retry task. Check the flower logs for details.', 'danger');
+                resetButton();
+            }
+        });
+    });
+
 }(jQuery));

flower/templates/task.html

@@ -16,6 +16,11 @@ <h2>{{ getattr(task, 'name', None) }}
                 <button class="btn btn-danger float-end" id="task-terminate">Terminate</button>
             {% elif task.state == "RECEIVED" or task.state == "RETRY" %}
                 <button class="btn btn-danger float-end" id="task-revoke">Revoke</button>
+            {% elif task.state == "FAILURE" %}
+                <button class="btn btn-warning float-end" id="task-retry">
+                    <span class="spinner-border spinner-border-sm d-none" role="status" aria-hidden="true"></span>
+                    Retry
+                </button>
             {% end %}
           </h2>
         </div>

flower/urls.py

@@ -51,6 +51,7 @@
     (r"/api/task/timeout/(.+)", control.TaskTimout),
     (r"/api/task/rate-limit/(.+)", control.TaskRateLimit),
     (r"/api/task/revoke/(.+)", control.TaskRevoke),
+    (r"/api/task/reapply/(.+)", tasks.TaskReapply),
     # Metrics
     (r"/metrics", monitor.Metrics),
     (r"/healthcheck", monitor.Healthcheck),

flower/utils/tasks.py

@@ -1,4 +1,6 @@
+import ast
 import datetime
+import json
 import time
 
 from .search import parse_search_terms, satisfies_search_terms
@@ -68,3 +70,118 @@ def get_task_by_id(events, task_id):
 
 def as_dict(task):
     return task.as_dict()
+
+
+# Upper bound on the size of a stored args/kwargs string we are willing to
+# parse. Celery truncates event args far below this; anything larger is not
+# a legitimate task representation and is rejected before parsing.
+MAX_ARG_LENGTH = 65536
+
+
+def _parse_literal(value, kind):
+    """
+    Parse a string holding a JSON value or a Python literal (the two formats
+    celery events use for task args/kwargs) into a Python value.
+
+    Only ``json.loads`` and ``ast.literal_eval`` are used, so no code is ever
+    executed. Raises ValueError for anything that is not a plain literal,
+    including truncated reprs such as ``'(1, 2, ...'``.
+    """
+    if len(value) > MAX_ARG_LENGTH:
+        raise ValueError(f"Task {kind} too long to parse ({len(value)} bytes)")
+    try:
+        return json.loads(value)
+    except json.JSONDecodeError:
+        pass
+    except (MemoryError, RecursionError) as exc:
+        raise ValueError(f"Could not parse task {kind}: {value!r}") from exc
+    try:
+        return ast.literal_eval(value)
+    except (ValueError, TypeError, SyntaxError, MemoryError, RecursionError) as exc:
+        raise ValueError(f"Could not parse task {kind}: {value!r}") from exc
+
+
+def _reject_truncated(obj, kind):
+    """
+    Reject values truncated by celery.
+
+    Celery stores event args/kwargs via saferepr, which truncates long
+    strings by cutting them and appending '...' inside the quotes, so a
+    truncated repr like "(7, 'Bearer...')" still parses cleanly. Refuse
+    such values rather than reapply a task with corrupted arguments.
+    A legitimate string ending in '...' is indistinguishable from a
+    truncated one, so it is rejected too.
+    """
+    if isinstance(obj, str):
+        if obj.endswith('...'):
+            raise ValueError(f"Task {kind} appear truncated by celery: {obj!r}")
+    elif isinstance(obj, (list, tuple, set)):
+        for item in obj:
+            _reject_truncated(item, kind)
+    elif isinstance(obj, dict):
+        for key, value in obj.items():
+            _reject_truncated(key, kind)
+            _reject_truncated(value, kind)
+
+
+def parse_args(args):
+    """
+    Parse the string representation of a task's positional arguments
+    into a list.
+
+    Raises ValueError if the string cannot be restored to the original
+    arguments (e.g. it was truncated by celery), so callers never reapply
+    a task with wrong arguments.
+    """
+    if not args:
+        return []
+    parsed = _parse_literal(args, 'args')
+    if isinstance(parsed, tuple):
+        parsed = list(parsed)
+    if not isinstance(parsed, list):
+        raise ValueError(f"Task args must be a list or tuple: {args!r}")
+    _reject_truncated(parsed, 'args')
+    return parsed
+
+
+def parse_kwargs(kwargs):
+    """
+    Parse the string representation of a task's keyword arguments
+    into a dict.
+
+    Raises ValueError if the string cannot be restored to the original
+    keyword arguments.
+    """
+    if not kwargs:
+        return {}
+    parsed = _parse_literal(kwargs, 'kwargs')
+    if not isinstance(parsed, dict):
+        raise ValueError(f"Task kwargs must be a dict: {kwargs!r}")
+    _reject_truncated(parsed, 'kwargs')
+    return parsed
+
+
+def make_json_serializable(obj):
+    """
+    Recursively convert parsed argument values to JSON-serializable types.
+
+    Raises TypeError for values with no faithful JSON equivalent (sets,
+    bytes, non-string dict keys, ...) so callers fail loudly instead of
+    reapplying a task with arguments of the wrong type. Tuples become
+    lists because that is what the original task received from the JSON
+    serializer anyway.
+    """
+    if obj is None or isinstance(obj, (str, int, float, bool)):
+        return obj
+    if isinstance(obj, (list, tuple)):
+        return [make_json_serializable(item) for item in obj]
+    if isinstance(obj, dict):
+        result = {}
+        for key, value in obj.items():
+            if not isinstance(key, str):
+                raise TypeError(f"Dict key {key!r} is not JSON serializable")
+            result[key] = make_json_serializable(value)
+        return result
+    if isinstance(obj, (datetime.datetime, datetime.date)):
+        return obj.isoformat()
+    raise TypeError(f"Value of type {type(obj).__name__} is not JSON serializable: {obj!r}")