wait: print a failed task's output

osism/commands/wait.py

@@ -13,6 +13,19 @@
 # continuously, so this only fires on a task that is genuinely wedged.
 DEFAULT_STALL_REPORT_SECONDS = 600
 
+# How many trailing output lines to print for a task that ended in FAILURE.
+# Enough to carry an Ansible ``fatal:`` block and the recap that follows it,
+# without replaying a whole play into the log of a collection run where
+# several roles can fail at once.
+FAILED_TASK_OUTPUT_LINES = 50
+
+# Records a completed task's stream ends with that are not output:
+# ``finish_task_output`` appends one ``rc`` and one ``action: quit``
+# (``osism/utils/__init__.py``). Every Celery task calls
+# ``run_ansible_in_environment`` at most once, so there is at most one such
+# pair and it is always at the very end of the stream.
+STREAM_CONTROL_RECORDS = 2
+
 
 def stall_report_seconds():
     """Seconds of silence before a STARTED task is reported.
@@ -74,6 +87,52 @@ def peek_task_output(redis_conn, task_id, now=None):
     )
 
 
+def tail_task_output(redis_conn, task_id, limit):
+    """Return a task's last ``limit`` output lines without consuming them.
+
+    Same non-destructive read as ``peek_task_output`` -- ``xrevrange``, no
+    ``xdel`` -- but it keeps the lines rather than just the newest one, and
+    reverses them back into emit order, because ``xrevrange`` answers
+    newest first and a play printed backwards is no diagnosis at all.
+
+    Only ``stdout`` records count. A stream also carries the ``rc`` and
+    ``action: quit`` records ``finish_task_output`` appends when the task
+    completes, and this helper -- unlike ``peek_task_output``, which by
+    construction only ever sees a task still in flight -- runs after
+    completion, which is exactly when they exist. Taken as output they
+    would append a bare rc and ``quit`` to the tail, spend two slots of
+    ``limit``, and make a task that failed before writing a single line
+    look like it produced two. Hence the allowlist, and the over-read by
+    ``STREAM_CONTROL_RECORDS`` so filtering them out does not shorten the
+    tail.
+
+    ``lines`` is the number of ``stdout`` records and ``omitted`` how many
+    the tail leaves out. Both are derived here rather than left to the
+    caller, for the same reason ``peek_task_output`` derives
+    ``stalled_for``: everything computed from a Redis reply then sits
+    inside whatever guard wraps the call, and a malformed reply cannot
+    raise past it.
+    """
+    entries = redis_conn.xrevrange(
+        task_id, "+", "-", count=limit + STREAM_CONTROL_RECORDS
+    )
+    if not entries:
+        return SimpleNamespace(lines=0, tail=[], omitted=0)
+
+    # All control records live at the tail of the stream, so the over-read
+    # above sees every one of them and the subtraction is exact rather than
+    # an estimate.
+    control = sum(1 for _, fields in entries if fields.get(b"type") != b"stdout")
+    lines = redis_conn.xlen(task_id) - control
+    tail = [
+        fields.get(b"content", b"").decode().rstrip("\n")
+        for _, fields in reversed(entries)
+        if fields.get(b"type") == b"stdout"
+    ][-limit:]
+
+    return SimpleNamespace(lines=lines, tail=tail, omitted=lines - len(tail))
+
+
 class Run(Command):
     def get_parser(self, prog_name):
         parser = super(Run, self).get_parser(prog_name)
@@ -194,6 +253,48 @@ def _report_stall(self, task_id):
                 f"Last output: {peek.last_line}"
             )
 
+    def _report_failure_output(self, task_id):
+        """Print what a task that ended in FAILURE last emitted.
+
+        ``result.get()`` is not usable here -- Celery re-raises the task's
+        exception from it, which would replace the exit code with a
+        traceback -- so the play output is read from the task's own Redis
+        stream instead. Without this a failed role is as opaque as a hung
+        one: on the collection path nothing ever drains the stream, so the
+        output is intact in Redis and simply never looked at.
+        """
+        if self._peek_disabled:
+            return
+
+        try:
+            tail = tail_task_output(utils.redis, task_id, FAILED_TASK_OUTPUT_LINES)
+        except Exception as exc:
+            # Same contract as the stall peek: the non-``--live`` path
+            # never needed Redis, so a read failure must stay cosmetic --
+            # a task that has already failed cleanly with rc 1 must not
+            # acquire a traceback on top. Report once, then stop trying.
+            logger.warning(
+                f"Cannot read the output stream of failed task {task_id}: {exc}"
+            )
+            self._peek_disabled = True
+            return
+
+        if not tail.lines:
+            logger.error(f"Task {task_id} produced no output before it failed")
+            return
+
+        if tail.omitted:
+            logger.error(
+                f"Last {len(tail.tail)} of {tail.lines} output lines of "
+                f"failed task {task_id} "
+                f"({tail.omitted} earlier lines not shown):"
+            )
+        else:
+            logger.error(f"Output of failed task {task_id} ({tail.lines} lines):")
+
+        for line in tail.tail:
+            print(line)
+
     def take_action(self, parsed_args):
         from celery import Celery
         from celery.result import AsyncResult
@@ -256,7 +357,12 @@ def take_action(self, parsed_args):
                         print(f"{task_id} = {result.state}")
 
                     # Deliberately no result.get() here even with --output:
-                    # Celery re-raises the task's exception from it.
+                    # Celery re-raises the task's exception from it. The
+                    # play output comes from the task's Redis stream
+                    # instead.
+                    if output:
+                        self._report_failure_output(task_id)
+
                     rc = 1
 
                 elif result.state == "STARTED":

tests/unit/commands/test_wait.py

@@ -566,3 +566,216 @@ def test_script_format_prints_failure_state(capsys, loguru_logs):
     assert mocks.rc == 1
     assert capsys.readouterr().out == "taskid1 = FAILURE\n"
     assert not any("taskid1" in record["message"] for record in loguru_logs)
+
+
+# --- a failed task's output ---------------------------------------------------
+
+
+def _stdout(content):
+    """A stream record as ``push_task_output`` writes it."""
+    return {b"type": b"stdout", b"content": content}
+
+
+def _control_pair(rc=b"2"):
+    """The records ``finish_task_output`` appends when a task completes.
+
+    Newest first, matching ``xrevrange`` order: the ``action`` record is
+    written last, so it comes back first.
+    """
+    return [
+        (b"1787674033909-0", {b"type": b"action", b"content": b"quit"}),
+        (b"1787674033908-0", {b"type": b"rc", b"content": rc}),
+    ]
+
+
+def _run_failure_tail(*, entries=(), xlen=0, args=None, redis_error=None):
+    """Drive one FAILURE iteration with a controlled output stream.
+
+    Mirrors ``_run_started_peek``, including the ``utils.redis`` cache
+    eviction: the lazy ``__getattr__`` stores the resolved connection in
+    module globals, so it has to be dropped for the patched factory to be
+    picked up.
+    """
+    cmd = wait.Run(MagicMock(), MagicMock())
+    parsed_args = cmd.get_parser("test").parse_args(args or ["taskid1", "--output"])
+
+    conn = MagicMock()
+    if redis_error is not None:
+        conn.xrevrange.side_effect = redis_error
+    else:
+        conn.xrevrange.return_value = list(entries)
+        conn.xlen.return_value = xlen
+
+    osism_utils.__dict__.pop("redis", None)
+
+    with patch("celery.Celery"), patch(
+        "celery.result.AsyncResult", side_effect=[_make_result("FAILURE")]
+    ), patch("osism.commands.wait.time.sleep"), patch(
+        "osism.utils._init_redis", return_value=conn
+    ):
+        rc = cmd.take_action(parsed_args)
+
+    osism_utils.__dict__.pop("redis", None)
+
+    return SimpleNamespace(rc=rc, conn=conn)
+
+
+def test_tail_returns_the_last_lines_in_emit_order_without_consuming():
+    """``xrevrange`` yields newest first; the tail has to read chronologically.
+
+    Printing a play backwards would be worse than printing nothing. The
+    read must also stay non-destructive, or it steals output from
+    ``--live`` and from the operator.
+    """
+    r = MagicMock()
+    r.xrevrange.return_value = [
+        (b"1787674033907-0", _stdout(b"fatal: [node-0]: FAILED!\n")),
+        (b"1787674033906-0", _stdout(b"TASK [keystone : Bootstrap]\n")),
+    ]
+    r.xlen.return_value = 142
+
+    tail = wait.tail_task_output(r, "taskid1", 2)
+
+    assert tail.lines == 142
+    assert tail.omitted == 140
+    assert tail.tail == ["TASK [keystone : Bootstrap]", "fatal: [node-0]: FAILED!"]
+    r.xdel.assert_not_called()
+    r.xrevrange.assert_called_once_with("taskid1", "+", "-", count=4)
+
+
+def test_failed_task_with_output_prints_what_it_last_emitted(capsys, loguru_logs):
+    """The whole point: a failed nutshell role must name its own error.
+
+    Before this, the FAILURE branch printed the state line and nothing
+    else, so a role that failed was exactly as opaque as one that hung.
+    """
+    mocks = _run_failure_tail(
+        entries=_control_pair()
+        + [
+            (b"1787674033907-0", _stdout(b"fatal: [node-0]: FAILED!\n")),
+            (b"1787674033906-0", _stdout(b"TASK [keystone : Bootstrap]\n")),
+        ],
+        xlen=4,
+    )
+
+    assert mocks.rc == 1
+    assert capsys.readouterr().out == (
+        "TASK [keystone : Bootstrap]\nfatal: [node-0]: FAILED!\n"
+    )
+    assert any("taskid1" in record["message"] for record in loguru_logs)
+
+
+def test_failed_task_output_says_how_much_it_left_out(loguru_logs):
+    """A truncated tail must say so, or it reads as the whole run."""
+    _run_failure_tail(
+        entries=_control_pair() + [(b"1787674033907-0", _stdout(b"last\n"))],
+        xlen=143,
+    )
+
+    assert any(
+        "141" in record["message"] and "taskid1" in record["message"]
+        for record in loguru_logs
+    )
+
+
+def test_failed_task_that_emitted_nothing_is_reported_as_such(loguru_logs):
+    """An empty stream is a diagnosis: the task died before its first line."""
+    _run_failure_tail(entries=[], xlen=0)
+
+    assert any("no output" in record["message"] for record in loguru_logs)
+
+
+def test_failed_task_output_read_failure_never_breaks_wait(loguru_logs):
+    """Redis is not a hard dependency of the non-``--live`` path.
+
+    A failed task already sets rc 1; a broken read must not add a
+    traceback on top of it.
+    """
+    mocks = _run_failure_tail(redis_error=RuntimeError("redis down"))
+
+    assert mocks.rc == 1
+    assert any("redis down" in record["message"] for record in loguru_logs)
+
+
+def test_failed_task_without_output_flag_does_not_read_redis():
+    """``--output`` gates the payload, exactly as it does for SUCCESS."""
+    mocks = _run_failure_tail(args=["taskid1"], entries=[], xlen=0)
+
+    assert mocks.rc == 1
+    mocks.conn.xrevrange.assert_not_called()
+
+
+def test_failed_task_output_survives_a_malformed_redis_reply(loguru_logs):
+    """Everything computed from the reply belongs inside the helper.
+
+    The stream length shapes the truncation notice. Computed in the
+    caller, a reply that reads fine but does not behave like an integer
+    raises *after* the guarded call and turns a reporting feature into a
+    crash on a task that had already failed cleanly with rc 1. Derived
+    inside ``tail_task_output`` -- as ``peek_task_output`` derives
+    ``stalled_for`` -- it cannot escape the guard.
+    """
+    conn = MagicMock()
+    conn.xrevrange.return_value = [(b"1787674033907-0", _stdout(b"last\n"))]
+    conn.xlen.return_value = "not-a-number"
+
+    cmd = wait.Run(MagicMock(), MagicMock())
+    parsed_args = cmd.get_parser("test").parse_args(["taskid1", "--output"])
+
+    osism_utils.__dict__.pop("redis", None)
+    with patch("celery.Celery"), patch(
+        "celery.result.AsyncResult", side_effect=[_make_result("FAILURE")]
+    ), patch("osism.commands.wait.time.sleep"), patch(
+        "osism.utils._init_redis", return_value=conn
+    ):
+        rc = cmd.take_action(parsed_args)
+    osism_utils.__dict__.pop("redis", None)
+
+    assert rc == 1
+    assert any(
+        "Cannot read the output stream" in record["message"] for record in loguru_logs
+    )
+
+
+def test_tail_excludes_the_completion_control_records():
+    """`finish_task_output` appends `rc` and `action: quit` before the task
+    raises, so a completed task's stream ends with records that are not
+    output. Printing them appends bare `2` and `quit` to the play tail, and
+    they eat two slots of the line budget."""
+    r = MagicMock()
+    r.xrevrange.return_value = _control_pair() + [
+        (b"1787674033907-0", _stdout(b"fatal: [node-0]: FAILED!\n")),
+        (b"1787674033906-0", _stdout(b"TASK [keystone : Bootstrap]\n")),
+    ]
+    r.xlen.return_value = 4
+
+    tail = wait.tail_task_output(r, "taskid1", 50)
+
+    assert tail.tail == ["TASK [keystone : Bootstrap]", "fatal: [node-0]: FAILED!"]
+    assert tail.lines == 2
+    assert tail.omitted == 0
+
+
+def test_tail_reads_past_the_control_records_to_fill_the_budget():
+    """The over-read has to cover them, or the caller silently gets
+    `limit - 2` lines whenever the task completed."""
+    r = MagicMock()
+    r.xrevrange.return_value = _control_pair() + [
+        (f"178767403390{i}-0".encode(), _stdout(f"line{i}\n".encode()))
+        for i in range(3)
+    ]
+    r.xlen.return_value = 5
+
+    wait.tail_task_output(r, "taskid1", 3)
+
+    r.xrevrange.assert_called_once_with("taskid1", "+", "-", count=5)
+
+
+def test_failed_task_with_only_control_records_reports_no_output(loguru_logs):
+    """A task that failed before writing a line still has two records, so a
+    raw stream length reads as output and the no-output diagnosis -- the
+    one thing distinguishing died-mid-play from died-before-first-line --
+    never fires."""
+    _run_failure_tail(entries=_control_pair(), xlen=2)
+
+    assert any("no output" in record["message"] for record in loguru_logs)