<s> import multiprocessing <EOL> import Queue <EOL> from datetime import datetime , timedelta <EOL> import zlib <EOL> import time <EOL> from paste . deploy import appconfig <EOL> from contextlib import contextmanager <EOL> import os <EOL> import errno <EOL> import fcntl <EOL> import sys <EOL> import traceback <EOL> import cPickle <EOL> import cStringIO <EOL> from eventlet import sleep <EOL> from swift . common . memcached import MemcacheRing <EOL> from slogging . internal_proxy import InternalProxy <EOL> from swift . common . utils import get_logger <EOL> from swift . common . exceptions import ChunkReadTimeout , LockTimeout <EOL> class BadFileDownload ( Exception ) : <EOL> def __init__ ( self , status_code = None ) : <EOL> self . status_code = status_code <EOL> class WorkerError ( Exception ) : <EOL> def __init__ ( self ) : <EOL> self . tb_str = '<STR_LIT>' <EOL> class LogProcessorCommon ( object ) : <EOL> def __init__ ( self , conf , logger , log_route = '<STR_LIT>' ) : <EOL> if isinstance ( logger , tuple ) : <EOL> self . logger = get_logger ( * logger , log_route = log_route ) <EOL> else : <EOL> self . logger = logger <EOL> self . memcache = MemcacheRing ( [ s . strip ( ) for s in <EOL> conf . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( '<STR_LIT:U+002C>' ) <EOL> if s . strip ( ) ] ) <EOL> self . conf = conf <EOL> self . _internal_proxy = None <EOL> self . lookback_hours = int ( conf . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . lookback_window = int ( conf . get ( '<STR_LIT>' , <EOL> self . lookback_hours ) ) <EOL> self . log_processor_account = conf [ '<STR_LIT>' ] <EOL> self . log_processor_container = conf . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . processed_files_object_name = conf . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> @ property <EOL> def internal_proxy ( self ) : <EOL> if self . _internal_proxy is None : <EOL> proxy_server_conf_loc = self . conf . get ( '<STR_LIT>' ) <EOL> if proxy_server_conf_loc is None : <EOL> stats_conf = self . conf . get ( '<STR_LIT>' , { } ) <EOL> proxy_server_conf_loc = stats_conf . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> if proxy_server_conf_loc : <EOL> proxy_server_conf = appconfig ( <EOL> '<STR_LIT>' % proxy_server_conf_loc , <EOL> name = '<STR_LIT>' ) <EOL> else : <EOL> proxy_server_conf = None <EOL> self . _internal_proxy = InternalProxy ( proxy_server_conf , <EOL> self . logger , <EOL> retries = <NUM_LIT:3> , <EOL> memcache = self . memcache ) <EOL> return self . _internal_proxy <EOL> def calculate_lookback ( self ) : <EOL> if self . lookback_hours == <NUM_LIT:0> : <EOL> lookback_start = None <EOL> lookback_end = None <EOL> else : <EOL> delta_hours = timedelta ( hours = self . lookback_hours ) <EOL> lookback_start = datetime . now ( ) - delta_hours <EOL> lookback_start = lookback_start . strftime ( '<STR_LIT>' ) <EOL> if self . lookback_window == <NUM_LIT:0> : <EOL> lookback_end = None <EOL> else : <EOL> delta_window = timedelta ( hours = self . lookback_window ) <EOL> lookback_end = datetime . now ( ) - delta_hours + delta_window <EOL> lookback_end = lookback_end . strftime ( '<STR_LIT>' ) <EOL> self . logger . debug ( '<STR_LIT>' % lookback_start ) <EOL> self . logger . debug ( '<STR_LIT>' % lookback_end ) <EOL> return ( lookback_start , lookback_end ) <EOL> def load_already_processed_files ( self ) : <EOL> try : <EOL> processed_files_stream = self . get_object_data ( <EOL> self . log_processor_account , <EOL> self . log_processor_container , <EOL> self . processed_files_object_name , <EOL> compressed = True ) <EOL> buf = '<STR_LIT:\n>' . join ( x for x in processed_files_stream ) <EOL> if buf : <EOL> already_processed_files = cPickle . loads ( buf ) <EOL> else : <EOL> already_processed_files = set ( ) <EOL> except BadFileDownload , err : <EOL> if err . status_code == <NUM_LIT> : <EOL> already_processed_files = set ( ) <EOL> else : <EOL> self . logger . error ( _ ( '<STR_LIT>' <EOL> '<STR_LIT>' ) ) <EOL> return <EOL> self . logger . debug ( _ ( '<STR_LIT>' ) % len ( already_processed_files ) ) <EOL> return already_processed_files <EOL> def save_processed_files ( self , processed_files_data ) : <EOL> s = cPickle . dumps ( processed_files_data , cPickle . HIGHEST_PROTOCOL ) <EOL> f = cStringIO . StringIO ( s ) <EOL> return self . internal_proxy . upload_file ( f , self . log_processor_account , <EOL> self . log_processor_container , <EOL> self . processed_files_object_name ) <EOL> def get_object_data ( self , swift_account , container_name , object_name , <EOL> compressed = False ) : <EOL> '''<STR_LIT>''' <EOL> self . logger . debug ( '<STR_LIT>' % <EOL> ( swift_account , <EOL> container_name , <EOL> object_name , <EOL> compressed ) ) <EOL> code , o = self . internal_proxy . get_object ( swift_account , container_name , <EOL> object_name ) <EOL> if code < <NUM_LIT:200> or code >= <NUM_LIT> : <EOL> raise BadFileDownload ( code ) <EOL> last_part = '<STR_LIT>' <EOL> last_compressed_part = '<STR_LIT>' <EOL> d = zlib . decompressobj ( <NUM_LIT:16> + zlib . MAX_WBITS ) <EOL> try : <EOL> for chunk in o : <EOL> if compressed : <EOL> try : <EOL> chunk = d . decompress ( chunk ) <EOL> except zlib . error : <EOL> self . logger . debug ( _ ( '<STR_LIT>' ) <EOL> % '<STR_LIT:/>' . join ( ( swift_account , container_name , <EOL> object_name ) ) ) <EOL> raise BadFileDownload ( ) <EOL> parts = chunk . split ( '<STR_LIT:\n>' ) <EOL> parts [ <NUM_LIT:0> ] = last_part + parts [ <NUM_LIT:0> ] <EOL> for part in parts [ : - <NUM_LIT:1> ] : <EOL> yield part <EOL> last_part = parts [ - <NUM_LIT:1> ] <EOL> if last_part : <EOL> yield last_part <EOL> except ChunkReadTimeout : <EOL> raise BadFileDownload ( ) <EOL> def get_container_listing ( self , swift_account , container_name , <EOL> start_date = None , end_date = None , <EOL> listing_filter = None ) : <EOL> '''<STR_LIT>''' <EOL> search_key = None <EOL> if start_date is not None : <EOL> try : <EOL> parsed_date = time . strptime ( start_date , '<STR_LIT>' ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> year = '<STR_LIT>' % parsed_date . tm_year <EOL> month = '<STR_LIT>' % parsed_date . tm_mon <EOL> day = '<STR_LIT>' % parsed_date . tm_mday <EOL> hour = '<STR_LIT>' % parsed_date . tm_hour <EOL> search_key = '<STR_LIT:/>' . join ( [ year , month , day , hour ] ) <EOL> end_key = None <EOL> if end_date is not None : <EOL> try : <EOL> parsed_date = time . strptime ( end_date , '<STR_LIT>' ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> year = '<STR_LIT>' % parsed_date . tm_year <EOL> month = '<STR_LIT>' % parsed_date . tm_mon <EOL> day = '<STR_LIT>' % parsed_date . tm_mday <EOL> hour = '<STR_LIT>' % ( parsed_date . tm_hour + <NUM_LIT:1> ) <EOL> end_key = '<STR_LIT:/>' . join ( [ year , month , day , hour ] ) <EOL> container_listing = self . internal_proxy . get_container_list ( <EOL> swift_account , <EOL> container_name , <EOL> marker = search_key , <EOL> end_marker = end_key ) <EOL> results = [ ] <EOL> if listing_filter is None : <EOL> listing_filter = set ( ) <EOL> for item in container_listing : <EOL> name = item [ '<STR_LIT:name>' ] <EOL> if name not in listing_filter : <EOL> results . append ( name ) <EOL> return results <EOL> def multiprocess_collate ( processor_klass , processor_args , processor_method , <EOL> items_to_process , worker_count , logger = None ) : <EOL> '''<STR_LIT:U+0020>''' <EOL> results = [ ] <EOL> in_queue = multiprocessing . Queue ( ) <EOL> out_queue = multiprocessing . Queue ( ) <EOL> for _junk in range ( worker_count ) : <EOL> p = multiprocessing . Process ( target = collate_worker , <EOL> args = ( processor_klass , <EOL> processor_args , <EOL> processor_method , <EOL> in_queue , <EOL> out_queue , <EOL> logger ) ) <EOL> p . start ( ) <EOL> results . append ( p ) <EOL> for x in items_to_process : <EOL> in_queue . put ( x ) <EOL> for _junk in range ( worker_count ) : <EOL> in_queue . put ( None ) <EOL> in_queue . close ( ) <EOL> while True : <EOL> try : <EOL> item , data = out_queue . get_nowait ( ) <EOL> except Queue . Empty : <EOL> sleep ( <NUM_LIT> ) <EOL> except Exception : <EOL> if logger : <EOL> logger . exception ( '<STR_LIT>' ) <EOL> else : <EOL> if isinstance ( data , WorkerError ) : <EOL> if logger : <EOL> logger . error ( data . tb_str ) <EOL> else : <EOL> yield item , data <EOL> if not any ( r . is_alive ( ) for r in results ) and out_queue . empty ( ) : <EOL> break <EOL> def collate_worker ( processor_klass , processor_args , processor_method , in_queue , <EOL> out_queue , logger = None ) : <EOL> '''<STR_LIT>''' <EOL> try : <EOL> p = processor_klass ( * processor_args ) <EOL> while True : <EOL> item = in_queue . get ( ) <EOL> if item is None : <EOL> break <EOL> try : <EOL> method = getattr ( p , processor_method ) <EOL> except AttributeError : <EOL> return <EOL> try : <EOL> ret = method ( * item ) <EOL> except : <EOL> err_type , err , tb = sys . exc_info ( ) <EOL> ret = WorkerError ( ) <EOL> ret . tb_str = '<STR_LIT>' . join ( traceback . format_tb ( tb ) ) <EOL> out_queue . put ( ( item , ret ) ) <EOL> except Exception , err : <EOL> if logger : <EOL> logger . exception ( '<STR_LIT>' ) <EOL> finally : <EOL> in_queue . close ( ) <EOL> out_queue . close ( ) </s>
