flush after prompting and individual loop items

ci_complete

changelogs/fragments/flush-stdout-midtask.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - Fix displaying prompts and individual loop item results with unbuffered output (https://github.com/ansible/ansible/issues/81155, https://github.com/ansible/ansible/issues/79662).

lib/ansible/executor/task_queue_manager.py

@@ -362,8 +362,8 @@ class TaskQueueManager:
         self.terminate()
         self._final_q.close()
         self._cleanup_processes()
-        # We no longer flush on every write in ``Display.display``
-        # just ensure we've flushed during cleanup
+        # We no longer flush on every write in ``Display.display`` by default.
+        # Ensure we've flushed during cleanup
         sys.stdout.flush()
         sys.stderr.flush()
 

lib/ansible/plugins/callback/default.py

@@ -259,7 +259,7 @@ class CallbackModule(CallbackBase):
         self._clean_results(result._result, result._task.action)
         if self._run_is_verbose(result):
             msg += " => %s" % self._dump_results(result._result)
-        self._display.display(msg, color=color)
+        self._display.display(msg, color=color, flush=True)
 
     def v2_runner_item_on_failed(self, result):
         if self._last_task_banner != result._task._uuid:
@@ -274,7 +274,8 @@ class CallbackModule(CallbackBase):
         self._display.display(
             msg + " (item=%s) => %s" % (self._get_item_label(result._result), self._dump_results(result._result)),
             color=C.COLOR_ERROR,
-            stderr=self.get_option('display_failed_stderr')
+            stderr=self.get_option('display_failed_stderr'),
+            flush=True,
         )
 
     def v2_runner_item_on_skipped(self, result):
@@ -286,7 +287,7 @@ class CallbackModule(CallbackBase):
             msg = "skipping: [%s] => (item=%s) " % (result._host.get_name(), self._get_item_label(result._result))
             if self._run_is_verbose(result):
                 msg += " => %s" % self._dump_results(result._result)
-            self._display.display(msg, color=C.COLOR_SKIP)
+            self._display.display(msg, color=C.COLOR_SKIP, flush=True)
 
     def v2_playbook_on_include(self, included_file):
         msg = 'included: %s for %s' % (included_file._filename, ", ".join([h.name for h in included_file._hosts]))

lib/ansible/utils/display.py

@@ -28,6 +28,7 @@ else:
 import collections.abc as c
 import codecs
 import ctypes.util
+import errno
 import fcntl
 import getpass
 import io
@@ -398,6 +399,7 @@ class Display(metaclass=Singleton):
         screen_only: bool = False,
         log_only: bool = False,
         newline: bool = True,
+        flush: bool = False,
     ) -> None:
         """ Display a message to the user
 
@@ -433,18 +435,17 @@ class Display(metaclass=Singleton):
             with self._lock:
                 fileobj.write(msg2)
 
-            # With locks, and the fact that we aren't printing from forks
-            # just write, and let the system flush. Everything should come out peachy
-            # I've left this code for historical purposes, or in case we need to add this
-            # back at a later date. For now ``TaskQueueManager.cleanup`` will perform a
-            # final flush at shutdown.
-            # try:
-            #     fileobj.flush()
-            # except IOError as e:
-            #     # Ignore EPIPE in case fileobj has been prematurely closed, eg.
-            #     # when piping to "head -n1"
-            #     if e.errno != errno.EPIPE:
-            #         raise
+                # With locks, and the fact that we aren't printing from forks
+                # just write, and let the system flush by default. Everything should come out peachy.
+                # ``TaskQueueManager.cleanup`` will perform a final flush at shutdown.
+                if flush:
+                    try:
+                        fileobj.flush()
+                    except IOError as e:
+                        # Ignore EPIPE in case fileobj has been prematurely closed, eg.
+                        # when piping to "head -n1"
+                        if e.errno != errno.EPIPE:
+                            raise
 
         if logger and not screen_only:
             self._log(nocolor, color)
@@ -759,7 +760,7 @@ class Display(metaclass=Singleton):
         #         # can't catch in the results_thread_main daemon thread
         #         raise AnsiblePromptInterrupt('user interrupt')
 
-        self.display(msg)
+        self.display(msg, flush=True)
         result = b''
         with self._lock:
             original_stdin_settings = termios.tcgetattr(self._stdin_fd)

test/integration/targets/pause/test-pause.py

@@ -312,3 +312,26 @@ pause_test.expect(r"\(ctrl\+C then 'C' = continue early, ctrl\+C then 'A' = abor
 pause_test.send('\r')
 pause_test.expect(pexpect.EOF)
 pause_test.close()
+
+# Test the prompt is flushed before cleanup with unbuffered output
+pause_test = pexpect.spawn("bash", env=os.environ, timeout=2)
+pause_test.logfile = log_buffer
+pause_test.sendline('ansible localhost -m pause -a \'prompt="Type the first letter of the alphabet", echo=false\' | cat -u')
+pause_test.expect(r'\[pause\]')
+pause_test.expect(r'"Type the first letter of the alphabet", \(output is hidden\):')
+pause_test.sendline(r'a')
+for expected in [
+    r'localhost | SUCCESS => {',
+    r'    "changed": false,',
+    r'    "delta": 0,',
+    r'    "echo": false,',
+    r'    "rc": 0,',
+    r'    "start": "[^"]*",',
+    r'    "stderr": "[^"]*",',
+    r'    "stdout": "Paused for [^ ]* minutes",',
+    r'    "stop": "[^"]*",',
+    r'    "user_input": "a"',
+    r'}',
+]:
+    pause_test.expect(expected)
+pause_test.close()