zuul-stream: automatically remove streaming files

When using protocol version 1, send a finalise message when streaming
is complete so that the zuul_console daemon can delete the temporary
file.

We test this by inspecting the Ansible console output, which logs a
message with the UUID of the streaming job.  We dump the temporary
files on the remote side and make sure a console file for that job
isn't present.

Change-Id: I823156dc2bcae91bd6d9770bd1520aa55ad875b4

playbooks/zuul-stream/fixtures/test-stream.yaml

@@ -64,3 +64,15 @@
 
     - name: Print binary data
       command: echo -e '\x80abc'
+
+    - name: Find any console log files
+      find:
+        paths: /tmp
+        patterns: 'console-*.log'
+      register: _tmp_files
+
+    # We check this list in zuul-stream/functional.yaml to make sure
+    # we're cleaning up console log files.
+    - name: Dump tmp files
+      debug:
+        var: _tmp_files

playbooks/zuul-stream/functional.yaml

@@ -21,6 +21,11 @@
         PYTHONPATH: "{{ python_path }}"
       register: _success_output
 
+    - name: Save raw output to file
+      copy:
+        content: '{{ _success_output.stdout }}'
+        dest: 'console-job-output-success-19887.txt'
+
     - name: Save output
       shell: |
           mv job-output.txt job-output-success-19887.txt
@@ -31,6 +36,20 @@
         that:
           - "'[node1] Reports streaming version: 1' in _success_output.stdout"
 
+    # Streamer puts out a line like
+    #  [node1] Starting to log 916b2084-4bbb-80e5-248e-000000000016-1-node1 for task TASK: Print binary data
+    # One of the tasks in job-output shows find: results;
+    # the console file for this task should not be there.
+    - name: Validate temporary files removed
+      shell: |
+        for f in $(grep 'Starting to log' console-job-output-success-19887.txt | awk '{print $5}'); do
+          echo "Checking ${f}"
+          if grep -q '"path": "/tmp/console-'${f}'.log"' job-output-success-19887.txt; then
+            echo "*** /tmp/${f}.log still exists"
+            exit 1
+          fi
+        done
+
     # NOTE(ianw) 2022-07 : we deliberatly have this second step to run
     # against the console setup by the infrastructure executor in the
     # job pre playbooks as a backwards compatability sanity check.
@@ -43,6 +62,12 @@
         ZUUL_JOB_LOG_CONFIG: "{{ ansible_user_dir}}/logging.json"
         ZUUL_JOBDIR: "{{ ansible_user_dir}}"
         PYTHONPATH: "{{ python_path }}"
+      register: _success_output
+
+    - name: Save raw output to file
+      copy:
+        content: '{{ _success_output.stdout }}'
+        dest: 'console-job-output-success-19885.txt'
 
     - name: Save output
       shell: |

playbooks/zuul-stream/post.yaml

@@ -18,6 +18,7 @@
       with_items:
         - logging.json
         - ansible.cfg
+        - console-job-output-success-19887.txt
         - job-output-success-19887.txt
         - job-output-success-19887.json
         - job-output-success-19885.txt

zuul/ansible/base/callback/zuul_stream.py

@@ -211,6 +211,16 @@ class CallbackModule(default.CallbackModule):
                 done = self._log_streamline(
                     host, line.decode("utf-8", "backslashreplace"))
                 if done:
+                    if self._zuul_console_version > 0:
+                        try:
+                            # reestablish connection and tell console to
+                            # clean up
+                            s = self._read_log_connect(host, ip, port)
+                            s.send(f'f:{log_id}\n'.encode('utf-8'))
+                            s.close()
+                        except Exception:
+                            # Don't worry if this fails
+                            pass
                     return
             else:
                 more = s.recv(4096)

zuul/ansible/base/library/zuul_console.py

@@ -174,11 +174,17 @@ class Server(object):
             except UnicodeDecodeError:
                 pass
 
+    def _clean_uuid(self, log_uuid):
+        # use path split to make use the input isn't trying to be clever
+        # and construct some path like /tmp/console-/../../something
+        return os.path.split(log_uuid)[-1]
+
     def handleOneConnection(self, conn):
         # V1 protocol
         # -----------
         #  v:<ver>    get version number, <ver> is remote version
         #  s:<uuid>   send logs for <uuid>
+        #  f:<uuid>   finalise/cleanup <uuid>
         while True:
             command = self.get_command(conn)
             if command.startswith('v:'):
@@ -187,8 +193,16 @@ class Server(object):
                 # for now.
                 conn.send(f'{ZUUL_CONSOLE_PROTO_VERSION}\n'.encode('utf-8'))
                 continue
+            elif command.startswith('f:'):
+                log_uuid = self._clean_uuid(command[2:])
+                try:
+                    os.unlink(self.path.format(log_uuid=log_uuid))
+                except Exception:
+                    # something might have cleaned /tmp
+                    pass
+                continue
             elif command.startswith('s:'):
-                log_uuid = command[2:]
+                log_uuid = self._clean_uuid(command[2:])
                 break
             else:
                 # NOTE(ianw): 2022-07-21 In releases < 6.3.0 the streaming
@@ -196,13 +210,9 @@ class Server(object):
                 # default assume that is what is coming in here.  We can
                 # remove this fallback when we decide it is no longer
                 # necessary.
-                log_uuid = command
+                log_uuid = self._clean_uuid(command)
                 break
 
-        # use path split to make use the input isn't trying to be clever
-        # and construct some path like /tmp/console-/../../something
-        log_uuid = os.path.split(log_uuid)[-1]
-
         # FIXME: this won't notice disconnects until it tries to send
         console = None
         try: