summaryrefslogtreecommitdiffstats
path: root/vnftest/onap/runners/base.py
blob: 5170bbe865a5eef8baeb344623dc964c7b94684e (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
##############################################################################
# Copyright 2018 EuropeanSoftwareMarketingLtd.
# ===================================================================
#  Licensed under the ApacheLicense, Version2.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
#
# 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
##############################################################################
# vnftest comment: this is a modified copy of
# rally/rally/benchmark/runners/base.py

from __future__ import absolute_import
import logging
import multiprocessing
import subprocess
import time
import traceback
import importlib
from Queue import Empty

import vnftest.common.utils as utils
from vnftest.onap.steps import base as base_step
from vnftest.onap.steps.onap_api_call import OnapApiCall

log = logging.getLogger(__name__)


def _execute_shell_command(command):
    """execute shell script with error handling"""
    exitcode = 0
    try:
        output = subprocess.check_output(command, shell=True)
    except Exception:
        exitcode = -1
        output = traceback.format_exc()
        log.error("exec command '%s' error:\n ", command)
        log.error(traceback.format_exc())

    return exitcode, output


def _single_action(seconds, command, queue):
    """entrypoint for the single action process"""
    log.debug("single action, fires after %d seconds (from now)", seconds)
    time.sleep(seconds)
    log.debug("single action: executing command: '%s'", command)
    ret_code, data = _execute_shell_command(command)
    if ret_code < 0:
        log.error("single action error! command:%s", command)
        queue.put({'single-action-data': data})
        return
    log.debug("single action data: \n%s", data)
    queue.put({'single-action-data': data})


def _periodic_action(interval, command, queue):
    """entrypoint for the periodic action process"""
    log.debug("periodic action, fires every: %d seconds", interval)
    time_spent = 0
    while True:
        time.sleep(interval)
        time_spent += interval
        log.debug("periodic action, executing command: '%s'", command)
        ret_code, data = _execute_shell_command(command)
        if ret_code < 0:
            log.error("periodic action error! command:%s", command)
            queue.put({'periodic-action-data': data})
            break
        log.debug("periodic action data: \n%s", data)
        queue.put({'periodic-action-data': data})


class Runner(object):
    runners = []

    @staticmethod
    def get_cls(runner_type):
        """return class of specified type"""
        for runner in utils.itersubclasses(Runner):
            if runner_type == runner.__execution_type__:
                return runner
        raise RuntimeError("No such runner_type %s" % runner_type)

    @staticmethod
    def get_types():
        """return a list of known runner type (class) names"""
        types = []
        for runner in utils.itersubclasses(Runner):
            types.append(runner)
        return types

    @staticmethod
    def get(runner_cfg):
        """Returns instance of a step runner for execution type.
        """
        return Runner.get_cls(runner_cfg["type"])(runner_cfg)

    @staticmethod
    def release(runner):
        """Release the runner"""
        if runner in Runner.runners:
            Runner.runners.remove(runner)

    @staticmethod
    def terminate(runner):
        """Terminate the runner"""
        if runner.process and runner.process.is_alive():
            runner.process.terminate()

    @staticmethod
    def terminate_all():
        """Terminate all runners (subprocesses)"""
        log.debug("Terminating all runners", exc_info=True)

        # release dumper process as some errors before any runner is created
        if not Runner.runners:
            return

        for runner in Runner.runners:
            log.debug("Terminating runner: %s", runner)
            if runner.process:
                runner.process.terminate()
                runner.process.join()
            if runner.periodic_action_process:
                log.debug("Terminating periodic action process")
                runner.periodic_action_process.terminate()
                runner.periodic_action_process = None
            Runner.release(runner)

    def __init__(self, config):
        self.config = config
        self.periodic_action_process = None
        self.output_queue = multiprocessing.Queue()
        self.result_queue = multiprocessing.Queue()
        self.process = None
        self.aborted = multiprocessing.Event()
        Runner.runners.append(self)

    def run_post_stop_action(self):
        """run a potentially configured post-stop action"""
        if "post-stop-action" in self.config:
            command = self.config["post-stop-action"]["command"]
            log.debug("post stop action: command: '%s'", command)
            ret_code, data = _execute_shell_command(command)
            if ret_code < 0:
                log.error("post action error! command:%s", command)
                self.result_queue.put({'post-stop-action-data': data})
                return
            log.debug("post-stop data: \n%s", data)
            self.result_queue.put({'post-stop-action-data': data})

    def _run_step(self, cls, method_name, step_cfg, context_cfg):
        raise NotImplementedError

    def run(self, step_cfg, context_cfg):
        step_type = step_cfg["type"]
        class_name = base_step.Step.get(step_type)
        path_split = class_name.split(".")
        module_path = ".".join(path_split[:-1])
        module = importlib.import_module(module_path)
        cls = getattr(module, path_split[-1])

        self.config['object'] = class_name
        self.aborted.clear()

        # run a potentially configured pre-start action
        if "pre-start-action" in self.config:
            command = self.config["pre-start-action"]["command"]
            log.debug("pre start action: command: '%s'", command)
            ret_code, data = _execute_shell_command(command)
            if ret_code < 0:
                log.error("pre-start action error! command:%s", command)
                self.result_queue.put({'pre-start-action-data': data})
                return
            log.debug("pre-start data: \n%s", data)
            self.result_queue.put({'pre-start-action-data': data})

        if "single-shot-action" in self.config:
            single_action_process = multiprocessing.Process(
                target=_single_action,
                name="single-shot-action",
                args=(self.config["single-shot-action"]["after"],
                      self.config["single-shot-action"]["command"],
                      self.result_queue))
            single_action_process.start()

        if "periodic-action" in self.config:
            self.periodic_action_process = multiprocessing.Process(
                target=_periodic_action,
                name="periodic-action",
                args=(self.config["periodic-action"]["interval"],
                      self.config["periodic-action"]["command"],
                      self.result_queue))
            self.periodic_action_process.start()

        self._run_step(cls, "run", step_cfg, context_cfg)

    def abort(self):
        """Abort the execution of a step"""
        self.aborted.set()

    QUEUE_JOIN_INTERVAL = 5

    def poll(self, timeout=QUEUE_JOIN_INTERVAL):
        self.process.join(timeout)
        return self.process.exitcode

    def join(self, outputs, result, interval=QUEUE_JOIN_INTERVAL):
        while self.process.exitcode is None:
            # drain the queue while we are running otherwise we won't terminate
            outputs.update(self.get_output())
            result.extend(self.get_result())
            self.process.join(interval)
        # drain after the process has exited
        outputs.update(self.get_output())
        result.extend(self.get_result())

        self.process.terminate()
        if self.periodic_action_process:
            self.periodic_action_process.join(1)
            self.periodic_action_process.terminate()
            self.periodic_action_process = None

        self.run_post_stop_action()
        return self.process.exitcode

    def get_output(self):
        result = {}
        while not self.output_queue.empty():
            log.debug("output_queue size %s", self.output_queue.qsize())
            try:
                result.update(self.output_queue.get(True, 1))
            except Empty:
                pass
        return result

    def get_result(self):
        result = []
        while not self.result_queue.empty():
            log.debug("result_queue size %s", self.result_queue.qsize())
            try:
                result.append(self.result_queue.get(True, 1))
            except Empty:
                pass
        return result