st2actions.runners.RemoteScriptRunner - Code Metrics - Inspection of "Merge pull request #2347 from StackStorm/ac-vars" - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( fe66bc...2a24eb )

by Manas

created 2016-01-05 23:05 UTC

st2actions.runners.RemoteScriptRunner A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	43
Duplicated Lines	0 %

Metric	Value
wmc	3
dl	0
loc	43
rs	10

# 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
import sys
import traceback
import uuid

from oslo_config import cfg

from st2common import log as logging
from st2actions.runners.ssh.fabric_runner import BaseFabricRunner
from st2actions.runners.ssh.fabric_runner import RUNNER_REMOTE_DIR
from st2actions.runners.ssh.paramiko_ssh_runner import BaseParallelSSHRunner
from st2common.models.system.action import FabricRemoteScriptAction
from st2common.models.system.paramiko_script_action import ParamikoRemoteScriptAction

__all__ = [
    'get_runner',
    'ParamikoRemoteScriptRunner',
    'RemoteScriptRunner'
]

LOG = logging.getLogger(__name__)


def get_runner():
    if cfg.CONF.ssh_runner.use_paramiko_ssh_runner:
        return ParamikoRemoteScriptRunner(str(uuid.uuid4()))
    return RemoteScriptRunner(str(uuid.uuid4()))


class RemoteScriptRunner(BaseFabricRunner):
    def run(self, action_parameters):
        remote_action = self._get_remote_action(action_parameters)

        LOG.debug('Will execute remote_action : %s.', str(remote_action))
        result = self._run(remote_action)
        LOG.debug('Executed remote_action: %s. Result is : %s.', remote_action, result)
        status = self._get_result_status(result, cfg.CONF.ssh_runner.allow_partial_failure)

        return (status, result, None)

    def _get_remote_action(self, action_parameters):
        # remote script actions without entry_point don't make sense, user probably wanted to use
        # "run-remote" action
        if not self.entry_point:
            msg = ('Action "%s" is missing entry_point attribute. Perhaps wanted to use '
                   '"run-remote" runner?')
            raise Exception(msg % (self.action_name))

        script_local_path_abs = self.entry_point
        pos_args, named_args = self._get_script_args(action_parameters)
        named_args = self._transform_named_args(named_args)
        env_vars = self._get_env_vars()
        remote_dir = self.runner_parameters.get(RUNNER_REMOTE_DIR,
                                                cfg.CONF.ssh_runner.remote_dir)
        remote_dir = os.path.join(remote_dir, self.liveaction_id)
        return FabricRemoteScriptAction(self.action_name,
                                        str(self.liveaction_id),
                                        script_local_path_abs,
                                        self.libs_dir_path,
                                        named_args=named_args,
                                        positional_args=pos_args,
                                        env_vars=env_vars,
                                        on_behalf_user=self._on_behalf_user,
                                        user=self._username,
                                        password=self._password,
                                        private_key=self._private_key,
                                        remote_dir=remote_dir,
                                        hosts=self._hosts,
                                        parallel=self._parallel,
                                        sudo=self._sudo,
                                        timeout=self._timeout,
                                        cwd=self._cwd)


class ParamikoRemoteScriptRunner(BaseParallelSSHRunner):
    def run(self, action_parameters):
        remote_action = self._get_remote_action(action_parameters)

        LOG.debug('Executing remote action.', extra={'_action_params': remote_action})
        result = self._run(remote_action)
        LOG.debug('Executed remote action.', extra={'_result': result})
        status = self._get_result_status(result, cfg.CONF.ssh_runner.allow_partial_failure)

        return (status, result, None)

    def _run(self, remote_action):
        try:
            copy_results = self._copy_artifacts(remote_action)
        except:
            # If for whatever reason there is a top level exception,
            # we just bail here.
            error = 'Failed copying content to remote boxes.'
            LOG.exception(error)
            _, ex, tb = sys.exc_info()
            copy_results = self._generate_error_results(' '.join([error, str(ex)]), tb)
            return copy_results

        try:
            exec_results = self._run_script_on_remote_host(remote_action)
            try:
                remote_dir = remote_action.get_remote_base_dir()
                LOG.debug('Deleting remote execution dir.', extra={'_remote_dir': remote_dir})
                delete_results = self._parallel_ssh_client.delete_dir(path=remote_dir,
                                                                      force=True)
                LOG.debug('Deleted remote execution dir.', extra={'_result': delete_results})
            except:
                LOG.exception('Failed deleting remote dir.', extra={'_remote_dir': remote_dir})
            finally:
                return exec_results
        except:
            error = 'Failed executing script on remote boxes.'
            LOG.exception(error, extra={'_action_params': remote_action})
            _, ex, tb = sys.exc_info()
            exec_results = self._generate_error_results(' '.join([error, str(ex)]), tb)
            return exec_results

    def _copy_artifacts(self, remote_action):
        # First create remote execution directory.
        remote_dir = remote_action.get_remote_base_dir()
        LOG.debug('Creating remote execution dir.', extra={'_path': remote_dir})
        self._parallel_ssh_client.mkdir(path=remote_action.get_remote_base_dir())

        # Copy the script to remote dir in remote host.
        local_script_abs_path = remote_action.get_local_script_abs_path()
        remote_script_abs_path = remote_action.get_remote_script_abs_path()
        file_mode = 0744
        extra = {'_local_script': local_script_abs_path, '_remote_script': remote_script_abs_path,
                 'mode': file_mode}
        LOG.debug('Copying local script to remote box.', extra=extra)
        self._parallel_ssh_client.put(local_path=local_script_abs_path,
                                      remote_path=remote_script_abs_path,
                                      mirror_local_mode=False, mode=file_mode)

        # If `lib` exist for the script, copy that to remote host.
        local_libs_path = remote_action.get_local_libs_path_abs()
        if os.path.exists(local_libs_path):
            extra = {'_local_libs': local_libs_path, '_remote_path': remote_dir}
            LOG.debug('Copying libs to remote host.', extra=extra)
            self._parallel_ssh_client.put(local_path=local_libs_path,
                                          remote_path=remote_dir,
                                          mirror_local_mode=True)

    def _run_script_on_remote_host(self, remote_action):
        command = remote_action.get_full_command_string()
        LOG.info('Command to run: %s', command)
        results = self._parallel_ssh_client.run(command, timeout=remote_action.get_timeout())
        LOG.debug('Results from script: %s', results)
        return results

    def _get_remote_action(self, action_parameters):
        # remote script actions without entry_point don't make sense, user probably wanted to use
        # "run-remote" action
        if not self.entry_point:
            msg = ('Action "%s" is missing entry_point attribute. Perhaps wanted to use '
                   '"run-remote" runner?')
            raise Exception(msg % (self.action_name))

        script_local_path_abs = self.entry_point
        pos_args, named_args = self._get_script_args(action_parameters)
        named_args = self._transform_named_args(named_args)
        env_vars = self._get_env_vars()
        remote_dir = self.runner_parameters.get(RUNNER_REMOTE_DIR,
                                                cfg.CONF.ssh_runner.remote_dir)
        remote_dir = os.path.join(remote_dir, self.liveaction_id)
        return ParamikoRemoteScriptAction(self.action_name,
                                          str(self.liveaction_id),
                                          script_local_path_abs,
                                          self.libs_dir_path,
                                          named_args=named_args,
                                          positional_args=pos_args,
                                          env_vars=env_vars,
                                          on_behalf_user=self._on_behalf_user,
                                          user=self._username,
                                          password=self._password,
                                          private_key=self._private_key,
                                          remote_dir=remote_dir,
                                          hosts=self._hosts,
                                          parallel=self._parallel,
                                          sudo=self._sudo,
                                          timeout=self._timeout,
                                          cwd=self._cwd)

    @staticmethod
    def _generate_error_results(error, tb):
        error_dict = {
            'error': error,
            'traceback': ''.join(traceback.format_tb(tb, 20)) if tb else '',
            'failed': True,
            'succeeded': False,
            'return_code': 255
        }
        return error_dict


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
17			import sys
18			import traceback
19			import uuid
20
21			from oslo_config import cfg
22
23			from st2common import log as logging
24			from st2actions.runners.ssh.fabric_runner import BaseFabricRunner
25			from st2actions.runners.ssh.fabric_runner import RUNNER_REMOTE_DIR
26			from st2actions.runners.ssh.paramiko_ssh_runner import BaseParallelSSHRunner
27			from st2common.models.system.action import FabricRemoteScriptAction
28			from st2common.models.system.paramiko_script_action import ParamikoRemoteScriptAction
29
30			__all__ = [
31			'get_runner',
32			'ParamikoRemoteScriptRunner',
33			'RemoteScriptRunner'
34			]
35
36			LOG = logging.getLogger(__name__)
37
38
39			def get_runner():
40			if cfg.CONF.ssh_runner.use_paramiko_ssh_runner:
41			return ParamikoRemoteScriptRunner(str(uuid.uuid4()))
42			return RemoteScriptRunner(str(uuid.uuid4()))
43
44
45			class RemoteScriptRunner(BaseFabricRunner):
46			def run(self, action_parameters):
47			remote_action = self._get_remote_action(action_parameters)
48
49			LOG.debug('Will execute remote_action : %s.', str(remote_action))
50			result = self._run(remote_action)
51			LOG.debug('Executed remote_action: %s. Result is : %s.', remote_action, result)
52			status = self._get_result_status(result, cfg.CONF.ssh_runner.allow_partial_failure)
53
54			return (status, result, None)
55
56			def _get_remote_action(self, action_parameters):
57			# remote script actions without entry_point don't make sense, user probably wanted to use
58			# "run-remote" action
59			if not self.entry_point:
60			msg = ('Action "%s" is missing entry_point attribute. Perhaps wanted to use '
61			'"run-remote" runner?')
62			raise Exception(msg % (self.action_name))
63
64			script_local_path_abs = self.entry_point
65			pos_args, named_args = self._get_script_args(action_parameters)
66			named_args = self._transform_named_args(named_args)
67			env_vars = self._get_env_vars()
68			remote_dir = self.runner_parameters.get(RUNNER_REMOTE_DIR,
69			cfg.CONF.ssh_runner.remote_dir)
70			remote_dir = os.path.join(remote_dir, self.liveaction_id)
71			return FabricRemoteScriptAction(self.action_name,
72			str(self.liveaction_id),
73			script_local_path_abs,
74			self.libs_dir_path,
75			named_args=named_args,
76			positional_args=pos_args,
77			env_vars=env_vars,
78			on_behalf_user=self._on_behalf_user,
79			user=self._username,
80			password=self._password,
81			private_key=self._private_key,
82			remote_dir=remote_dir,
83			hosts=self._hosts,
84			parallel=self._parallel,
85			sudo=self._sudo,
86			timeout=self._timeout,
87			cwd=self._cwd)
88
89
90			class ParamikoRemoteScriptRunner(BaseParallelSSHRunner):
91			def run(self, action_parameters):
92			remote_action = self._get_remote_action(action_parameters)
93
94			LOG.debug('Executing remote action.', extra={'_action_params': remote_action})
95			result = self._run(remote_action)
96			LOG.debug('Executed remote action.', extra={'_result': result})
97			status = self._get_result_status(result, cfg.CONF.ssh_runner.allow_partial_failure)
98
99			return (status, result, None)
100
101			def _run(self, remote_action):
102			try:
103			copy_results = self._copy_artifacts(remote_action)
104			except:
105			# If for whatever reason there is a top level exception,
106			# we just bail here.
107			error = 'Failed copying content to remote boxes.'
108			LOG.exception(error)
109			_, ex, tb = sys.exc_info()
110			copy_results = self._generate_error_results(' '.join([error, str(ex)]), tb)
111			return copy_results
112
113			try:
114			exec_results = self._run_script_on_remote_host(remote_action)
115			try:
116			remote_dir = remote_action.get_remote_base_dir()
117			LOG.debug('Deleting remote execution dir.', extra={'_remote_dir': remote_dir})
118			delete_results = self._parallel_ssh_client.delete_dir(path=remote_dir,
119			force=True)
120			LOG.debug('Deleted remote execution dir.', extra={'_result': delete_results})
121			except:
122			LOG.exception('Failed deleting remote dir.', extra={'_remote_dir': remote_dir})
123			finally:
124			return exec_results
125			except:
126			error = 'Failed executing script on remote boxes.'
127			LOG.exception(error, extra={'_action_params': remote_action})
128			_, ex, tb = sys.exc_info()
129			exec_results = self._generate_error_results(' '.join([error, str(ex)]), tb)
130			return exec_results
131
132			def _copy_artifacts(self, remote_action):
133			# First create remote execution directory.
134			remote_dir = remote_action.get_remote_base_dir()
135			LOG.debug('Creating remote execution dir.', extra={'_path': remote_dir})
136			self._parallel_ssh_client.mkdir(path=remote_action.get_remote_base_dir())
137
138			# Copy the script to remote dir in remote host.
139			local_script_abs_path = remote_action.get_local_script_abs_path()
140			remote_script_abs_path = remote_action.get_remote_script_abs_path()
141			file_mode = 0744
142			extra = {'_local_script': local_script_abs_path, '_remote_script': remote_script_abs_path,
143			'mode': file_mode}
144			LOG.debug('Copying local script to remote box.', extra=extra)
145			self._parallel_ssh_client.put(local_path=local_script_abs_path,
146			remote_path=remote_script_abs_path,
147			mirror_local_mode=False, mode=file_mode)
148
149			# If `lib` exist for the script, copy that to remote host.
150			local_libs_path = remote_action.get_local_libs_path_abs()
151			if os.path.exists(local_libs_path):
152			extra = {'_local_libs': local_libs_path, '_remote_path': remote_dir}
153			LOG.debug('Copying libs to remote host.', extra=extra)
154			self._parallel_ssh_client.put(local_path=local_libs_path,
155			remote_path=remote_dir,
156			mirror_local_mode=True)
157
158			def _run_script_on_remote_host(self, remote_action):
159			command = remote_action.get_full_command_string()
160			LOG.info('Command to run: %s', command)
161			results = self._parallel_ssh_client.run(command, timeout=remote_action.get_timeout())
162			LOG.debug('Results from script: %s', results)
163			return results
164
165			def _get_remote_action(self, action_parameters):
166			# remote script actions without entry_point don't make sense, user probably wanted to use
167			# "run-remote" action
168			if not self.entry_point:
169			msg = ('Action "%s" is missing entry_point attribute. Perhaps wanted to use '
170			'"run-remote" runner?')
171			raise Exception(msg % (self.action_name))
172
173			script_local_path_abs = self.entry_point
174			pos_args, named_args = self._get_script_args(action_parameters)
175			named_args = self._transform_named_args(named_args)
176			env_vars = self._get_env_vars()
177			remote_dir = self.runner_parameters.get(RUNNER_REMOTE_DIR,
178			cfg.CONF.ssh_runner.remote_dir)
179			remote_dir = os.path.join(remote_dir, self.liveaction_id)
180			return ParamikoRemoteScriptAction(self.action_name,
181			str(self.liveaction_id),
182			script_local_path_abs,
183			self.libs_dir_path,
184			named_args=named_args,
185			positional_args=pos_args,
186			env_vars=env_vars,
187			on_behalf_user=self._on_behalf_user,
188			user=self._username,
189			password=self._password,
190			private_key=self._private_key,
191			remote_dir=remote_dir,
192			hosts=self._hosts,
193			parallel=self._parallel,
194			sudo=self._sudo,
195			timeout=self._timeout,
196			cwd=self._cwd)
197
198			@staticmethod
199			def _generate_error_results(error, tb):
200			error_dict = {
201			'error': error,
202			'traceback': ''.join(traceback.format_tb(tb, 20)) if tb else '',
203			'failed': True,
204			'succeeded': False,
205			'return_code': 255
206			}
207			return error_dict
208

StackStorm / st2

Push — master ( fe66bc...2a24eb )

st2actions.runners.RemoteScriptRunner A

Complexity

Size/Duplication

Duplication Side-by-Side

Filter issues like