BaseParallelSSHRunner.__init__() - Code Metrics - Inspection of "Merge pull request #3657 from StackStorm/python_ru..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( f1fe9e...5c5de8 )

unknown

created 2017-09-14 10:00 UTC

BaseParallelSSHRunner.init() A

↳ Parent: BaseParallelSSHRunner

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

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

# 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.

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.services.action import store_execution_output_data

__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
        if self.context:
            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
        }

        def make_store_stdout_line_func(execution_db, action_db):
            def store_stdout_line(line):
                if cfg.CONF.actionrunner.stream_output:
                    store_execution_output_data(execution_db=execution_db, action_db=action_db,
                                                data=line, output_type='stdout')

            return store_stdout_line

        def make_store_stderr_line_func(execution_db, action_db):
            def store_stderr_line(line):
                if cfg.CONF.actionrunner.stream_output:
                    store_execution_output_data(execution_db=execution_db, action_db=action_db,
                                                data=line, output_type='stderr')

            return store_stderr_line

        handle_stdout_line_func = make_store_stdout_line_func(execution_db=self.execution,
                                                              action_db=self.action)
        handle_stderr_line_func = make_store_stderr_line_func(execution_db=self.execution,
                                                              action_db=self.action)

        if len(self._hosts) == 1:
            # We only support streaming output when running action on one host. That is because
            # the action output is tied to a particulat execution. User can still achieve output
            # streaming for multiple hosts by running one execution per host.
            client_kwargs['handle_stdout_line_func'] = handle_stdout_line_func
            client_kwargs['handle_stderr_line_func'] = handle_stderr_line_func

        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 post_run(self, status, result):
        super(BaseParallelSSHRunner, self).post_run(status=status, result=result)

        # Ensure we close the connection when the action execution finishes
        if self._parallel_ssh_client:
            self._parallel_ssh_client.close()

    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			from oslo_config import cfg
17			import six
18
19			from st2common.runners.base import ShellRunnerMixin
20			from st2common.runners.base import ActionRunner
21			from st2common.constants.runners import REMOTE_RUNNER_PRIVATE_KEY_HEADER
22			from st2common.runners.parallel_ssh import ParallelSSHClient
23			from st2common import log as logging
24			from st2common.constants.action import LIVEACTION_STATUS_SUCCEEDED
25			from st2common.constants.action import LIVEACTION_STATUS_TIMED_OUT
26			from st2common.constants.action import LIVEACTION_STATUS_FAILED
27			from st2common.constants.runners import REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT
28			from st2common.exceptions.actionrunner import ActionRunnerPreRunError
29			from st2common.services.action import store_execution_output_data
30
31			__all__ = [
32			'BaseParallelSSHRunner'
33			]
34
35			LOG = logging.getLogger(__name__)
36
37			# constants to lookup in runner_parameters.
38			RUNNER_HOSTS = 'hosts'
39			RUNNER_USERNAME = 'username'
40			RUNNER_PASSWORD = 'password'
41			RUNNER_PRIVATE_KEY = 'private_key'
42			RUNNER_PARALLEL = 'parallel'
43			RUNNER_SUDO = 'sudo'
44			RUNNER_ON_BEHALF_USER = 'user'
45			RUNNER_REMOTE_DIR = 'dir'
46			RUNNER_COMMAND = 'cmd'
47			RUNNER_CWD = 'cwd'
48			RUNNER_ENV = 'env'
49			RUNNER_KWARG_OP = 'kwarg_op'
50			RUNNER_TIMEOUT = 'timeout'
51			RUNNER_SSH_PORT = 'port'
52			RUNNER_BASTION_HOST = 'bastion_host'
53			RUNNER_PASSPHRASE = 'passphrase'
54
55
56			class BaseParallelSSHRunner(ActionRunner, ShellRunnerMixin):
57
58			def __init__(self, runner_id):
59			super(BaseParallelSSHRunner, self).__init__(runner_id=runner_id)
60			self._hosts = None
61			self._parallel = True
62			self._sudo = False
63			self._on_behalf_user = None
64			self._username = None
65			self._password = None
66			self._private_key = None
67			self._passphrase = None
68			self._kwarg_op = '--'
69			self._cwd = None
70			self._env = None
71			self._ssh_port = None
72			self._timeout = None
73			self._bastion_host = None
74			self._on_behalf_user = cfg.CONF.system_user.user
75
76			self._ssh_key_file = None
77			self._parallel_ssh_client = None
78			self._max_concurrency = cfg.CONF.ssh_runner.max_parallel_actions
79
80			def pre_run(self):
81			super(BaseParallelSSHRunner, self).pre_run()
82
83			LOG.debug('Entering BaseParallelSSHRunner.pre_run() for liveaction_id="%s"',
84			self.liveaction_id)
85			hosts = self.runner_parameters.get(RUNNER_HOSTS, '').split(',')
86			self._hosts = [h.strip() for h in hosts if len(h) > 0]
87			if len(self._hosts) < 1:
88			raise ActionRunnerPreRunError('No hosts specified to run action for action %s.',
89			self.liveaction_id)
90			self._username = self.runner_parameters.get(RUNNER_USERNAME, None)
91			self._password = self.runner_parameters.get(RUNNER_PASSWORD, None)
92			self._private_key = self.runner_parameters.get(RUNNER_PRIVATE_KEY, None)
93			self._passphrase = self.runner_parameters.get(RUNNER_PASSPHRASE, None)
94
95			self._ssh_port = self.runner_parameters.get(RUNNER_SSH_PORT, None)
96			self._ssh_key_file = self._private_key
97			self._parallel = self.runner_parameters.get(RUNNER_PARALLEL, True)
98			self._sudo = self.runner_parameters.get(RUNNER_SUDO, False)
99			self._sudo = self._sudo if self._sudo else False
100			if self.context:
101			self._on_behalf_user = self.context.get(RUNNER_ON_BEHALF_USER, self._on_behalf_user)
102			self._cwd = self.runner_parameters.get(RUNNER_CWD, None)
103			self._env = self.runner_parameters.get(RUNNER_ENV, {})
104			self._kwarg_op = self.runner_parameters.get(RUNNER_KWARG_OP, '--')
105			self._timeout = self.runner_parameters.get(RUNNER_TIMEOUT,
106			REMOTE_RUNNER_DEFAULT_ACTION_TIMEOUT)
107			self._bastion_host = self.runner_parameters.get(RUNNER_BASTION_HOST, None)
108
109			LOG.info('[BaseParallelSSHRunner="%s", liveaction_id="%s"] Finished pre_run.',
110			self.runner_id, self.liveaction_id)
111
112			concurrency = int(len(self._hosts) / 3) + 1 if self._parallel else 1
113			if concurrency > self._max_concurrency:
114			LOG.debug('Limiting parallel SSH concurrency to %d.', concurrency)
115			concurrency = self._max_concurrency
116
117			client_kwargs = {
118			'hosts': self._hosts,
119			'user': self._username,
120			'port': self._ssh_port,
121			'concurrency': concurrency,
122			'bastion_host': self._bastion_host,
123			'raise_on_any_error': False,
124			'connect': True
125			}
126
127			def make_store_stdout_line_func(execution_db, action_db):
128			def store_stdout_line(line):
129			if cfg.CONF.actionrunner.stream_output:
130			store_execution_output_data(execution_db=execution_db, action_db=action_db,
131			data=line, output_type='stdout')
132
133			return store_stdout_line
134
135			def make_store_stderr_line_func(execution_db, action_db):
136			def store_stderr_line(line):
137			if cfg.CONF.actionrunner.stream_output:
138			store_execution_output_data(execution_db=execution_db, action_db=action_db,
139			data=line, output_type='stderr')
140
141			return store_stderr_line
142
143			handle_stdout_line_func = make_store_stdout_line_func(execution_db=self.execution,
144			action_db=self.action)
145			handle_stderr_line_func = make_store_stderr_line_func(execution_db=self.execution,
146			action_db=self.action)
147
148			if len(self._hosts) == 1:
149			# We only support streaming output when running action on one host. That is because
150			# the action output is tied to a particulat execution. User can still achieve output
151			# streaming for multiple hosts by running one execution per host.
152			client_kwargs['handle_stdout_line_func'] = handle_stdout_line_func
153			client_kwargs['handle_stderr_line_func'] = handle_stderr_line_func
154
155			if self._password:
156			client_kwargs['password'] = self._password
157			elif self._private_key:
158			# Determine if the private_key is a path to the key file or the raw key material
159			is_key_material = self._is_private_key_material(private_key=self._private_key)
160
161			if is_key_material:
162			# Raw key material
163			client_kwargs['pkey_material'] = self._private_key
164			else:
165			# Assume it's a path to the key file, verify the file exists
166			client_kwargs['pkey_file'] = self._private_key
167
168			if self._passphrase:
169			client_kwargs['passphrase'] = self._passphrase
170			else:
171			# Default to stanley key file specified in the config
172			client_kwargs['pkey_file'] = self._ssh_key_file
173
174			self._parallel_ssh_client = ParallelSSHClient(**client_kwargs)
175
176			def post_run(self, status, result):
177			super(BaseParallelSSHRunner, self).post_run(status=status, result=result)
178
179			# Ensure we close the connection when the action execution finishes
180			if self._parallel_ssh_client:
181			self._parallel_ssh_client.close()
182
183			def _is_private_key_material(self, private_key):
184			return private_key and REMOTE_RUNNER_PRIVATE_KEY_HEADER in private_key.lower()
185
186			def _get_env_vars(self):
187			"""
188			:rtype: ``dict``
189			"""
190			env_vars = {}
191
192			if self._env:
193			env_vars.update(self._env)
194
195			# Include common st2 env vars
196			st2_env_vars = self._get_common_action_env_variables()
197			env_vars.update(st2_env_vars)
198
199			return env_vars
200
201			@staticmethod
202			def _get_result_status(result, allow_partial_failure):
203
204			if 'error' in result and 'traceback' in result:
205			# Assume this is a global failure where the result dictionary doesn't contain entry
206			# per host
207			timeout = False
208			success = result.get('succeeded', False)
209			status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
210			timeout=timeout)
211			return status
212
213			success = not allow_partial_failure
214			timeout = True
215
216			for r in six.itervalues(result):
217			r_succeess = r.get('succeeded', False) if r else False
218			r_timeout = r.get('timeout', False) if r else False
219
220			timeout &= r_timeout
221
222			if allow_partial_failure:
223			success \|= r_succeess
224			if success:
225			break
226			else:
227			success &= r_succeess
228			if not success:
229			break
230
231			status = BaseParallelSSHRunner._get_status_for_success_and_timeout(success=success,
232			timeout=timeout)
233
234			return status
235
236			@staticmethod
237			def _get_status_for_success_and_timeout(success, timeout):
238			if success:
239			status = LIVEACTION_STATUS_SUCCEEDED
240			elif timeout:
241			# Note: Right now we only set status to timeout if all the hosts have timed out
242			status = LIVEACTION_STATUS_TIMED_OUT
243			else:
244			status = LIVEACTION_STATUS_FAILED
245			return status
246

StackStorm / st2

Push — master ( f1fe9e...5c5de8 )

BaseParallelSSHRunner.__init__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

BaseParallelSSHRunner.init() A