diff options
| -rw-r--r-- | resources/tools/presentation/input_data_parser.py | 136 | ||||
| -rw-r--r-- | resources/tools/presentation/specification.yaml | 1 |
2 files changed, 47 insertions, 90 deletions
diff --git a/resources/tools/presentation/input_data_parser.py b/resources/tools/presentation/input_data_parser.py index 7b36352ed9..71035adf02 100644 --- a/resources/tools/presentation/input_data_parser.py +++ b/resources/tools/presentation/input_data_parser.py @@ -19,9 +19,8 @@ - filter the data using tags, """ -import multiprocessing -import os import re +import resource import pandas as pd import logging @@ -37,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 @@ -1160,13 +1158,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 @@ -1174,7 +1169,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 @@ -1182,9 +1176,6 @@ class InputData(object): logs = list() - logging.info(" Processing the job/build: {0}: {1}". - format(job, build["build"])) - logs.append(("INFO", " Processing the job/build: {0}: {1}". format(job, build["build"]))) @@ -1204,7 +1195,7 @@ class InputData(object): "'{build}', or it is damaged. Skipped.". format(job=job, build=build["build"]))) if success: - logs.append(("INFO", " Processing data from the build '{0}' ...". + logs.append(("INFO", " Processing data from the build '{0}' ...". format(build["build"]))) data = self._parse_tests(job, build, logs) if data is None: @@ -1242,11 +1233,11 @@ class InputData(object): file_name = self._cfg.input["file-name"] full_name = join( self._cfg.environment["paths"]["DIR[WORKING,DATA]"], - "{job}{sep}{build}{sep}{name}". - format(job=job, - sep=SEPARATOR, - build=build["build"], - name=file_name)) + "{job}{sep}{build}{sep}{name}".format( + job=job, + sep=SEPARATOR, + build=build["build"], + name=file_name)) try: remove(full_name) logs.append(("INFO", @@ -1254,19 +1245,23 @@ class InputData(object): format(name=full_name))) except OSError as err: logs.append(("ERROR", - "Cannot remove the file '{0}': {1}". - format(full_name, repr(err)))) - + "Cannot remove the file '{0}': {1}". + format(full_name, repr(err)))) logs.append(("INFO", " Done.")) - result = { - "data": data, - "state": state, - "job": job, - "build": build, - "logs": logs - } - data_queue.put(result) + for level, line in logs: + if level == "INFO": + logging.info(line) + elif level == "ERROR": + logging.error(line) + elif level == "DEBUG": + logging.debug(line) + elif level == "CRITICAL": + logging.critical(line) + elif level == "WARNING": + logging.warning(line) + + 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 @@ -1279,73 +1274,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() + result = self._download_and_parse_build(job, build, repeat) + build_nr = result["build"]["build"] - logging.info("Done.") + 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 + + self._cfg.set_input_file_name( + job, build_nr, result["build"]["file-name"]) + + self._cfg.set_input_state(job, build_nr, result["state"]) - while not data_queue.empty(): - result = data_queue.get() - - job = result["job"] - build_nr = result["build"]["build"] - - 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 - - self._cfg.set_input_file_name(job, build_nr, - result["build"]["file-name"]) - - self._cfg.set_input_state(job, build_nr, result["state"]) - - for item in result["logs"]: - if item[0] == "INFO": - logging.info(item[1]) - elif item[0] == "ERROR": - logging.error(item[1]) - elif item[0] == "DEBUG": - logging.debug(item[1]) - elif item[0] == "CRITICAL": - logging.critical(item[1]) - elif item[0] == "WARNING": - logging.warning(item[1]) - - del data_queue - - # 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.") diff --git a/resources/tools/presentation/specification.yaml b/resources/tools/presentation/specification.yaml index 548666bf5a..ba05649fa8 100644 --- a/resources/tools/presentation/specification.yaml +++ b/resources/tools/presentation/specification.yaml @@ -76,6 +76,7 @@ URL[JENKINS,CSIT]: "https://jenkins.fd.io/view/csit/job" URL[JENKINS,HC]: "https://jenkins.fd.io/view/hc2vpp/job" URL[VIRL,DNV]: "http://10.30.51.28/download/denverton/{release}/robot-plugin.zip" + URL[NEXUS,LOG]: "https://logs.fd.io/production/vex-yul-rot-jenkins-1" # URL[VIRL,DNV]: "http://download.it/from/nexus" URL[NEXUS]: "https://docs.fd.io/csit" DIR[NEXUS]: "report/_static/archive" |