From 20b87d3ab996bae9afd8b3308965621e124c5218 Mon Sep 17 00:00:00 2001 From: Ravi Jagannadhan Date: Fri, 16 Mar 2018 12:29:24 -0700 Subject: [PATCH 1/3] adding profiler hooks and collection to uploader --- conductor/lib/uploader.py | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/conductor/lib/uploader.py b/conductor/lib/uploader.py index cc9b1044..905e867d 100644 --- a/conductor/lib/uploader.py +++ b/conductor/lib/uploader.py @@ -5,10 +5,14 @@ import os import Queue import sys +import tempfile import thread from threading import Thread +import time import traceback +import yappi # yappi is a profiler specializing in multi-threading + from conductor import CONFIG from conductor.lib import api_client, common, worker, client_db, loggeria @@ -587,6 +591,27 @@ def handle_upload_response(self, project, upload_files, upload_id=None, md5_only def main(self, run_one_loop=False): logger.info('Uploader Started. Checking for uploads...') + # for profiling, grab the environment variable's value + # to determine where to put the profiling data + + if "CONDUCTOR_PROFILE_DATA_PATH" not in os.environ: + profile_path = tempfile.gettempdir() + else: + profile_path = os.environ["CONDUCTOR_PROFILE_DATA_PATH"] + + if not profile_path: + profile_path = tempfile.gettempdir() + + timestr = time.strftime("%Y%m%d-%H%M%S") + func_filename = '{}_func_profile.dmp'.format(timestr) + thread_filename = '{}_thread_profile.txt'.format(timestr) + + self.profile_func_filename = os.path.join(profile_path, func_filename) + self.profile_thread_filename = os.path.join(profile_path, thread_filename) + + # start profiling + yappi.start() + while not common.SIGINT_EXIT: try: # TODO: we should pass args as url params, not http data @@ -629,6 +654,20 @@ def main(self, run_one_loop=False): except KeyboardInterrupt: logger.info("ctrl-c exit") + + # stop profiling here + yappi.stop() + + # There are two collections we want + # the function stats and the thread stats + func_stats = yappi.get_func_stats() + pstats_func = yappi.convert2pstats(func_stats) + pstats_func.dump_stats(self.profile_func_filename) + + thread_stats = yappi.get_thread_stats() + thread_stats_file = open(self.profile_thread_filename, 'w') + thread_stats.print_all(thread_stats_file) + thread_stats_file.close() break except: logger.exception('Caught exception:\n') From a5d4a778682399cdd78de699025988aeedf7824b Mon Sep 17 00:00:00 2001 From: Ravi Jagannadhan Date: Fri, 16 Mar 2018 12:40:42 -0700 Subject: [PATCH 2/3] Removing duplicate import --- conductor/lib/uploader.py | 1 - 1 file changed, 1 deletion(-) diff --git a/conductor/lib/uploader.py b/conductor/lib/uploader.py index 905e867d..fc7e9c99 100644 --- a/conductor/lib/uploader.py +++ b/conductor/lib/uploader.py @@ -1,5 +1,4 @@ import datetime -import time import json import logging import os From 3b993a33b2c8b96e238b011ae2fe22f11abf2feb Mon Sep 17 00:00:00 2001 From: Ravi Jagannadhan Date: Sun, 25 Mar 2018 10:50:52 -0700 Subject: [PATCH 3/3] Adding SIGTERM handler --- conductor/lib/uploader.py | 51 +++++++++++++++++++++++++++------------ 1 file changed, 35 insertions(+), 16 deletions(-) diff --git a/conductor/lib/uploader.py b/conductor/lib/uploader.py index fc7e9c99..6fdc6cf8 100644 --- a/conductor/lib/uploader.py +++ b/conductor/lib/uploader.py @@ -3,6 +3,7 @@ import logging import os import Queue +import signal import sys import tempfile import thread @@ -586,10 +587,7 @@ def handle_upload_response(self, project, upload_files, upload_id=None, md5_only except: return traceback.format_exc() - - def main(self, run_one_loop=False): - logger.info('Uploader Started. Checking for uploads...') - + def profiling_start(self): # for profiling, grab the environment variable's value # to determine where to put the profiling data @@ -609,7 +607,39 @@ def main(self, run_one_loop=False): self.profile_thread_filename = os.path.join(profile_path, thread_filename) # start profiling + logger.info('starting profiling') yappi.start() + return True + + def profiling_stop(self): + logger.info('stopping profiling') + yappi.stop() + + # There are two collections we want + # the function stats and the thread stats + func_stats = yappi.get_func_stats() + pstats_func = yappi.convert2pstats(func_stats) + pstats_func.dump_stats(self.profile_func_filename) + + thread_stats = yappi.get_thread_stats() + thread_stats_file = open(self.profile_thread_filename, 'w') + thread_stats.print_all(thread_stats_file) + thread_stats_file.close() + + def sigterm_handler(self, signal, frame): + logger.info("Caught SIGTERM, exiting...") + self.profiling_stop() + sys.exit(0) + + def main(self, run_one_loop=False): + logger.info('Uploader Started. Checking for uploads...') + + # in some cases, this uploader will be terminated with a SIGTERM + # in which case, we need to exit gracefully via a handler + signal.signal(signal.SIGTERM, self.sigterm_handler) + + # begin profiling + self.profiling_start() while not common.SIGINT_EXIT: try: @@ -655,18 +685,7 @@ def main(self, run_one_loop=False): logger.info("ctrl-c exit") # stop profiling here - yappi.stop() - - # There are two collections we want - # the function stats and the thread stats - func_stats = yappi.get_func_stats() - pstats_func = yappi.convert2pstats(func_stats) - pstats_func.dump_stats(self.profile_func_filename) - - thread_stats = yappi.get_thread_stats() - thread_stats_file = open(self.profile_thread_filename, 'w') - thread_stats.print_all(thread_stats_file) - thread_stats_file.close() + self.profiling_stop() break except: logger.exception('Caught exception:\n')