BaseParallelSSHRunner.__init__() - Code Metrics - Inspection of "Paramiko ssh config file support" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#3058)

by Lakshmi

created 2016-11-28 23:55 UTC

BaseParallelSSHRunner.init() A

↳ Parent: BaseParallelSSHRunner

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	21
rs	9.3142

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.

import os


from oslo_config import cfg
import six

from st2common.runners.base import ShellRunnerMixin
from st2common.runners.base import ActionRunner
from st2common.constants.runners import REMOTE_RUNNER_PRIVATE_KEY_HEADER
from st2common.runners.parallel_ssh import ParallelSSHClient
from st2common import log as logging
from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
from st2common.constants.action import LIVEACTION_STATUS_TIMED_OUT
from st2common.constants.action import LIVEACTION_STATUS_FAILED
from st2common.constants.runners import REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT
from st2common.exceptions.actionrunner import ActionRunnerPreRunError
from st2common.exceptions.ssh import InvalidCredentialsException


__all__ = [
    'BaseParallelSSHRunner'
]

LOG = logging.getLogger(__name__)

# constants to lookup in runner_parameters.
RUNNER_HOSTS = 'hosts'
RUNNER_USERNAME = 'username'
RUNNER_PASSWORD = 'password'
RUNNER_PRIVATE_KEY = 'private_key'
RUNNER_PARALLEL = 'parallel'
RUNNER_SUDO = 'sudo'
RUNNER_ON_BEHALF_USER = 'user'
RUNNER_REMOTE_DIR = 'dir'
RUNNER_COMMAND = 'cmd'
RUNNER_CWD = 'cwd'
RUNNER_ENV = 'env'
RUNNER_KWARG_OP = 'kwarg_op'
RUNNER_TIMEOUT = 'timeout'
RUNNER_SSH_PORT = 'port'
RUNNER_BASTION_HOST = 'bastion_host'
RUNNER_PASSPHRASE = 'passphrase'


class BaseParallelSSHRunner(ActionRunner, ShellRunnerMixin):

    def __init__(self, runner_id):
        super(BaseParallelSSHRunner, self).__init__(runner_id=runner_id)
        self._hosts = None
        self._parallel = True
        self._sudo = False
        self._on_behalf_user = None
        self._username = None
        self._password = None
        self._private_key = None
        self._passphrase = None
        self._kwarg_op = '--'
        self._cwd = None
        self._env = None
        self._ssh_port = None
        self._timeout = None
        self._bastion_host = None
        self._on_behalf_user = cfg.CONF.system_user.user

        self._ssh_key_file = None
        self._parallel_ssh_client = None
        self._max_concurrency = cfg.CONF.ssh_runner.max_parallel_actions

    def pre_run(self):
        super(BaseParallelSSHRunner, self).pre_run()

        LOG.debug('Entering BaseParallelSSHRunner.pre_run() for liveaction_id="%s"',
                  self.liveaction_id)
        hosts = self.runner_parameters.get(RUNNER_HOSTS, '').split(',')
        self._hosts = [h.strip() for h in hosts if len(h) > 0]
        if len(self._hosts) < 1:
            raise ActionRunnerPreRunError('No hosts specified to run action for action %s.',
                                          self.liveaction_id)
        self._username = self.runner_parameters.get(RUNNER_USERNAME, None)
        self._password = self.runner_parameters.get(RUNNER_PASSWORD, None)
        self._private_key = self.runner_parameters.get(RUNNER_PRIVATE_KEY, None)
        self._passphrase = self.runner_parameters.get(RUNNER_PASSPHRASE, None)

        self._ssh_port = self.runner_parameters.get(RUNNER_SSH_PORT, None)
        self._ssh_key_file = self._private_key
        self._parallel = self.runner_parameters.get(RUNNER_PARALLEL, True)
        self._sudo = self.runner_parameters.get(RUNNER_SUDO, False)
        self._sudo = self._sudo if self._sudo else False
        self._on_behalf_user = self.context.get(RUNNER_ON_BEHALF_USER, self._on_behalf_user)
        self._cwd = self.runner_parameters.get(RUNNER_CWD, None)
        self._env = self.runner_parameters.get(RUNNER_ENV, {})
        self._kwarg_op = self.runner_parameters.get(RUNNER_KWARG_OP, '--')
        self._timeout = self.runner_parameters.get(RUNNER_TIMEOUT,
                                                   REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT)
        self._bastion_host = self.runner_parameters.get(RUNNER_BASTION_HOST, None)

        LOG.info('[BaseParallelSSHRunner="%s", liveaction_id="%s"] Finished pre_run.',
                 self.runner_id, self.liveaction_id)

        concurrency = int(len(self._hosts) / 3) + 1 if self._parallel else 1
        if concurrency > self._max_concurrency:
            LOG.debug('Limiting parallel SSH concurrency to %d.', concurrency)
            concurrency = self._max_concurrency

        client_kwargs = {
            'hosts': self._hosts,
            'user': self._username,
            'port': self._ssh_port,
            'concurrency': concurrency,
            'bastion_host': self._bastion_host,
            'raise_on_any_error': False,
            'connect': True
        }

        if self._password:
            client_kwargs['password'] = self._password
        elif self._private_key:
            # Determine if the private_key is a path to the key file or the raw key material
            is_key_material = self._is_private_key_material(private_key=self._private_key)

            if is_key_material:
                # Raw key material
                client_kwargs['pkey_material'] = self._private_key
            else:
                # Assume it's a path to the key file, verify the file exists
                client_kwargs['pkey_file'] = self._private_key

            if self._passphrase:
                client_kwargs['passphrase'] = self._passphrase
        else:
            # Default to stanley key file specified in the config
            client_kwargs['pkey_file'] = self._ssh_key_file

        self._parallel_ssh_client = ParallelSSHClient(**client_kwargs)

    def _is_private_key_material(self, private_key):
        return private_key and REMOTE_RUNNER_PRIVATE_KEY_HEADER in private_key.lower()

    def _get_env_vars(self):
        """
        :rtype: ``dict``
        """
        env_vars = {}

        if self._env:
            env_vars.update(self._env)

        # Include common st2 env vars
        st2_env_vars = self._get_common_action_env_variables()
        env_vars.update(st2_env_vars)

        return env_vars

    @staticmethod
    def _get_result_status(result, allow_partial_failure):

        if 'error' in result and 'traceback' in result:
            # Assume this is a global failure where the result dictionary doesn't contain entry
            # per host
            timeout = False
            success = result.get('succeeded', False)
            status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
                                                                               timeout=timeout)
            return status

        success = not allow_partial_failure
        timeout = True

        for r in six.itervalues(result):
            r_succeess = r.get('succeeded', False) if r else False
            r_timeout = r.get('timeout', False) if r else False

            timeout &= r_timeout

            if allow_partial_failure:
                success |= r_succeess
                if success:
                    break
            else:
                success &= r_succeess
                if not success:
                    break

        status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
                                                                           timeout=timeout)

        return status

    @staticmethod
    def _get_status_for_success_and_timeout(success, timeout):
        if success:
            status = LIVEACTION_STATUS_SUCCEEDED
        elif timeout:
            # Note: Right now we only set status to timeout if all the hosts have timed out
            status = LIVEACTION_STATUS_TIMED_OUT
        else:
            status = LIVEACTION_STATUS_FAILED
        return status


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
			0 ignored issues – show Unused Code introduced 2016-11-28 23:59 UTC by Report Bug Copy Issue Report The import `os` seems to be unused. Loading history...
17
18			from oslo_config import cfg
19			import six
20
21			from st2common.runners.base import ShellRunnerMixin
22			from st2common.runners.base import ActionRunner
23			from st2common.constants.runners import REMOTE_RUNNER_PRIVATE_KEY_HEADER
24			from st2common.runners.parallel_ssh import ParallelSSHClient
25			from st2common import log as logging
26			from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
27			from st2common.constants.action import LIVEACTION_STATUS_TIMED_OUT
28			from st2common.constants.action import LIVEACTION_STATUS_FAILED
29			from st2common.constants.runners import REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT
30			from st2common.exceptions.actionrunner import ActionRunnerPreRunError
31			from st2common.exceptions.ssh import InvalidCredentialsException
			0 ignored issues – show Unused Code introduced 2016-11-28 23:59 UTC by Report Bug Copy Issue Report Unused InvalidCredentialsException imported from st2common.exceptions.ssh Loading history...
32
33			__all__ = [
34			'BaseParallelSSHRunner'
35			]
36
37			LOG = logging.getLogger(__name__)
38
39			# constants to lookup in runner_parameters.
40			RUNNER_HOSTS = 'hosts'
41			RUNNER_USERNAME = 'username'
42			RUNNER_PASSWORD = 'password'
43			RUNNER_PRIVATE_KEY = 'private_key'
44			RUNNER_PARALLEL = 'parallel'
45			RUNNER_SUDO = 'sudo'
46			RUNNER_ON_BEHALF_USER = 'user'
47			RUNNER_REMOTE_DIR = 'dir'
48			RUNNER_COMMAND = 'cmd'
49			RUNNER_CWD = 'cwd'
50			RUNNER_ENV = 'env'
51			RUNNER_KWARG_OP = 'kwarg_op'
52			RUNNER_TIMEOUT = 'timeout'
53			RUNNER_SSH_PORT = 'port'
54			RUNNER_BASTION_HOST = 'bastion_host'
55			RUNNER_PASSPHRASE = 'passphrase'
56
57
58			class BaseParallelSSHRunner(ActionRunner, ShellRunnerMixin):
59
60			def __init__(self, runner_id):
61			super(BaseParallelSSHRunner, self).__init__(runner_id=runner_id)
62			self._hosts = None
63			self._parallel = True
64			self._sudo = False
65			self._on_behalf_user = None
66			self._username = None
67			self._password = None
68			self._private_key = None
69			self._passphrase = None
70			self._kwarg_op = '--'
71			self._cwd = None
72			self._env = None
73			self._ssh_port = None
74			self._timeout = None
75			self._bastion_host = None
76			self._on_behalf_user = cfg.CONF.system_user.user
77
78			self._ssh_key_file = None
79			self._parallel_ssh_client = None
80			self._max_concurrency = cfg.CONF.ssh_runner.max_parallel_actions
81
82			def pre_run(self):
83			super(BaseParallelSSHRunner, self).pre_run()
84
85			LOG.debug('Entering BaseParallelSSHRunner.pre_run() for liveaction_id="%s"',
86			self.liveaction_id)
87			hosts = self.runner_parameters.get(RUNNER_HOSTS, '').split(',')
88			self._hosts = [h.strip() for h in hosts if len(h) > 0]
89			if len(self._hosts) < 1:
90			raise ActionRunnerPreRunError('No hosts specified to run action for action %s.',
91			self.liveaction_id)
92			self._username = self.runner_parameters.get(RUNNER_USERNAME, None)
93			self._password = self.runner_parameters.get(RUNNER_PASSWORD, None)
94			self._private_key = self.runner_parameters.get(RUNNER_PRIVATE_KEY, None)
95			self._passphrase = self.runner_parameters.get(RUNNER_PASSPHRASE, None)
96
97			self._ssh_port = self.runner_parameters.get(RUNNER_SSH_PORT, None)
98			self._ssh_key_file = self._private_key
99			self._parallel = self.runner_parameters.get(RUNNER_PARALLEL, True)
100			self._sudo = self.runner_parameters.get(RUNNER_SUDO, False)
101			self._sudo = self._sudo if self._sudo else False
102			self._on_behalf_user = self.context.get(RUNNER_ON_BEHALF_USER, self._on_behalf_user)
103			self._cwd = self.runner_parameters.get(RUNNER_CWD, None)
104			self._env = self.runner_parameters.get(RUNNER_ENV, {})
105			self._kwarg_op = self.runner_parameters.get(RUNNER_KWARG_OP, '--')
106			self._timeout = self.runner_parameters.get(RUNNER_TIMEOUT,
107			REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT)
108			self._bastion_host = self.runner_parameters.get(RUNNER_BASTION_HOST, None)
109
110			LOG.info('[BaseParallelSSHRunner="%s", liveaction_id="%s"] Finished pre_run.',
111			self.runner_id, self.liveaction_id)
112
113			concurrency = int(len(self._hosts) / 3) + 1 if self._parallel else 1
114			if concurrency > self._max_concurrency:
115			LOG.debug('Limiting parallel SSH concurrency to %d.', concurrency)
116			concurrency = self._max_concurrency
117
118			client_kwargs = {
119			'hosts': self._hosts,
120			'user': self._username,
121			'port': self._ssh_port,
122			'concurrency': concurrency,
123			'bastion_host': self._bastion_host,
124			'raise_on_any_error': False,
125			'connect': True
126			}
127
128			if self._password:
129			client_kwargs['password'] = self._password
130			elif self._private_key:
131			# Determine if the private_key is a path to the key file or the raw key material
132			is_key_material = self._is_private_key_material(private_key=self._private_key)
133
134			if is_key_material:
135			# Raw key material
136			client_kwargs['pkey_material'] = self._private_key
137			else:
138			# Assume it's a path to the key file, verify the file exists
139			client_kwargs['pkey_file'] = self._private_key
140
141			if self._passphrase:
142			client_kwargs['passphrase'] = self._passphrase
143			else:
144			# Default to stanley key file specified in the config
145			client_kwargs['pkey_file'] = self._ssh_key_file
146
147			self._parallel_ssh_client = ParallelSSHClient(**client_kwargs)
148
149			def _is_private_key_material(self, private_key):
150			return private_key and REMOTE_RUNNER_PRIVATE_KEY_HEADER in private_key.lower()
151
152			def _get_env_vars(self):
153			"""
154			:rtype: ``dict``
155			"""
156			env_vars = {}
157
158			if self._env:
159			env_vars.update(self._env)
160
161			# Include common st2 env vars
162			st2_env_vars = self._get_common_action_env_variables()
163			env_vars.update(st2_env_vars)
164
165			return env_vars
166
167			@staticmethod
168			def _get_result_status(result, allow_partial_failure):
169
170			if 'error' in result and 'traceback' in result:
171			# Assume this is a global failure where the result dictionary doesn't contain entry
172			# per host
173			timeout = False
174			success = result.get('succeeded', False)
175			status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
176			timeout=timeout)
177			return status
178
179			success = not allow_partial_failure
180			timeout = True
181
182			for r in six.itervalues(result):
183			r_succeess = r.get('succeeded', False) if r else False
184			r_timeout = r.get('timeout', False) if r else False
185
186			timeout &= r_timeout
187
188			if allow_partial_failure:
189			success \|= r_succeess
190			if success:
191			break
192			else:
193			success &= r_succeess
194			if not success:
195			break
196
197			status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
198			timeout=timeout)
199
200			return status
201
202			@staticmethod
203			def _get_status_for_success_and_timeout(success, timeout):
204			if success:
205			status = LIVEACTION_STATUS_SUCCEEDED
206			elif timeout:
207			# Note: Right now we only set status to timeout if all the hosts have timed out
208			status = LIVEACTION_STATUS_TIMED_OUT
209			else:
210			status = LIVEACTION_STATUS_FAILED
211			return status
212

StackStorm / st2

Pull Request — master (#3058)

BaseParallelSSHRunner.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

BaseParallelSSHRunner.init() A