<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from __future__ import division <EOL> from __future__ import print_function <EOL> import datetime <EOL> import json <EOL> import logging <EOL> import multiprocessing <EOL> import os <EOL> import time <EOL> import phlcon_reviewstatecache <EOL> import phlgitx_refcache <EOL> import phlmp_cyclingpool <EOL> import phlsys_conduit <EOL> import phlsys_fs <EOL> import phlsys_git <EOL> import phlsys_strtotime <EOL> import phlsys_subprocess <EOL> import phlsys_timer <EOL> import phlurl_watcher <EOL> import abdmail_mailer <EOL> import abdt_classicnaming <EOL> import abdt_compositenaming <EOL> import abdt_conduit <EOL> import abdt_differresultcache <EOL> import abdt_errident <EOL> import abdt_exhandlers <EOL> import abdt_fs <EOL> import abdt_git <EOL> import abdt_logging <EOL> import abdt_rbranchnaming <EOL> import abdt_tryloop <EOL> import abdi_processexitcodes <EOL> import abdi_processrepo <EOL> import abdi_repoargs <EOL> _LOGGER = logging . getLogger ( __name__ ) <EOL> def do ( <EOL> repo_configs , <EOL> sys_admin_emails , <EOL> sleep_secs , <EOL> is_no_loop , <EOL> external_report_command , <EOL> mail_sender , <EOL> max_workers , <EOL> overrun_secs ) : <EOL> conduit_manager = _ConduitManager ( ) <EOL> fs_accessor = abdt_fs . make_default_accessor ( ) <EOL> url_watcher_wrapper = phlurl_watcher . FileCacheWatcherWrapper ( <EOL> fs_accessor . layout . urlwatcher_cache_path ) <EOL> if max_workers == <NUM_LIT:0> : <EOL> max_workers = determine_max_workers_default ( ) <EOL> repo_list = [ ] <EOL> for name , config in repo_configs : <EOL> repo_list . append ( <EOL> _ArcydManagedRepository ( <EOL> name , <EOL> config , <EOL> conduit_manager , <EOL> url_watcher_wrapper , <EOL> sys_admin_emails , <EOL> mail_sender ) ) <EOL> max_overrun_workers = max_workers // <NUM_LIT:2> <EOL> pool = phlmp_cyclingpool . CyclingPool ( <EOL> repo_list , max_workers , max_overrun_workers ) <EOL> cycle_timer = phlsys_timer . Timer ( ) <EOL> cycle_timer . start ( ) <EOL> exit_code = None <EOL> while exit_code is None : <EOL> sleep_timer = phlsys_timer . Timer ( ) <EOL> sleep_timer . start ( ) <EOL> with abdt_logging . remote_io_read_event_context ( <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> abdt_tryloop . critical_tryloop ( <EOL> url_watcher_wrapper . watcher . refresh , <EOL> abdt_errident . GIT_SNOOP , <EOL> '<STR_LIT>' ) <EOL> with abdt_logging . remote_io_read_event_context ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> conduit_manager . refresh_conduits ( ) <EOL> with abdt_logging . misc_operation_event_context ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' . format ( max_workers , len ( repo_list ) ) ) : <EOL> if max_workers > <NUM_LIT:1> : <EOL> for i , res in pool . cycle_results ( overrun_secs = overrun_secs ) : <EOL> repo = repo_list [ i ] <EOL> repo . merge_from_worker ( res ) <EOL> else : <EOL> for r in repo_list : <EOL> r ( ) <EOL> url_watcher_wrapper . save ( ) <EOL> report = { <EOL> "<STR_LIT>" : cycle_timer . restart ( ) , <EOL> "<STR_LIT>" : pool . num_active_jobs , <EOL> } <EOL> _LOGGER . debug ( "<STR_LIT>" . format ( report ) ) <EOL> if external_report_command : <EOL> report_json = json . dumps ( report ) <EOL> full_path = os . path . abspath ( external_report_command ) <EOL> with abdt_logging . misc_operation_event_context ( <EOL> '<STR_LIT>' , external_report_command ) : <EOL> try : <EOL> phlsys_subprocess . run ( full_path , stdin = report_json ) <EOL> except phlsys_subprocess . CalledProcessError as e : <EOL> _LOGGER . error ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" . format ( <EOL> external_report_command , type ( e ) . __name__ ) ) <EOL> _LOGGER . error ( "<STR_LIT>" . format ( <EOL> e ) ) <EOL> if is_no_loop : <EOL> exit_code = abdi_processexitcodes . ExitCodes . ec_exit <EOL> elif os . path . isfile ( fs_accessor . layout . killfile ) : <EOL> exit_code = abdi_processexitcodes . ExitCodes . ec_exit <EOL> if phlsys_fs . read_text_file ( fs_accessor . layout . killfile ) : <EOL> _LOGGER . info ( "<STR_LIT>" . format ( <EOL> phlsys_fs . read_text_file ( fs_accessor . layout . killfile ) ) ) <EOL> else : <EOL> _LOGGER . info ( "<STR_LIT>" ) <EOL> os . remove ( fs_accessor . layout . killfile ) <EOL> elif os . path . isfile ( fs_accessor . layout . reloadfile ) : <EOL> _LOGGER . info ( "<STR_LIT>" ) <EOL> exit_code = abdi_processexitcodes . ExitCodes . ec_reload <EOL> os . remove ( fs_accessor . layout . reloadfile ) <EOL> secs_to_sleep = float ( sleep_secs ) - float ( sleep_timer . duration ) <EOL> if secs_to_sleep > <NUM_LIT:0> and exit_code is None : <EOL> with abdt_logging . misc_operation_event_context ( <EOL> '<STR_LIT>' , secs_to_sleep ) : <EOL> time . sleep ( secs_to_sleep ) <EOL> for i , res in pool . finish_results ( ) : <EOL> repo = repo_list [ i ] <EOL> repo . merge_from_worker ( res ) <EOL> url_watcher_wrapper . save ( ) <EOL> return exit_code <EOL> def determine_max_workers_default ( ) : <EOL> max_workers = <NUM_LIT:1> <EOL> try : <EOL> max_workers = multiprocessing . cpu_count ( ) <EOL> _LOGGER . debug ( <EOL> "<STR_LIT>" . format ( <EOL> max_workers ) ) <EOL> except NotImplementedError : <EOL> _LOGGER . warning ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return max_workers <EOL> class _RecordingWatcherWrapper ( object ) : <EOL> def __init__ ( self , watcher ) : <EOL> self . _watcher = watcher <EOL> self . _tested_urls = set ( ) <EOL> def peek_has_url_recently_changed ( self , url ) : <EOL> return self . _watcher . peek_has_url_recently_changed ( url ) <EOL> def has_url_recently_changed ( self , url ) : <EOL> self . _tested_urls . add ( url ) <EOL> return self . _watcher . has_url_recently_changed ( url ) <EOL> def get_data_for_merging ( self ) : <EOL> data = self . _watcher . get_data_for_merging ( ) <EOL> tested = self . _tested_urls <EOL> new_data = { k : v for k , v in data . iteritems ( ) if k in tested } <EOL> return new_data <EOL> class _RepoActiveRetryState ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , retry_timestr_list ) : <EOL> self . _is_active = True <EOL> self . _reactivate_time = None <EOL> self . _retry_delays = [ <EOL> phlsys_strtotime . duration_string_to_time_delta ( s ) <EOL> for s in retry_timestr_list <EOL> ] <EOL> def calc_active ( self ) : <EOL> if not self . _is_active and self . _reactivate_time is not None : <EOL> if datetime . datetime . utcnow ( ) >= self . _reactivate_time : <EOL> self . _is_active = True <EOL> self . _reactivate_time = None <EOL> return self . _is_active <EOL> @ property <EOL> def is_active ( self ) : <EOL> return self . _is_active <EOL> def disable ( self ) : <EOL> self . _is_active = False <EOL> retry_delay = None <EOL> if self . _retry_delays : <EOL> retry_delay = self . _retry_delays . pop ( <NUM_LIT:0> ) <EOL> self . _reactivate_time = datetime . datetime . utcnow ( ) + retry_delay <EOL> else : <EOL> self . _reactivate_time = None <EOL> return retry_delay <EOL> @ property <EOL> def reactivate_time ( self ) : <EOL> return self . _reactivate_time <EOL> class _ArcydManagedRepository ( object ) : <EOL> def __init__ ( <EOL> self , <EOL> repo_name , <EOL> repo_args , <EOL> conduit_manager , <EOL> url_watcher_wrapper , <EOL> sys_admin_emails , <EOL> mail_sender ) : <EOL> self . _active_state = _RepoActiveRetryState ( <EOL> retry_timestr_list = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> sys_repo = phlsys_git . Repo ( repo_args . repo_path ) <EOL> self . _refcache_repo = phlgitx_refcache . Repo ( sys_repo ) <EOL> self . _differ_cache = abdt_differresultcache . Cache ( self . _refcache_repo ) <EOL> self . _abd_repo = abdt_git . Repo ( <EOL> self . _refcache_repo , <EOL> self . _differ_cache , <EOL> "<STR_LIT>" , <EOL> repo_args . repo_desc ) <EOL> self . _name = repo_name <EOL> self . _args = repo_args <EOL> self . _conduit_manager = conduit_manager <EOL> conduit_cache = conduit_manager . get_conduit_and_cache_for_args ( <EOL> repo_args ) <EOL> self . _arcyd_conduit , self . _review_cache = conduit_cache <EOL> self . _mail_sender = mail_sender <EOL> self . _url_watcher_wrapper = url_watcher_wrapper <EOL> self . _mail_sender = mail_sender <EOL> self . _on_exception = abdt_exhandlers . make_exception_delay_handler ( <EOL> sys_admin_emails , repo_name ) <EOL> def __call__ ( self ) : <EOL> watcher = _RecordingWatcherWrapper ( <EOL> self . _url_watcher_wrapper . watcher ) <EOL> old_active_reviews = set ( self . _review_cache . active_reviews ) <EOL> was_active = self . _active_state . is_active <EOL> if self . _active_state . calc_active ( ) : <EOL> if not was_active : <EOL> _LOGGER . info ( <EOL> '<STR_LIT>' . format ( self . _name ) ) <EOL> try : <EOL> _process_repo ( <EOL> self . _abd_repo , <EOL> self . _name , <EOL> self . _args , <EOL> self . _arcyd_conduit , <EOL> watcher , <EOL> self . _mail_sender ) <EOL> except Exception : <EOL> retry_delay = self . _active_state . disable ( ) <EOL> _LOGGER . info ( <EOL> '<STR_LIT>' . format ( <EOL> self . _name , retry_delay ) ) <EOL> self . _on_exception ( retry_delay ) <EOL> else : <EOL> _LOGGER . debug ( <EOL> '<STR_LIT>' . format ( <EOL> self . _name , self . _active_state . reactivate_time ) ) <EOL> return ( <EOL> self . _review_cache . active_reviews - old_active_reviews , <EOL> self . _active_state , <EOL> watcher . get_data_for_merging ( ) , <EOL> self . _refcache_repo . peek_hash_ref_pairs ( ) , <EOL> self . _differ_cache . get_cache ( ) <EOL> ) <EOL> def merge_from_worker ( self , results ) : <EOL> ( <EOL> active_reviews , <EOL> active_state , <EOL> watcher_data , <EOL> hash_ref_pairs , <EOL> differ_cache <EOL> ) = results <EOL> self . _review_cache . merge_additional_active_reviews ( active_reviews ) <EOL> self . _active_state = active_state <EOL> self . _refcache_repo . set_hash_ref_pairs ( hash_ref_pairs ) <EOL> self . _differ_cache . set_cache ( differ_cache ) <EOL> self . _url_watcher_wrapper . watcher . merge_data_consume_only ( watcher_data ) <EOL> class _ConduitManager ( object ) : <EOL> def __init__ ( self ) : <EOL> super ( _ConduitManager , self ) . __init__ ( ) <EOL> self . _conduits_caches = { } <EOL> def get_conduit_and_cache_for_args ( self , args ) : <EOL> key = ( <EOL> args . instance_uri , <EOL> args . arcyd_user , <EOL> args . arcyd_cert , <EOL> args . https_proxy <EOL> ) <EOL> if key not in self . _conduits_caches : <EOL> conduit = [ None ] <EOL> def connect ( ) : <EOL> conduit [ <NUM_LIT:0> ] = phlsys_conduit . MultiConduit ( <EOL> args . instance_uri , <EOL> args . arcyd_user , <EOL> args . arcyd_cert , <EOL> https_proxy = args . https_proxy ) <EOL> abdt_tryloop . tryloop ( <EOL> connect , abdt_errident . CONDUIT_CONNECT , args . instance_uri ) <EOL> multi_conduit = conduit [ <NUM_LIT:0> ] <EOL> cache = phlcon_reviewstatecache . make_from_conduit ( multi_conduit ) <EOL> arcyd_conduit = abdt_conduit . Conduit ( multi_conduit , cache ) <EOL> self . _conduits_caches [ key ] = ( arcyd_conduit , cache ) <EOL> else : <EOL> arcyd_conduit , cache = self . _conduits_caches [ key ] <EOL> return arcyd_conduit , cache <EOL> def refresh_conduits ( self ) : <EOL> for conduit , cache in self . _conduits_caches . itervalues ( ) : <EOL> abdt_tryloop . critical_tryloop ( <EOL> cache . refresh_active_reviews , <EOL> abdt_errident . CONDUIT_REFRESH , <EOL> conduit . describe ( ) ) <EOL> def fetch_if_needed ( url_watcher , snoop_url , repo , repo_desc ) : <EOL> did_fetch = False <EOL> if not snoop_url or url_watcher . peek_has_url_recently_changed ( snoop_url ) : <EOL> abdt_tryloop . tryloop ( <EOL> repo . checkout_master_fetch_prune , <EOL> abdt_errident . FETCH_PRUNE , <EOL> repo_desc ) <EOL> did_fetch = True <EOL> if did_fetch and snoop_url : <EOL> url_watcher . has_url_recently_changed ( snoop_url ) <EOL> return did_fetch <EOL> def _process_repo ( <EOL> repo , <EOL> unused_repo_name , <EOL> args , <EOL> arcyd_conduit , <EOL> url_watcher , <EOL> mail_sender ) : <EOL> fetch_if_needed ( <EOL> url_watcher , <EOL> abdi_repoargs . get_repo_snoop_url ( args ) , <EOL> repo , <EOL> args . repo_desc ) <EOL> admin_emails = set ( _flatten_list ( args . admin_emails ) ) <EOL> mailer = abdmail_mailer . Mailer ( <EOL> mail_sender , <EOL> admin_emails , <EOL> args . repo_desc , <EOL> args . instance_uri ) <EOL> branch_url_callable = None <EOL> if args . branch_url_format : <EOL> def make_branch_url ( branch_name ) : <EOL> return args . branch_url_format . format ( <EOL> branch = branch_name , <EOL> repo_url = args . repo_url ) <EOL> branch_url_callable = make_branch_url <EOL> branch_naming = abdt_compositenaming . Naming ( <EOL> abdt_classicnaming . Naming ( ) , <EOL> abdt_rbranchnaming . Naming ( ) ) <EOL> branches = abdt_git . get_managed_branches ( <EOL> repo , <EOL> args . repo_desc , <EOL> branch_naming , <EOL> branch_url_callable ) <EOL> abdi_processrepo . process_branches ( branches , arcyd_conduit , mailer ) <EOL> def _flatten_list ( hierarchy ) : <EOL> for x in hierarchy : <EOL> if hasattr ( x , '<STR_LIT>' ) and not isinstance ( x , str ) : <EOL> for y in _flatten_list ( x ) : <EOL> yield y <EOL> else : <EOL> yield x </s>
