Fix debug logs failing with persistent connection (#33049)

* Fix debug logs failing with persistent connection

Fixes #33047

*  As debug logs are written on stdout, it interrupts
   the communication between ansible-connection(background)
   process and main process. To avoid this add a string similar
   to exactly identify the response string.

*  Remove unwanted code in ansible-connection

*  Fix review comments

* Fix spurious log emitted on ansible-connection stdout issue

*  ansible-connection which runs as a background process sends a
   json string (contains response received from remote device)
   to foreground ansible-playbook process over stdout.

*  If in case debug flag is enabled the connection_loader api
   invoked from ansible-connection `ssh = connection_loader.get('ssh', class_only=True)`
   results in emitting debug logs on stdout. This  spurious log
   interfere with the actual response and results in failure while
   reading json string in ansible-playbook process

* To avoid this save stdout of ansible-connection and redirect it string
  buffer to accumulate all the logs emitted by core API's

* Add these logs in `result['messages']` which is send a json string after reinstating saved stdout

*  Remove unwanted code in ansible-connection

* Fix review comment

bin/ansible-connection

@@ -13,11 +13,9 @@ except Exception:
 
 import fcntl
 import os
-import shlex
 import signal
 import socket
 import sys
-import time
 import traceback
 import errno
 import json
@@ -25,7 +23,7 @@ import json
 from ansible import constants as C
 from ansible.module_utils._text import to_bytes, to_native, to_text
 from ansible.module_utils.six import PY3
-from ansible.module_utils.six.moves import cPickle
+from ansible.module_utils.six.moves import cPickle, StringIO
 from ansible.module_utils.connection import Connection, ConnectionError, send_data, recv_data
 from ansible.module_utils.service import fork_process
 from ansible.playbook.play_context import PlayContext
@@ -154,54 +152,6 @@ class ConnectionProcess(object):
                     setattr(self.connection, '_connected', False)
         display.display('shutdown complete', log_only=True)
 
-    def do_EXEC(self, data):
-        cmd = data.split(b'EXEC: ')[1]
-        return self.connection.exec_command(cmd)
-
-    def do_PUT(self, data):
-        (op, src, dst) = shlex.split(to_native(data))
-        return self.connection.fetch_file(src, dst)
-
-    def do_FETCH(self, data):
-        (op, src, dst) = shlex.split(to_native(data))
-        return self.connection.put_file(src, dst)
-
-    def do_CONTEXT(self, data):
-        pc_data = data.split(b'CONTEXT: ', 1)[1]
-
-        if PY3:
-            pc_data = cPickle.loads(pc_data, encoding='bytes')
-        else:
-            pc_data = cPickle.loads(pc_data)
-
-        pc = PlayContext()
-        pc.deserialize(pc_data)
-
-        try:
-            self.connection.update_play_context(pc)
-        except AttributeError:
-            pass
-
-        return (0, 'ok', '')
-
-    def do_RUN(self, data):
-        timeout = self.play_context.timeout
-        while bool(timeout):
-            if os.path.exists(self.socket_path):
-                break
-            time.sleep(1)
-            timeout -= 1
-        socket_bytes = to_bytes(self.socket_path, errors='surrogate_or_strict')
-        return 0, b'\n#SOCKET_PATH#: %s\n' % socket_bytes, ''
-
-
-def communicate(sock, data):
-    send_data(sock, data)
-    rc = int(recv_data(sock), 10)
-    stdout = recv_data(sock)
-    stderr = recv_data(sock)
-    return (rc, stdout, stderr)
-
 
 def main():
     """ Called to initiate the connect to the remote device
@@ -217,6 +167,10 @@ def main():
     else:
         stdin = sys.stdin
 
+    # Note: update the below log capture code after Display.display() is refactored.
+    saved_stdout = sys.stdout
+    sys.stdout = StringIO()
+
     try:
         # read the play context data via stdin, which means depickling it
         cur_line = stdin.readline()
@@ -308,11 +262,13 @@ def main():
                         'exception': traceback.format_exc()
                     })
 
+    messages.append(sys.stdout.getvalue())
     result.update({
         'messages': messages,
         'socket_path': socket_path
     })
 
+    sys.stdout = saved_stdout
     if 'exception' in result:
         rc = 1
         sys.stderr.write(json.dumps(result))

lib/ansible/executor/task_executor.py

@@ -833,6 +833,7 @@ class TaskExecutor:
         stdin.write(src)
 
         stdin.write(b'\n#END_INIT#\n')
+        stdin.flush()
 
         (stdout, stderr) = p.communicate()
         stdin.close()
@@ -849,7 +850,7 @@ class TaskExecutor:
         if 'error' in result:
             if self._play_context.verbosity > 2:
                 msg = "The full traceback is:\n" + result['exception']
-                display.display(result['exception'], color=C.COLOR_ERROR)
+                display.display(msg, color=C.COLOR_ERROR)
             raise AnsibleError(result['error'])
 
         return result['socket_path']

lib/ansible/plugins/connection/persistent.py

@@ -88,6 +88,7 @@ class Connection(ConnectionBase):
         stdin.write(src)
 
         stdin.write(b'\n#END_INIT#\n')
+        stdin.flush()
 
         (stdout, stderr) = p.communicate()
         stdin.close()
@@ -104,7 +105,7 @@ class Connection(ConnectionBase):
         if 'error' in result:
             if self._play_context.verbosity > 2:
                 msg = "The full traceback is:\n" + result['exception']
-                display.display(result['exception'], color=C.COLOR_ERROR)
+                display.display(msg, color=C.COLOR_ERROR)
             raise AnsibleError(result['error'])
 
         return result['socket_path']

lib/ansible/utils/display.py

@@ -131,6 +131,8 @@ class Display:
                 # characters that are invalid in the user's locale
                 msg2 = to_text(msg2, self._output_encoding(stderr=stderr), errors='replace')
 
+            # Note: After Display() class is refactored need to update the log capture
+            # code in 'bin/ansible-connection' (and other relevant places).
             if not stderr:
                 fileobj = sys.stdout
             else: