<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import datetime <EOL> from django . conf import settings <EOL> from django . core . cache import cache <EOL> import commonware . log <EOL> from celery import Celery <EOL> from celery . signals import task_failure , task_postrun , task_prerun <EOL> from celery . task . sets import TaskSet <EOL> from django_statsd . clients import statsd <EOL> from raven import Client <EOL> from raven . contrib . celery import register_signal , register_logger_signal <EOL> from olympia . amo . utils import chunked , utc_millesecs_from_epoch <EOL> log = commonware . log . getLogger ( '<STR_LIT>' ) <EOL> app = Celery ( '<STR_LIT>' ) <EOL> task = app . task <EOL> app . config_from_object ( '<STR_LIT>' ) <EOL> app . autodiscover_tasks ( settings . INSTALLED_APPS ) <EOL> raven_client = Client ( settings . SENTRY_DSN ) <EOL> register_logger_signal ( raven_client ) <EOL> register_signal ( raven_client ) <EOL> register_logger_signal ( raven_client ) <EOL> @ task_failure . connect <EOL> def process_failure_signal ( exception , traceback , sender , task_id , <EOL> signal , args , kwargs , einfo , ** kw ) : <EOL> """<STR_LIT>""" <EOL> exc_info = ( type ( exception ) , exception , traceback ) <EOL> log . error ( <EOL> u'<STR_LIT>' . format ( * exc_info ) , <EOL> exc_info = exc_info , <EOL> extra = { <EOL> '<STR_LIT:data>' : { <EOL> '<STR_LIT>' : task_id , <EOL> '<STR_LIT>' : sender , <EOL> '<STR_LIT:args>' : args , <EOL> '<STR_LIT>' : kwargs <EOL> } <EOL> } ) <EOL> @ task_prerun . connect <EOL> def start_task_timer ( task_id , task , ** kw ) : <EOL> timer = TaskTimer ( ) <EOL> log . info ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( id = task_id , name = task . name , <EOL> current_dt = timer . current_datetime ) ) <EOL> expiration = <NUM_LIT> * <NUM_LIT> <EOL> cache . set ( timer . cache_key ( task_id ) , timer . current_epoch_ms , expiration ) <EOL> @ task_postrun . connect <EOL> def track_task_run_time ( task_id , task , ** kw ) : <EOL> timer = TaskTimer ( ) <EOL> start_time = cache . get ( timer . cache_key ( task_id ) ) <EOL> if start_time is None : <EOL> log . info ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( id = task_id , name = task . name , <EOL> current_dt = timer . current_datetime ) ) <EOL> else : <EOL> run_time = timer . current_epoch_ms - start_time <EOL> log . info ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> . format ( id = task_id , name = task . name , <EOL> current_dt = timer . current_datetime , <EOL> run_time = run_time ) ) <EOL> statsd . timing ( '<STR_LIT>' . format ( task . name ) , run_time ) <EOL> cache . delete ( timer . cache_key ( task_id ) ) <EOL> class TaskTimer ( object ) : <EOL> def __init__ ( self ) : <EOL> self . current_datetime = datetime . datetime . now ( ) <EOL> self . current_epoch_ms = utc_millesecs_from_epoch ( <EOL> self . current_datetime ) <EOL> def cache_key ( self , task_id ) : <EOL> return '<STR_LIT>' . format ( task_id ) <EOL> def create_subtasks ( task , qs , chunk_size , * args ) : <EOL> """<STR_LIT>""" <EOL> ts = [ task . subtask ( args = ( chunk , ) + args ) <EOL> for chunk in chunked ( qs , chunk_size ) ] <EOL> TaskSet ( ts ) . apply_async ( ) </s>
