<s> import time <EOL> import collections <EOL> import datetime <EOL> from uuid import uuid4 <EOL> import Queue <EOL> from urllib import unquote <EOL> import os <EOL> import cPickle <EOL> import cStringIO <EOL> import functools <EOL> import random <EOL> import errno <EOL> from swift . common . daemon import Daemon <EOL> from swift . common . utils import get_logger , TRUE_VALUES , split_path <EOL> from swift . common . exceptions import LockTimeout , ChunkReadTimeout <EOL> from slogging . log_common import LogProcessorCommon , multiprocess_collate , BadFileDownload <EOL> from slogging . file_buffer import FileBuffer <EOL> month_map = '<STR_LIT>' . split ( ) <EOL> MEMOIZE_KEY_LIMIT = <NUM_LIT> <EOL> MEMOIZE_FLUSH_RATE = <NUM_LIT> <EOL> def make_clf_from_parts ( parts ) : <EOL> format = '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' '<STR_LIT>' <EOL> try : <EOL> return format % parts <EOL> except KeyError : <EOL> return None <EOL> def memoize ( func ) : <EOL> cache = { } <EOL> @ functools . wraps ( func ) <EOL> def wrapped ( * args ) : <EOL> key = tuple ( args ) <EOL> if key in cache : <EOL> return cache [ key ] <EOL> result = func ( * args ) <EOL> len_cache = len ( cache ) <EOL> if len_cache > MEMOIZE_KEY_LIMIT : <EOL> cache_keys = cache . keys ( ) <EOL> for _unused in xrange ( int ( len_cache * MEMOIZE_FLUSH_RATE ) ) : <EOL> index_to_delete = random . randrange ( <NUM_LIT:0> , len ( cache_keys ) ) <EOL> key_to_delete = cache_keys . pop ( index_to_delete ) <EOL> del cache [ key_to_delete ] <EOL> cache [ key ] = result <EOL> return result <EOL> return wrapped <EOL> class AccessLogDelivery ( LogProcessorCommon ) : <EOL> def __init__ ( self , conf , logger ) : <EOL> super ( AccessLogDelivery , self ) . __init__ ( conf , logger , <EOL> '<STR_LIT>' ) <EOL> self . frequency = int ( conf . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . metadata_key = conf . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) . lower ( ) <EOL> self . server_name = conf . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . working_dir = conf . get ( '<STR_LIT>' , '<STR_LIT>' ) . rstrip ( '<STR_LIT:/>' ) <EOL> buffer_limit = conf . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . file_buffer = FileBuffer ( buffer_limit , logger ) <EOL> self . hidden_ips = [ x . strip ( ) for x in <EOL> conf . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( '<STR_LIT:U+002C>' ) if x . strip ( ) ] <EOL> self . source_account = conf [ '<STR_LIT>' ] <EOL> self . source_container = conf . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def get_logs_to_process ( self , already_processed_files ) : <EOL> lookback_start , lookback_end = self . calculate_lookback ( ) <EOL> logs_to_process = self . get_container_listing ( <EOL> self . source_account , <EOL> self . source_container , <EOL> lookback_start , <EOL> lookback_end , <EOL> already_processed_files ) <EOL> logs_to_process = [ ( self . source_account , self . source_container , x ) <EOL> for x in logs_to_process ] <EOL> self . logger . info ( _ ( '<STR_LIT>' ) % <EOL> len ( logs_to_process ) ) <EOL> return logs_to_process <EOL> def process_one_file ( self , account , container , object_name ) : <EOL> files_to_upload = set ( ) <EOL> try : <EOL> year , month , day , hour , _unused = object_name . split ( '<STR_LIT:/>' , <NUM_LIT:4> ) <EOL> except ValueError : <EOL> self . logger . info ( _ ( '<STR_LIT>' % object_name ) ) <EOL> return <EOL> filename_pattern = '<STR_LIT>' % ( self . working_dir , year , <EOL> month , day , hour ) <EOL> self . logger . debug ( _ ( '<STR_LIT>' % object_name ) ) <EOL> compressed = object_name . endswith ( '<STR_LIT>' ) <EOL> stream = self . get_object_data ( account , container , object_name , <EOL> compressed = compressed ) <EOL> buff = collections . defaultdict ( list ) <EOL> for line in stream : <EOL> clf , account , container = self . convert_log_line ( line ) <EOL> if not clf or not account or not container : <EOL> continue <EOL> if self . get_container_save_log_flag ( account , container ) : <EOL> filename = filename_pattern % ( account , container ) <EOL> self . file_buffer . write ( filename , clf ) <EOL> files_to_upload . add ( filename ) <EOL> self . file_buffer . flush ( ) <EOL> return files_to_upload <EOL> @ memoize <EOL> def get_container_save_log_flag ( self , account , container ) : <EOL> key = '<STR_LIT>' % ( account , container ) <EOL> flag = self . memcache . get ( key ) <EOL> if flag is None : <EOL> metadata = self . internal_proxy . get_container_metadata ( account , <EOL> container ) <EOL> val = metadata . get ( self . metadata_key , '<STR_LIT>' ) <EOL> flag = val . lower ( ) in TRUE_VALUES <EOL> self . memcache . set ( key , flag , timeout = self . frequency ) <EOL> return flag <EOL> def convert_log_line ( self , raw_log ) : <EOL> parts = self . log_line_parser ( raw_log ) <EOL> if parts == { } : <EOL> return None , None , None <EOL> return ( make_clf_from_parts ( parts ) , <EOL> parts . get ( '<STR_LIT>' ) , <EOL> parts . get ( '<STR_LIT>' ) ) <EOL> def log_line_parser ( self , raw_log ) : <EOL> '''<STR_LIT>''' <EOL> d = { } <EOL> try : <EOL> log_arr = raw_log [ <NUM_LIT:16> : ] . split ( '<STR_LIT:U+0020>' ) <EOL> if len ( log_arr ) > <NUM_LIT> : <EOL> log_source = log_arr [ <NUM_LIT> ] <EOL> if log_source != '<STR_LIT:->' : <EOL> return { } <EOL> ( unused , <EOL> server , <EOL> client_ip , <EOL> lb_ip , <EOL> timestamp , <EOL> method , <EOL> request , <EOL> http_version , <EOL> code , <EOL> referrer , <EOL> user_agent , <EOL> auth_token , <EOL> bytes_in , <EOL> bytes_out , <EOL> etag , <EOL> trans_id , <EOL> headers , <EOL> processing_time ) = ( unquote ( x ) for x in log_arr [ : <NUM_LIT> ] ) <EOL> except ValueError : <EOL> self . logger . debug ( _ ( '<STR_LIT>' ) % repr ( raw_log ) ) <EOL> return { } <EOL> if server != self . server_name : <EOL> self . logger . debug ( _ ( '<STR_LIT>' '<STR_LIT>' ) % <EOL> { '<STR_LIT>' : server , '<STR_LIT>' : self . server_name } ) <EOL> return { } <EOL> try : <EOL> ( version , account , container_name , object_name ) = split_path ( request , <NUM_LIT:2> , <NUM_LIT:4> , True ) <EOL> except ValueError , e : <EOL> self . logger . debug ( _ ( '<STR_LIT>' ) % <EOL> { '<STR_LIT:error>' : e , '<STR_LIT>' : repr ( raw_log ) } ) <EOL> return { } <EOL> if container_name is not None : <EOL> container_name = container_name . split ( '<STR_LIT:?>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> if object_name is not None : <EOL> object_name = object_name . split ( '<STR_LIT:?>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> account = account . split ( '<STR_LIT:?>' , <NUM_LIT:1> ) [ <NUM_LIT:0> ] <EOL> if client_ip in self . hidden_ips : <EOL> client_ip = '<STR_LIT>' <EOL> d [ '<STR_LIT>' ] = client_ip <EOL> d [ '<STR_LIT>' ] = lb_ip <EOL> d [ '<STR_LIT>' ] = method <EOL> d [ '<STR_LIT>' ] = request <EOL> d [ '<STR_LIT>' ] = http_version <EOL> d [ '<STR_LIT:code>' ] = code <EOL> d [ '<STR_LIT>' ] = referrer <EOL> d [ '<STR_LIT>' ] = user_agent <EOL> d [ '<STR_LIT>' ] = auth_token <EOL> d [ '<STR_LIT>' ] = bytes_in <EOL> d [ '<STR_LIT>' ] = bytes_out <EOL> d [ '<STR_LIT>' ] = etag <EOL> d [ '<STR_LIT>' ] = trans_id <EOL> d [ '<STR_LIT>' ] = processing_time <EOL> day , month , year , hour , minute , second = timestamp . split ( '<STR_LIT:/>' ) <EOL> d [ '<STR_LIT>' ] = day <EOL> month = ( '<STR_LIT>' % month_map . index ( month ) ) . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:0>' ) <EOL> d [ '<STR_LIT>' ] = month <EOL> d [ '<STR_LIT>' ] = year <EOL> d [ '<STR_LIT>' ] = hour <EOL> d [ '<STR_LIT>' ] = minute <EOL> d [ '<STR_LIT>' ] = second <EOL> d [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> d [ '<STR_LIT>' ] = account <EOL> d [ '<STR_LIT>' ] = container_name <EOL> d [ '<STR_LIT:object_name>' ] = object_name <EOL> d [ '<STR_LIT>' ] = int ( d [ '<STR_LIT>' ] . replace ( '<STR_LIT:->' , '<STR_LIT:0>' ) ) <EOL> d [ '<STR_LIT>' ] = int ( d [ '<STR_LIT>' ] . replace ( '<STR_LIT:->' , '<STR_LIT:0>' ) ) <EOL> d [ '<STR_LIT:code>' ] = int ( d [ '<STR_LIT:code>' ] ) <EOL> return d <EOL> class AccessLogDeliveryDaemon ( Daemon ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , c ) : <EOL> self . conf = c <EOL> super ( AccessLogDeliveryDaemon , self ) . __init__ ( c ) <EOL> self . logger = get_logger ( c , log_route = '<STR_LIT>' ) <EOL> self . log_processor = AccessLogDelivery ( c , self . logger ) <EOL> self . log_delivery_account = c [ '<STR_LIT>' ] <EOL> self . log_delivery_container = c . get ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . target_container = c . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . frequency = int ( c . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> self . worker_count = int ( c . get ( '<STR_LIT>' , '<STR_LIT:1>' ) ) <EOL> self . working_dir = c . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if self . working_dir . endswith ( '<STR_LIT:/>' ) : <EOL> self . working_dir = self . working_dir . rstrip ( '<STR_LIT:/>' ) <EOL> def run_once ( self , * a , ** kw ) : <EOL> self . logger . info ( _ ( "<STR_LIT>" ) ) <EOL> start = time . time ( ) <EOL> already_processed_files = self . log_processor . load_already_processed_files ( ) <EOL> lookback_hours = kw . get ( '<STR_LIT>' ) <EOL> if lookback_hours : <EOL> self . log_processor . lookback_hours = lookback_hours <EOL> lookback_window = kw . get ( '<STR_LIT>' ) <EOL> if lookback_window : <EOL> self . log_processor . lookback_window = lookback_window <EOL> logs_to_process = self . log_processor . get_logs_to_process ( already_processed_files ) <EOL> if not logs_to_process : <EOL> self . logger . info ( _ ( "<STR_LIT>" ) % <EOL> ( ( time . time ( ) - start ) / <NUM_LIT> ) ) <EOL> return <EOL> processor_args = ( self . conf , self . logger ) <EOL> results = multiprocess_collate ( AccessLogDelivery , processor_args , <EOL> '<STR_LIT>' , logs_to_process , <EOL> self . worker_count ) <EOL> processed_files = already_processed_files <EOL> files_to_upload = set ( ) <EOL> for item , data in results : <EOL> a , c , o = item <EOL> processed_files . add ( o ) <EOL> if data : <EOL> files_to_upload . update ( data ) <EOL> len_working_dir = len ( self . working_dir ) + <NUM_LIT:1> <EOL> for filename in files_to_upload : <EOL> target_name = filename [ len_working_dir : ] <EOL> account , target_name = target_name . split ( '<STR_LIT:/>' , <NUM_LIT:1> ) <EOL> some_id = uuid4 ( ) . hex <EOL> target_name = '<STR_LIT>' % ( target_name , some_id ) <EOL> success = self . log_processor . internal_proxy . upload_file ( filename , <EOL> account , <EOL> self . target_container , <EOL> target_name ) <EOL> if success : <EOL> os . unlink ( filename ) <EOL> self . logger . debug ( '<STR_LIT>' % ( filename , <EOL> account ) ) <EOL> else : <EOL> self . logger . error ( '<STR_LIT>' % ( <EOL> filename , account ) ) <EOL> success = self . log_processor . save_processed_files ( processed_files ) <EOL> if not success : <EOL> self . logger . error ( '<STR_LIT>' ) <EOL> self . logger . info ( _ ( "<STR_LIT>" ) % <EOL> ( ( time . time ( ) - start ) / <NUM_LIT> ) ) <EOL> def run_forever ( self , * a , ** kw ) : <EOL> while True : <EOL> start_time = time . time ( ) <EOL> try : <EOL> self . run_once ( ) <EOL> except Exception : <EOL> self . logger . exception ( '<STR_LIT>' ) <EOL> end_time = time . time ( ) <EOL> sleep_time = self . frequency - ( end_time - start_time ) <EOL> time . sleep ( max ( <NUM_LIT:0> , sleep_time ) ) </s>
