Merge "Implement an opportunistic strategy and connection plugin"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
2b38db1e1c
127
connection/ssh.py
Normal file
127
connection/ssh.py
Normal file
@ -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()
|
||||
|
||||
@ -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.
|
||||
155
strategy/linear.py
Normal file
155
strategy/linear.py
Normal file
@ -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
|
||||
)
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user