<s> from __future__ import absolute_import , division , print_function , unicode_literals <EOL> from operator import itemgetter <EOL> import os , sys , json , logging , time , threading , warnings , collections <EOL> try : <EOL> import Queue <EOL> except ImportError : <EOL> import queue as Queue <EOL> import boto3 <EOL> from botocore . exceptions import ClientError <EOL> handler_base_class = logging . Handler <EOL> def _idempotent_create ( _callable , * args , ** kwargs ) : <EOL> try : <EOL> _callable ( * args , ** kwargs ) <EOL> except ClientError as e : <EOL> if e . response . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT>" ) != "<STR_LIT>" : <EOL> raise <EOL> class PyCWLWarning ( UserWarning ) : <EOL> pass <EOL> class CloudWatchLogHandler ( handler_base_class ) : <EOL> """<STR_LIT>""" <EOL> END = <NUM_LIT:1> <EOL> def __init__ ( self , log_group = __name__ , stream_name = None , use_queues = True , send_interval = <NUM_LIT> , <EOL> max_batch_size = <NUM_LIT> * <NUM_LIT> , max_batch_count = <NUM_LIT> , boto3_session = None , * args , ** kwargs ) : <EOL> handler_base_class . __init__ ( self , * args , ** kwargs ) <EOL> self . log_group = log_group <EOL> self . stream_name = stream_name <EOL> self . use_queues = use_queues <EOL> self . send_interval = send_interval <EOL> self . max_batch_size = max_batch_size <EOL> self . max_batch_count = max_batch_count <EOL> self . queues , self . sequence_tokens = { } , { } <EOL> self . threads = [ ] <EOL> self . shutting_down = False <EOL> self . cwl_client = ( boto3_session or boto3 ) . client ( "<STR_LIT>" ) <EOL> _idempotent_create ( self . cwl_client . create_log_group , logGroupName = self . log_group ) <EOL> def _submit_batch ( self , batch , stream_name ) : <EOL> if len ( batch ) < <NUM_LIT:1> : <EOL> return <EOL> sorted_batch = sorted ( batch , key = itemgetter ( '<STR_LIT>' ) , reverse = False ) <EOL> kwargs = dict ( logGroupName = self . log_group , logStreamName = stream_name , <EOL> logEvents = sorted_batch ) <EOL> if self . sequence_tokens [ stream_name ] is not None : <EOL> kwargs [ "<STR_LIT>" ] = self . sequence_tokens [ stream_name ] <EOL> try : <EOL> response = self . cwl_client . put_log_events ( ** kwargs ) <EOL> except ClientError as e : <EOL> if e . response . get ( "<STR_LIT>" , { } ) . get ( "<STR_LIT>" ) in ( "<STR_LIT>" , <EOL> "<STR_LIT>" ) : <EOL> kwargs [ "<STR_LIT>" ] = e . response [ "<STR_LIT>" ] [ "<STR_LIT>" ] . rsplit ( "<STR_LIT:U+0020>" , <NUM_LIT:1> ) [ - <NUM_LIT:1> ] <EOL> response = self . cwl_client . put_log_events ( ** kwargs ) <EOL> else : <EOL> raise <EOL> if "<STR_LIT>" in response : <EOL> raise Exception ( "<STR_LIT>" . format ( response ) ) <EOL> def emit ( self , message ) : <EOL> stream_name = self . stream_name <EOL> if stream_name is None : <EOL> stream_name = message . name <EOL> if stream_name not in self . sequence_tokens : <EOL> _idempotent_create ( self . cwl_client . create_log_stream , <EOL> logGroupName = self . log_group , logStreamName = stream_name ) <EOL> self . sequence_tokens [ stream_name ] = None <EOL> msg = dict ( timestamp = int ( message . created * <NUM_LIT:1000> ) , message = self . format ( message ) ) <EOL> if isinstance ( msg [ "<STR_LIT:message>" ] , collections . Mapping ) : <EOL> msg [ "<STR_LIT:message>" ] = json . dumps ( msg [ "<STR_LIT:message>" ] ) <EOL> if self . use_queues : <EOL> if stream_name not in self . queues : <EOL> self . queues [ stream_name ] = Queue . Queue ( ) <EOL> thread = threading . Thread ( target = self . batch_sender , <EOL> args = ( self . queues [ stream_name ] , stream_name , self . send_interval , <EOL> self . max_batch_size , self . max_batch_count ) ) <EOL> self . threads . append ( thread ) <EOL> thread . daemon = True <EOL> thread . start ( ) <EOL> if self . shutting_down : <EOL> warnings . warn ( "<STR_LIT>" , PyCWLWarning ) <EOL> else : <EOL> self . queues [ stream_name ] . put ( msg ) <EOL> else : <EOL> self . _submit_batch ( [ msg ] , stream_name ) <EOL> def batch_sender ( self , my_queue , stream_name , send_interval , max_batch_size , max_batch_count ) : <EOL> msg = None <EOL> def size ( msg ) : <EOL> return len ( msg [ "<STR_LIT:message>" ] ) + <NUM_LIT> <EOL> while msg != self . END : <EOL> cur_batch = [ ] if msg is None else [ msg ] <EOL> cur_batch_size = sum ( size ( msg ) for msg in cur_batch ) <EOL> cur_batch_msg_count = len ( cur_batch ) <EOL> cur_batch_deadline = time . time ( ) + send_interval <EOL> while True : <EOL> try : <EOL> msg = my_queue . get ( block = True , timeout = max ( <NUM_LIT:0> , cur_batch_deadline - time . time ( ) ) ) <EOL> except Queue . Empty : <EOL> msg = None <EOL> if msg is None or msg == self . END or cur_batch_size + size ( msg ) > max_batch_size or cur_batch_msg_count >= max_batch_count or time . time ( ) >= cur_batch_deadline : <EOL> self . _submit_batch ( cur_batch , stream_name ) <EOL> if msg is not None : <EOL> my_queue . task_done ( ) <EOL> break <EOL> elif msg : <EOL> cur_batch_size += size ( msg ) <EOL> cur_batch_msg_count += <NUM_LIT:1> <EOL> cur_batch . append ( msg ) <EOL> my_queue . task_done ( ) <EOL> def flush ( self ) : <EOL> self . shutting_down = True <EOL> for q in self . queues . values ( ) : <EOL> q . put ( self . END ) <EOL> for q in self . queues . values ( ) : <EOL> q . join ( ) </s>
