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
|
