X-Git-Url: https://gerrit.fd.io/r/gitweb?p=csit.git;a=blobdiff_plain;f=resources%2Ftools%2Fpresentation%2Finput_data_parser.py;h=bae7b5eead9da385d62a53d9c0efc593b2e068eb;hp=9c0e38073ca2f49e190f519e4b837c3a0d90c7a2;hb=d59cb151a877c88b4a4fac28c1e01fa5220c1fb6;hpb=cee3ad0f9cc29ffc67d9c87c58920252671beb21 diff --git a/resources/tools/presentation/input_data_parser.py b/resources/tools/presentation/input_data_parser.py index 9c0e38073c..bae7b5eead 100644 --- a/resources/tools/presentation/input_data_parser.py +++ b/resources/tools/presentation/input_data_parser.py @@ -19,8 +19,6 @@ - filter the data using tags, """ -import multiprocessing -import os import re import resource import pandas as pd @@ -38,7 +36,6 @@ from json import loads from jumpavg.AvgStdevMetadataFactory import AvgStdevMetadataFactory from input_data_files import download_and_unzip_data_file -from utils import Worker # Separator used in file names @@ -390,14 +387,12 @@ class ExecutionChecker(ResultVisitor): :returns: Nothing. """ - if msg.message.count("Arguments:"): - message = str(msg.message).replace(' ', '').replace('\n', '').\ - replace("'", '"').replace('b"', '"').\ - replace("honeycom", "honeycomb") - message = loads(message[11:-1]) + if msg.message.count("Setup of TG node"): + reg_tg_ip = re.compile( + r'Setup of TG node (\d{1,3}.\d{1,3}.\d{1,3}.\d{1,3}) done') try: - self._testbed = message["TG"]["host"] - except (KeyError, ValueError): + self._testbed = str(re.search(reg_tg_ip, msg.message).group(1)) + except (KeyError, ValueError, IndexError, AttributeError): pass finally: self._data["metadata"]["testbed"] = self._testbed @@ -1188,13 +1183,10 @@ class InputData(object): return checker.data - def _download_and_parse_build(self, pid, data_queue, job, build, repeat): + def _download_and_parse_build(self, job, build, repeat, pid=10000): """Download and parse the input data file. :param pid: PID of the process executing this method. - :param data_queue: Shared memory between processes. Queue which keeps - the result data. This data is then read by the main process and used - in further processing. :param job: Name of the Jenkins job which generated the processed input file. :param build: Information about the Jenkins build which generated the @@ -1202,7 +1194,6 @@ class InputData(object): :param repeat: Repeat the download specified number of times if not successful. :type pid: int - :type data_queue: multiprocessing.Manager().Queue() :type job: str :type build: dict :type repeat: int @@ -1283,14 +1274,6 @@ class InputData(object): format(full_name, repr(err)))) logs.append(("INFO", " Done.")) - result = { - "data": data, - "state": state, - "job": job, - "build": build - } - data_queue.put(result) - for level, line in logs: if level == "INFO": logging.info(line) @@ -1303,8 +1286,7 @@ class InputData(object): elif level == "WARNING": logging.warning(line) - logging.info("Memory allocation: {0:,d}MB".format( - resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000)) + return {"data": data, "state": state, "job": job, "build": build} def download_and_parse_data(self, repeat=1): """Download the input data files, parse input data from input files and @@ -1317,67 +1299,34 @@ class InputData(object): logging.info("Downloading and parsing input files ...") - work_queue = multiprocessing.JoinableQueue() - manager = multiprocessing.Manager() - data_queue = manager.Queue() - cpus = multiprocessing.cpu_count() - - workers = list() - for cpu in range(cpus): - worker = Worker(work_queue, - data_queue, - self._download_and_parse_build) - worker.daemon = True - worker.start() - workers.append(worker) - os.system("taskset -p -c {0} {1} > /dev/null 2>&1". - format(cpu, worker.pid)) - for job, builds in self._cfg.builds.items(): for build in builds: - work_queue.put((job, build, repeat)) - - work_queue.join() - - logging.info("Done.") - logging.info("Collecting data:") - - while not data_queue.empty(): - result = data_queue.get() - - job = result["job"] - build_nr = result["build"]["build"] - logging.info(" {job}-{build}".format(job=job, build=build_nr)) - - if result["data"]: - data = result["data"] - build_data = pd.Series({ - "metadata": pd.Series( - data["metadata"].values(), - index=data["metadata"].keys()), - "suites": pd.Series(data["suites"].values(), - index=data["suites"].keys()), - "tests": pd.Series(data["tests"].values(), - index=data["tests"].keys())}) - if self._input_data.get(job, None) is None: - self._input_data[job] = pd.Series() - self._input_data[job][str(build_nr)] = build_data + result = self._download_and_parse_build(job, build, repeat) + build_nr = result["build"]["build"] - self._cfg.set_input_file_name( - job, build_nr, result["build"]["file-name"]) + if result["data"]: + data = result["data"] + build_data = pd.Series({ + "metadata": pd.Series( + data["metadata"].values(), + index=data["metadata"].keys()), + "suites": pd.Series(data["suites"].values(), + index=data["suites"].keys()), + "tests": pd.Series(data["tests"].values(), + index=data["tests"].keys())}) - self._cfg.set_input_state(job, build_nr, result["state"]) + if self._input_data.get(job, None) is None: + self._input_data[job] = pd.Series() + self._input_data[job][str(build_nr)] = build_data - logging.info("Memory allocation: {0:,d}MB".format( - resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000)) + self._cfg.set_input_file_name( + job, build_nr, result["build"]["file-name"]) - del data_queue + self._cfg.set_input_state(job, build_nr, result["state"]) - # Terminate all workers - for worker in workers: - worker.terminate() - worker.join() + logging.info("Memory allocation: {0:,d}MB".format( + resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1000)) logging.info("Done.")