Implement an opportunistic strategy and connection plugin

This change is creating an opportunistic Ansible execution strategy and
an update the ssh connection plugin so that it supports direct access to lxc
containers within ever having to ssh into them.

The intention of this change is to speed up execution time and reliability by
tuning the execution environment within Ansible to run faster while also attempting
to subvert transient ssh issues.

Change-Id: Ide34513bf82523257bdd2a8a68dff165f9927c56
Signed-off-by: Kevin Carter <kevin.carter@rackspace.com>

connection/ssh.py

@@ -0,0 +1,127 @@
+# Copyright 2016, Rackspace US, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# (c) 2016, Kevin Carter <kevin.carter@rackspace.com>
+
+import imp
+import os
+
+# NOTICE(cloudnull): The connection plugin imported using the full path to the
+#                    file because the ssh connection plugin is not importable.
+import ansible.plugins.connection as conn
+SSH = imp.load_source(
+    'ssh',
+    os.path.join(os.path.dirname(conn.__file__), 'ssh.py')
+)
+
+class Connection(SSH.Connection):
+    """Transport options for LXC containers.
+
+    This transport option makes the assumption that the playbook context has
+    vars within it that contain "physical_host" which is the machine running a
+    given container and "container_name" which is the actual name of the
+    container. These options can be added into the playbook via vars set as
+    attributes or though the modification of the a given execution strategy to
+    set the attributes accordingly.
+
+    This plugin operates exactly the same way as the standard SSH plugin but
+    will pad pathing or add command syntax for lxc containers when a container
+    is detected at runtime.
+    """
+
+    transport = 'ssh'
+
+    def __init__(self, *args, **kwargs):
+        super(Connection, self).__init__(*args, **kwargs)
+        self.args = args
+        self.kwargs = kwargs
+        self.vars = self._play_context._attributes['vars']
+        self.container_name = self.vars.get('container_name')
+        self.physical_host = self.vars.get('physical_host')
+        self.physical_hostname = self.vars.get('physical_hostname')
+        if self._container_check():
+            self.host = self._play_context.remote_addr = self.physical_host
+
+    def _exec_command(self, cmd, in_data=None, sudoable=True):
+        """run a command on the remote host."""
+        if self._container_check():
+            lxc_command = 'lxc-attach --name %s' % self.container_name
+            cmd = '%s -- %s' % (lxc_command, cmd)
+
+        return super(Connection, self)._exec_command(cmd, in_data, sudoable)
+
+    def _container_check(self):
+        if self.container_name:
+            SSH.display.vvv(u'container_name: "%s"' % self.container_name)
+            if self.physical_hostname:
+                SSH.display.vvv(
+                    u'physical_hostname: "%s"' % self.physical_hostname
+                )
+                if self.container_name != self.physical_hostname:
+                    SSH.display.vvv(u'Container confirmed')
+                    return True
+
+        return False
+
+    def _container_path_pad(self, path, fake_path=False):
+        args = (
+            'ssh',
+            self.host,
+            u"lxc-info --name %s --pid | awk '/PID:/ {print $2}'"
+            % self.container_name
+        )
+        returncode, stdout, _ = self._run(
+            self._build_command(*args),
+            in_data=None,
+            sudoable=False
+        )
+        if returncode == 0:
+            pad = os.path.join(
+                '/proc/%s/root' % stdout.strip(),
+                path.lstrip(os.sep)
+            )
+            SSH.display.vvv(
+                u'The path has been padded with the following to support a'
+                u' container rootfs: [ %s ]' % pad
+            )
+            return pad
+        else:
+            raise SSH.AnsibleError(
+                u'No valid container info was found for container "%s" Please'
+                u' check the state of the container.' % self.container_name
+            )
+
+    def fetch_file(self, in_path, out_path):
+        """fetch a file from remote to local."""
+        if self._container_check():
+            in_path = self._container_path_pad(path=in_path)
+
+        return super(Connection, self).fetch_file(in_path, out_path)
+
+    def put_file(self, in_path, out_path):
+        """transfer a file from local to remote."""
+        if self._container_check():
+            out_path = self._container_path_pad(path=out_path)
+
+        return super(Connection, self).put_file(in_path, out_path)
+
+    def close(self):
+        # If we have a persistent ssh connection (ControlPersist), we can ask it
+        # to stop listening. Otherwise, there's nothing to do here.
+        if self._connected and self._persistent:
+            cmd = self._build_command('ssh', '-O', 'stop', self.host)
+            cmd = map(SSH.to_bytes, cmd)
+            p = SSH.subprocess.Popen(cmd, stdin=SSH.subprocess.PIPE, stdout=SSH.subprocess.PIPE, stderr=SSH.subprocess.PIPE)
+            p.communicate()
+

releasenotes/notes/opportunistic-strategy-and-connection-plugin-bc476fa3607dcc4a.yaml

@@ -0,0 +1,11 @@
+---
+features:
+  - An opportunistic Ansible execution strategy has been implemented. This
+    allows the Ansible linear strategy to skip tasks with conditionals faster
+    by never queuing the task when the conditional is evaluated to be false.
+  - The Ansible SSH plugin has been modified to support running commands within
+    containers without having to directly ssh into them. The change will detect
+    presence of a container. If a container is found the physical host will be
+    used as the SSH target and commands will be run directly. This will improve
+    system reliability and speed while also opening up the possibility for SSH
+    to be disabled from within the container itself.

strategy/linear.py

@@ -0,0 +1,155 @@
+# Copyright 2016, Rackspace US, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# (c) 2016, Kevin Carter <kevin.carter@rackspace.com>
+
+import copy
+import imp
+import os
+
+# NOTICE(cloudnull): The connection plugin imported using the full path to the
+#                    file because the linear strategy plugin is not importable.
+import ansible.plugins.strategy as strategy
+LINEAR = imp.load_source(
+    'ssh',
+    os.path.join(os.path.dirname(strategy.__file__), 'linear.py')
+)
+
+
+class StrategyModule(LINEAR.StrategyModule):
+    """Notes about this strategy.
+
+    When this strategy encounters a task with a "when" or "register" stanza it
+    will collect results immediately essentially forming a block. If the task
+    does not have a "when" or "register" stanza the results will be collected
+    after all tasks have been queued.
+
+    To improve execution speed if a task has a "when" conditional attached to
+    it the conditional will be rendered before queuing the task and should the
+    conditional evaluate to True the task will be queued. To ensure the correct
+    execution of playbooks this optimisation will only be used if there are no
+    lookups used with the task which is to guarantee proper task execution.
+
+    To optimize transport reliability if a task is using a "delegate_to" stanza
+    the connection method will change to paramiko if the connection option has
+    been set at "smart", the Ansible 2.x default. Regardless of the connection
+    method if a "delegate_to" is used the task will have pipelining disabled
+    for the duration of that specific task.
+
+    Container context will be added to the ``playbook_context`` which is used
+    to further optimise connectivity by only ever SSH'ing into a given host
+    machine instead of attempting an SSH connection into a container.
+    """
+
+    @staticmethod
+    def _check_when(host, task, templar, task_vars):
+        """Evaluate if conditionals are to be run.
+
+        This will error on the side of caution:
+            * If a conditional is detected to be valid the method will return
+              True.
+            * If there's ever an issue with the templated conditional the
+              method will also return True.
+            * If the task has a detected "with" the method will return True.
+
+        :param host: object
+        :param task: object
+        :param templar: object
+        :param task_vars: dict
+        """
+        try:
+            if not task.when or (task.when and task.register):
+                return True
+
+            _ds = getattr(task, '_ds', dict())
+            if any([i for i in _ds.keys() if i.startswith('with')]):
+                return True
+
+            conditional = task.evaluate_conditional(templar, task_vars)
+            if not conditional:
+                LINEAR.display.verbose(
+                    u'Task "%s" has been omitted from the job because the'
+                    u' conditional "%s" was evaluated as "%s"'
+                    % (task.name, task.when, conditional),
+                    host=host,
+                    caplevel=0
+                )
+                return False
+        except Exception:
+            return True
+        else:
+            return True
+
+    def _queue_task(self, host, task, task_vars, play_context):
+        """Queue a task to be sent to the worker.
+
+        Modify the playbook_context to support adding attributes for LXC
+        containers.
+        """
+        templar = LINEAR.Templar(loader=self._loader, variables=task_vars)
+        if not self._check_when(host, task, templar, task_vars):
+            return
+
+        _play_context = copy.deepcopy(play_context)
+        _vars = _play_context._attributes['vars']
+        if task.delegate_to:
+            # If a task uses delegation change teh play_context
+            #  to use paramiko with pipelining disabled for this
+            #  one task on its collection of hosts.
+            if _play_context.pipelining:
+                _play_context.pipelining = False
+                LINEAR.display.verbose(
+                    u'Because this is a task using "delegate_to"'
+                    u' pipelining has been disabled. but will be'
+                    u' restored upon completion of this task.',
+                    host=host,
+                    caplevel=0
+                )
+
+            if _play_context.connection == 'smart':
+                _play_context.connection = 'paramiko'
+                LINEAR.display.verbose(
+                    u'Delegated task transport changing from'
+                    u' "%s" to "%s". The context will be restored'
+                    u' once the task has completed.' % (
+                        _play_context.connection,
+                        _play_context.connection
+                    ),
+                    host=host,
+                    caplevel=0
+                )
+        else:
+            if 'physical_host' in task_vars:
+                physical_host = _vars.get('physical_host')
+                if not physical_host:
+                    physical_host = task_vars.get('physical_host')
+                    if physical_host:
+                        ph = self._inventory.get_host(physical_host)
+                        _vars['physical_host'] = ph.vars['ansible_ssh_host']
+                        _vars['physical_hostname'] = physical_host
+
+            if 'container_name' in task_vars:
+                container_name = _vars.get('container_name')
+                if not container_name:
+                    container_name = task_vars.get('container_name')
+                    if container_name:
+                        _vars['container_name'] = task_vars['container_name']
+
+        return super(StrategyModule, self)._queue_task(
+            host,
+            task,
+            task_vars,
+            _play_context
+        )
+