Move zuul_log_id injection to command action plugin

The log streaming callback is not being called in the same way
in Ansible 2.5 as it was in 2.3.  In particular, in some cases
different Task objects are used for different hosts.  This,
combined with the fact that the callback is only called once for
a given task means that in these cases we are unable to supply
the zuul_log_id to the Task object for the second host on a task.

This can be resolved by injecting the zuul_log_id within the command
action plugin based on the task uuid directly.

Change-Id: I7ff35263c52d93aeabe915532230964994c30850

playbooks/zuul-stream/templates/ansible.cfg.j2

@@ -3,7 +3,7 @@ inventory = {{ ansible_user_dir }}/inventory.yaml
 gathering = smart
 gather_subset = !all
 lookup_plugins = {{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/lookup
-action_plugins = {{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/action
+action_plugins = {{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/action-general:{{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/action
 callback_plugins = {{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/callback:{{ ansible_user_dir }}/src/git.openstack.org/openstack/ara/ara/plugins/callbacks
 module_utils = {{ ansible_user_dir }}/src/git.openstack.org/openstack-infra/zuul/zuul/ansible/module_utils
 stdout_callback = zuul_stream

tests/fixtures/config/remote-zuul-stream/git/org_project/roles/include-a-role/tasks/main.yaml

@@ -5,3 +5,6 @@
 
 - name: Include role shell task
   shell: echo "This is a shell task after an included role"
+
+- name: Include role command task
+  command: echo "This is a command task after an included role"

tests/remote/test_remote_zuul_stream.py

@@ -29,6 +29,9 @@ class TestZuulStream(AnsibleZuulTestCase):
         ansible_remote = os.environ.get('ZUUL_REMOTE_IPV4')
         self.assertIsNotNone(ansible_remote)
 
+        # on some systems this test may run longer than 30 seconds
+        self.wait_timeout = 60
+
     def _run_job(self, job_name):
         # Keep the jobdir around so we can inspect contents if an
         # assert fails. It will be cleaned up anyway as it is contained
@@ -96,9 +99,13 @@ class TestZuulStream(AnsibleZuulTestCase):
             self.assertLogLine(
                 'controller \| ok: Runtime: \d:\d\d:\d\d\.\d\d\d\d\d\d', text)
             self.assertLogLine('TASK \[Show contents of second file\]', text)
+            self.assertLogLine('compute1 \| command test two', text)
             self.assertLogLine('controller \| command test two', text)
+            self.assertLogLine('compute1 \| This is a rescue task', text)
             self.assertLogLine('controller \| This is a rescue task', text)
+            self.assertLogLine('compute1 \| This is an always task', text)
             self.assertLogLine('controller \| This is an always task', text)
+            self.assertLogLine('compute1 \| This is a handler', text)
             self.assertLogLine('controller \| This is a handler', text)
             self.assertLogLine('controller \| First free task', text)
             self.assertLogLine('controller \| Second free task', text)
@@ -106,6 +113,10 @@ class TestZuulStream(AnsibleZuulTestCase):
                                'included role', text)
             self.assertLogLine('compute1 \| This is a shell task after an '
                                'included role', text)
+            self.assertLogLine('controller \| This is a command task after an '
+                               'included role', text)
+            self.assertLogLine('compute1 \| This is a command task after an '
+                               'included role', text)
             self.assertLogLine(
                 'controller \| ok: Runtime: \d:\d\d:\d\d\.\d\d\d\d\d\d', text)
             self.assertLogLine('PLAY RECAP', text)

zuul/ansible/action-general/command.py

@@ -0,0 +1,29 @@
+# Copyright 2018 BMW Car IT GmbH
+#
+# This module is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This software is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this software.  If not, see <http://www.gnu.org/licenses/>.
+
+
+from zuul.ansible import paths
+command = paths._import_ansible_action_plugin("command")
+
+
+class ActionModule(command.ActionModule):
+
+    def run(self, tmp=None, task_vars=None):
+        # we need the zuul_log_id on shell and command tasks
+        host = paths._sanitize_filename(task_vars.get('inventory_hostname'))
+        if self._task.action in ('command', 'shell'):
+            self._task.args['zuul_log_id'] = "%s-%s" % (self._task._uuid, host)
+
+        return super(ActionModule, self).run(tmp, task_vars)

zuul/ansible/callback/zuul_stream.py

@@ -27,9 +27,9 @@ import os
 import socket
 import threading
 import time
-import uuid
 
 from ansible.plugins.callback import default
+from zuul.ansible import paths
 
 from zuul.ansible import logconfig
 
@@ -214,8 +214,6 @@ class CallbackModule(default.CallbackModule):
             task_name = task.get_name().strip()
 
         if task.action in ('command', 'shell'):
-            log_id = uuid.uuid4().hex
-            task.args['zuul_log_id'] = log_id
             play_vars = self._play._variable_manager._hostvars
 
             hosts = self._get_task_hosts(task)
@@ -229,6 +227,8 @@ class CallbackModule(default.CallbackModule):
                 if ip in ('localhost', '127.0.0.1'):
                     # Don't try to stream from localhost
                     continue
+                log_id = "%s-%s" % (
+                    task._uuid, paths._sanitize_filename(host))
                 streamer = threading.Thread(
                     target=self._read_log, args=(
                         host, ip, log_id, task_name, hosts))

zuul/ansible/paths.py

@@ -188,3 +188,7 @@ def _is_localhost_task(task):
         or task._task.delegate_to == 'localhost'):
         return True
     return False
+
+
+def _sanitize_filename(name):
+    return ''.join(c for c in name if c.isalnum())

zuul/executor/server.py

@@ -1409,12 +1409,18 @@ class AnsibleJob(object):
             # bump the timeout because busy nodes may take more than
             # 10s to respond
             config.write('timeout = 30\n')
+
+            # We need at least the general action dir as this overwrites the
+            # command action plugin for log streaming.
+            action_dirs = [self.executor_server.action_dir_general]
             if not trusted:
-                config.write('action_plugins = %s\n'
-                             % self.executor_server.action_dir)
+                action_dirs.append(self.executor_server.action_dir)
                 config.write('lookup_plugins = %s\n'
                              % self.executor_server.lookup_dir)
 
+            config.write('action_plugins = %s\n'
+                         % ':'.join(action_dirs))
+
             if jobdir_playbook.roles_path:
                 config.write('roles_path = %s\n' % ':'.join(
                     jobdir_playbook.roles_path))
@@ -1857,6 +1863,7 @@ class ExecutorServer(object):
 
         self.library_dir = os.path.join(plugin_dir, 'library')
         self.action_dir = os.path.join(plugin_dir, 'action')
+        self.action_dir_general = os.path.join(plugin_dir, 'action-general')
         self.callback_dir = os.path.join(plugin_dir, 'callback')
         self.strategy_dir = os.path.join(plugin_dir, 'strategy')
         self.lookup_dir = os.path.join(plugin_dir, 'lookup')