<s> """<STR_LIT>""" <EOL> import datetime <EOL> import hashlib <EOL> import json <EOL> import logging <EOL> import os <EOL> import socket <EOL> import sys <EOL> import time <EOL> import taskqueue_server <EOL> import tq_lib <EOL> from tq_config import TaskQueueConfig <EOL> from tq_lib import TASK_STATES <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> import appscale_info <EOL> import constants <EOL> import file_io <EOL> import monit_app_configuration <EOL> import monit_interface <EOL> sys . path . append ( os . path . join ( os . path . dirname ( __file__ ) , "<STR_LIT>" ) ) <EOL> from google . appengine . runtime import apiproxy_errors <EOL> from google . appengine . api import apiproxy_stub_map <EOL> from google . appengine . api import datastore_errors <EOL> from google . appengine . api import datastore_distributed <EOL> from google . appengine . api import datastore <EOL> from google . appengine . ext import db <EOL> from google . appengine . api . taskqueue import taskqueue_service_pb <EOL> sys . path . append ( TaskQueueConfig . CELERY_WORKER_DIR ) <EOL> class TaskName ( db . Model ) : <EOL> """<STR_LIT>""" <EOL> STORED_KIND_NAME = "<STR_LIT>" <EOL> timestamp = db . DateTimeProperty ( auto_now_add = True ) <EOL> queue = db . StringProperty ( required = True ) <EOL> state = db . StringProperty ( required = True ) <EOL> endtime = db . DateTimeProperty ( ) <EOL> app_id = db . StringProperty ( required = True ) <EOL> @ classmethod <EOL> def kind ( cls ) : <EOL> """<STR_LIT>""" <EOL> return cls . STORED_KIND_NAME <EOL> def setup_env ( ) : <EOL> """<STR_LIT>""" <EOL> os . environ [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> os . environ [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> os . environ [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> os . environ [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> class DistributedTaskQueue ( ) : <EOL> """<STR_LIT>""" <EOL> SETUP_WORKERS_TAGS = [ '<STR_LIT>' ] <EOL> STOP_WORKERS_TAGS = [ '<STR_LIT>' ] <EOL> MIN_MAX_CONCURRENCY = "<STR_LIT>" <EOL> LOG_DIR = "<STR_LIT>" <EOL> HARD_TIME_LIMIT = <NUM_LIT> <EOL> TASK_SOFT_TIME_LIMIT = <NUM_LIT> <EOL> PID_FILE_LOC = "<STR_LIT>" <EOL> CELERY_PORT = <NUM_LIT> <EOL> DEFAULT_EXPIRATION = <NUM_LIT:30> <EOL> DEFAULT_MAX_RETRIES = <NUM_LIT:0> <EOL> DEFAULT_MIN_BACKOFF = <NUM_LIT:1> <EOL> DEFAULT_MAX_BACKOFF = <NUM_LIT> <EOL> DEFAULT_MAX_DOUBLINGS = <NUM_LIT:1000> <EOL> TASK_NAME_KIND = "<STR_LIT>" <EOL> CELERY_ENV_VARS = { "<STR_LIT>" : True } <EOL> def __init__ ( self ) : <EOL> """<STR_LIT>""" <EOL> file_io . set_logging_format ( ) <EOL> file_io . mkdir ( self . LOG_DIR ) <EOL> file_io . mkdir ( TaskQueueConfig . CELERY_WORKER_DIR ) <EOL> file_io . mkdir ( TaskQueueConfig . CELERY_CONFIG_DIR ) <EOL> setup_env ( ) <EOL> self . __queue_info_cache = { } <EOL> master_db_ip = appscale_info . get_db_master_ip ( ) <EOL> connection_str = master_db_ip + "<STR_LIT::>" + str ( constants . DB_SERVER_PORT ) <EOL> ds_distrib = datastore_distributed . DatastoreDistributed ( <EOL> constants . DASHBOARD_APP_ID , connection_str , require_indexes = False ) <EOL> apiproxy_stub_map . apiproxy . RegisterStub ( '<STR_LIT>' , ds_distrib ) <EOL> os . environ [ '<STR_LIT>' ] = constants . DASHBOARD_APP_ID <EOL> self . __force_reload = False <EOL> def __parse_json_and_validate_tags ( self , json_request , tags ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> json_response = json . loads ( json_request ) <EOL> except ValueError : <EOL> json_response = { "<STR_LIT:error>" : True , <EOL> "<STR_LIT>" : "<STR_LIT>" } <EOL> return json_response <EOL> for tag in tags : <EOL> if tag not in json_response : <EOL> json_response = { '<STR_LIT:error>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' + tag + '<STR_LIT>' } <EOL> break <EOL> return json_response <EOL> def stop_worker ( self , json_request ) : <EOL> """<STR_LIT>""" <EOL> request = self . __parse_json_and_validate_tags ( json_request , <EOL> self . STOP_WORKERS_TAGS ) <EOL> logging . info ( "<STR_LIT>" . format ( request ) ) <EOL> if '<STR_LIT:error>' in request : <EOL> return json . dumps ( request ) <EOL> app_id = request [ '<STR_LIT>' ] <EOL> watch = "<STR_LIT>" + str ( app_id ) <EOL> try : <EOL> if monit_interface . stop ( watch ) : <EOL> stop_command = self . get_worker_stop_command ( app_id ) <EOL> os . system ( stop_command ) <EOL> TaskQueueConfig . remove_config_files ( app_id ) <EOL> result = { '<STR_LIT:error>' : False } <EOL> else : <EOL> result = { '<STR_LIT:error>' : True , '<STR_LIT>' : "<STR_LIT>" % watch } <EOL> except OSError , os_error : <EOL> result = { '<STR_LIT:error>' : True , '<STR_LIT>' : str ( os_error ) } <EOL> return json . dumps ( result ) <EOL> def get_worker_stop_command ( self , app_id ) : <EOL> """<STR_LIT>""" <EOL> stop_command = "<STR_LIT>" . format ( constants . APPSCALE_HOME , app_id ) <EOL> return stop_command <EOL> def reload_worker ( self , json_request ) : <EOL> """<STR_LIT>""" <EOL> request = self . __parse_json_and_validate_tags ( json_request , <EOL> self . SETUP_WORKERS_TAGS ) <EOL> logging . info ( "<STR_LIT>" . format ( request ) ) <EOL> if '<STR_LIT:error>' in request : <EOL> return json . dumps ( request ) <EOL> app_id = self . __cleanse ( request [ '<STR_LIT>' ] ) <EOL> config = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , app_id ) <EOL> old_queues = self . __queue_info_cache . get ( app_id , { '<STR_LIT>' : [ ] } ) <EOL> old_queue_dict = { } <EOL> for queue in old_queues [ '<STR_LIT>' ] : <EOL> old_queue_dict [ queue [ '<STR_LIT:name>' ] ] = queue <EOL> new_queue_dict = { } <EOL> try : <EOL> new_queues = config . load_queues_from_file ( app_id ) <EOL> for queue in new_queues [ '<STR_LIT>' ] : <EOL> new_queue_dict [ queue [ '<STR_LIT:name>' ] ] = queue <EOL> except ValueError , value_error : <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( value_error ) } ) <EOL> except NameError , name_error : <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( name_error ) } ) <EOL> except Exception , exception : <EOL> logging . error ( "<STR_LIT>" ) <EOL> logging . exception ( exception ) <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( exception ) } ) <EOL> reload_queues = False <EOL> for queue_name in old_queue_dict . keys ( ) : <EOL> if queue_name not in new_queue_dict : <EOL> logging . info ( "<STR_LIT>" . format ( app_id , queue_name ) ) <EOL> reload_queues = True <EOL> for queue_name in new_queue_dict . keys ( ) : <EOL> if queue_name not in old_queue_dict . keys ( ) : <EOL> logging . info ( "<STR_LIT>" . format ( app_id , queue_name ) ) <EOL> reload_queues = True <EOL> if reload_queues : <EOL> logging . info ( "<STR_LIT>" . format ( app_id , old_queue_dict ) ) <EOL> logging . info ( "<STR_LIT>" . format ( app_id , new_queue_dict ) ) <EOL> self . stop_worker ( json_request ) <EOL> self . start_worker ( json_request ) <EOL> self . __force_reload = True <EOL> else : <EOL> logging . info ( "<STR_LIT>" ) <EOL> self . __queue_info_cache [ app_id ] = new_queues <EOL> json_response = { '<STR_LIT:error>' : False } <EOL> return json . dumps ( json_response ) <EOL> def start_worker ( self , json_request ) : <EOL> """<STR_LIT>""" <EOL> request = self . __parse_json_and_validate_tags ( json_request , <EOL> self . SETUP_WORKERS_TAGS ) <EOL> logging . info ( "<STR_LIT>" . format ( request ) ) <EOL> if '<STR_LIT:error>' in request : <EOL> return json . dumps ( request ) <EOL> app_id = self . __cleanse ( request [ '<STR_LIT>' ] ) <EOL> config = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , app_id ) <EOL> try : <EOL> self . __queue_info_cache [ app_id ] = config . load_queues_from_file ( app_id ) <EOL> config . create_celery_file ( TaskQueueConfig . QUEUE_INFO_FILE ) <EOL> config . create_celery_worker_scripts ( TaskQueueConfig . QUEUE_INFO_FILE ) <EOL> except ValueError , value_error : <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( value_error ) } ) <EOL> except NameError , name_error : <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( name_error ) } ) <EOL> except Exception , exception : <EOL> logging . error ( "<STR_LIT>" ) <EOL> logging . exception ( exception ) <EOL> return json . dumps ( { "<STR_LIT:error>" : True , "<STR_LIT>" : str ( exception ) } ) <EOL> log_file = self . LOG_DIR + app_id + "<STR_LIT>" <EOL> command = [ "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" + TaskQueueConfig . get_celery_worker_module_name ( app_id ) , <EOL> "<STR_LIT>" + app_id , <EOL> "<STR_LIT>" + TaskQueueConfig . CELERY_WORKER_DIR , <EOL> "<STR_LIT>" + log_file , <EOL> "<STR_LIT>" + str ( self . HARD_TIME_LIMIT ) , <EOL> "<STR_LIT>" + str ( TaskQueueConfig . CELERY_CONCURRENCY ) , <EOL> "<STR_LIT>" + str ( self . TASK_SOFT_TIME_LIMIT ) , <EOL> "<STR_LIT>" + self . PID_FILE_LOC + '<STR_LIT>' + app_id + "<STR_LIT>" , <EOL> "<STR_LIT>" + TaskQueueConfig . CELERY_STATE_DIR + '<STR_LIT>' + app_id + "<STR_LIT>" , <EOL> "<STR_LIT>" ] <EOL> start_command = str ( '<STR_LIT:U+0020>' . join ( command ) ) <EOL> stop_command = self . get_worker_stop_command ( app_id ) <EOL> watch = "<STR_LIT>" + str ( app_id ) <EOL> monit_app_configuration . create_config_file ( watch , <EOL> start_command , <EOL> stop_command , <EOL> [ self . CELERY_PORT ] , <EOL> env_vars = self . CELERY_ENV_VARS ) <EOL> if monit_interface . start ( watch ) : <EOL> json_response = { '<STR_LIT:error>' : False } <EOL> else : <EOL> json_response = { '<STR_LIT:error>' : True , <EOL> '<STR_LIT>' : "<STR_LIT>" % watch } <EOL> return json . dumps ( json_response ) <EOL> def fetch_queue_stats ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueFetchQueueStatsRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueFetchQueueStatsResponse ( ) <EOL> for queue in request . queue_name_list ( ) : <EOL> stats_response = response . add_queuestats ( ) <EOL> count = TaskName . all ( ) . filter ( "<STR_LIT>" , TASK_STATES . QUEUED ) . filter ( "<STR_LIT>" , queue ) . filter ( "<STR_LIT>" , app_id ) . count ( ) <EOL> stats_response . set_num_tasks ( count ) <EOL> stats_response . set_oldest_eta_usec ( - <NUM_LIT:1> ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def purge_queue ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueuePurgeQueueRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueuePurgeQueueResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def delete ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueDeleteRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueDeleteResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def query_and_own_tasks ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueQueryAndOwnTasksRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueQueryAndOwnTasksResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def add ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueAddRequest ( http_data ) <EOL> request . set_app_id ( app_id ) <EOL> response = taskqueue_service_pb . TaskQueueAddResponse ( ) <EOL> bulk_request = taskqueue_service_pb . TaskQueueBulkAddRequest ( ) <EOL> bulk_response = taskqueue_service_pb . TaskQueueBulkAddResponse ( ) <EOL> bulk_request . add_add_request ( ) . CopyFrom ( request ) <EOL> self . __bulk_add ( bulk_request , bulk_response ) <EOL> if bulk_response . taskresult_size ( ) == <NUM_LIT:1> : <EOL> result = bulk_response . taskresult ( <NUM_LIT:0> ) . result ( ) <EOL> else : <EOL> err_code = taskqueue_service_pb . TaskQueueServiceError . INTERNAL_ERROR <EOL> return ( response . Encode ( ) , err_code , <EOL> "<STR_LIT>" ) <EOL> if result != taskqueue_service_pb . TaskQueueServiceError . OK : <EOL> return ( response . Encode ( ) , result , "<STR_LIT>" ) <EOL> elif bulk_response . taskresult ( <NUM_LIT:0> ) . has_chosen_task_name ( ) : <EOL> response . set_chosen_task_name ( <EOL> bulk_response . taskresult ( <NUM_LIT:0> ) . chosen_task_name ( ) ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def bulk_add ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueBulkAddRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueBulkAddResponse ( ) <EOL> self . __bulk_add ( request , response ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def __bulk_add ( self , request , response ) : <EOL> """<STR_LIT>""" <EOL> if request . add_request_size ( ) == <NUM_LIT:0> : <EOL> return <EOL> now = datetime . datetime . utcfromtimestamp ( time . time ( ) ) <EOL> error_found = False <EOL> for add_request in request . add_request_list ( ) : <EOL> task_result = response . add_taskresult ( ) <EOL> result = tq_lib . verify_task_queue_add_request ( add_request . app_id ( ) , <EOL> add_request , now ) <EOL> if result == taskqueue_service_pb . TaskQueueServiceError . SKIPPED : <EOL> task_name = None <EOL> if add_request . has_task_name ( ) : <EOL> task_name = add_request . task_name ( ) <EOL> namespaced_name = tq_lib . choose_task_name ( add_request . app_id ( ) , <EOL> add_request . queue_name ( ) , <EOL> user_chosen = task_name ) <EOL> add_request . set_task_name ( namespaced_name ) <EOL> task_result . set_chosen_task_name ( namespaced_name ) <EOL> else : <EOL> error_found = True <EOL> task_result . set_result ( result ) <EOL> if error_found : <EOL> return <EOL> for add_request , task_result in zip ( request . add_request_list ( ) , <EOL> response . taskresult_list ( ) ) : <EOL> try : <EOL> self . __enqueue_push_task ( add_request ) <EOL> except apiproxy_errors . ApplicationError , e : <EOL> task_result . set_result ( e . application_error ) <EOL> else : <EOL> task_result . set_result ( taskqueue_service_pb . TaskQueueServiceError . OK ) <EOL> def __method_mapping ( self , method ) : <EOL> """<STR_LIT>""" <EOL> if method == taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . GET : <EOL> return '<STR_LIT:GET>' <EOL> elif method == taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . POST : <EOL> return '<STR_LIT:POST>' <EOL> elif method == taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . HEAD : <EOL> return '<STR_LIT>' <EOL> elif method == taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . PUT : <EOL> return '<STR_LIT>' <EOL> elif method == taskqueue_service_pb . TaskQueueQueryTasksResponse_Task . DELETE : <EOL> return '<STR_LIT>' <EOL> def __check_and_store_task_names ( self , request ) : <EOL> """<STR_LIT>""" <EOL> task_name = request . task_name ( ) <EOL> item = TaskName . get_by_key_name ( task_name ) <EOL> logging . debug ( "<STR_LIT>" . format ( task_name ) ) <EOL> if item : <EOL> logging . warning ( "<STR_LIT>" ) <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . TASK_ALREADY_EXISTS ) <EOL> else : <EOL> new_name = TaskName ( key_name = task_name , state = tq_lib . TASK_STATES . QUEUED , <EOL> queue = request . queue_name ( ) , app_id = request . app_id ( ) ) <EOL> logging . debug ( "<STR_LIT>" . format ( str ( new_name ) ) ) <EOL> try : <EOL> db . put ( new_name ) <EOL> except datastore_errors . InternalError , internal_error : <EOL> logging . error ( str ( internal_error ) ) <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . DATASTORE_ERROR ) <EOL> def __enqueue_push_task ( self , request ) : <EOL> """<STR_LIT>""" <EOL> self . __validate_push_task ( request ) <EOL> self . __check_and_store_task_names ( request ) <EOL> args = self . get_task_args ( request ) <EOL> headers = self . get_task_headers ( request ) <EOL> countdown = int ( headers [ '<STR_LIT>' ] ) - int ( datetime . datetime . now ( ) . strftime ( "<STR_LIT:%s>" ) ) <EOL> task_func = self . __get_task_function ( request ) <EOL> result = task_func . apply_async ( kwargs = { '<STR_LIT>' : headers , <EOL> '<STR_LIT:args>' : args } , <EOL> expires = args [ '<STR_LIT>' ] , <EOL> acks_late = True , <EOL> countdown = countdown , <EOL> queue = TaskQueueConfig . get_celery_queue_name ( <EOL> request . app_id ( ) , request . queue_name ( ) ) , <EOL> routing_key = TaskQueueConfig . get_celery_queue_name ( <EOL> request . app_id ( ) , request . queue_name ( ) ) ) <EOL> def __get_task_function ( self , request ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> task_module = __import__ ( TaskQueueConfig . get_celery_worker_module_name ( request . app_id ( ) ) ) <EOL> if self . __force_reload : <EOL> start = time . time ( ) <EOL> reload ( task_module ) <EOL> time_taken = time . time ( ) - start <EOL> self . __force_reload = False <EOL> logging . info ( "<STR_LIT>" . format ( request . app_id ( ) , time_taken ) ) <EOL> task_func = getattr ( task_module , <EOL> TaskQueueConfig . get_queue_function_name ( request . queue_name ( ) ) ) <EOL> return task_func <EOL> except AttributeError , attribute_error : <EOL> logging . exception ( attribute_error ) <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> except ImportError , import_error : <EOL> logging . exception ( import_error ) <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> def get_task_args ( self , request ) : <EOL> """<STR_LIT>""" <EOL> args = { } <EOL> args [ '<STR_LIT>' ] = request . task_name ( ) <EOL> args [ '<STR_LIT:url>' ] = request . url ( ) <EOL> args [ '<STR_LIT>' ] = request . app_id ( ) <EOL> args [ '<STR_LIT>' ] = request . queue_name ( ) <EOL> args [ '<STR_LIT>' ] = self . __method_mapping ( request . method ( ) ) <EOL> args [ '<STR_LIT:body>' ] = request . body ( ) <EOL> args [ '<STR_LIT>' ] = request . payload ( ) <EOL> args [ '<STR_LIT:description>' ] = request . description ( ) <EOL> args [ '<STR_LIT>' ] = self . DEFAULT_MAX_RETRIES <EOL> args [ '<STR_LIT>' ] = self . __when_to_expire ( request ) <EOL> args [ '<STR_LIT>' ] = self . DEFAULT_MAX_RETRIES <EOL> args [ '<STR_LIT>' ] = self . DEFAULT_MAX_BACKOFF <EOL> args [ '<STR_LIT>' ] = self . DEFAULT_MIN_BACKOFF <EOL> args [ '<STR_LIT>' ] = self . DEFAULT_MAX_DOUBLINGS <EOL> if request . app_id ( ) not in self . __queue_info_cache : <EOL> try : <EOL> config = TaskQueueConfig ( TaskQueueConfig . RABBITMQ , request . app_id ( ) ) <EOL> self . __queue_info_cache [ request . app_id ( ) ] = config . load_queues_from_file ( request . app_id ( ) ) <EOL> except ValueError , value_error : <EOL> logging . error ( "<STR_LIT>" . format ( request . app_id ( ) ) ) <EOL> except NameError , name_error : <EOL> logging . error ( "<STR_LIT>" . format ( request . app_id ( ) ) ) <EOL> except Exception , exception : <EOL> logging . error ( "<STR_LIT>" ) <EOL> logging . exception ( exception ) <EOL> if request . app_id ( ) in self . __queue_info_cache : <EOL> queue_list = self . __queue_info_cache [ request . app_id ( ) ] [ '<STR_LIT>' ] <EOL> for queue in queue_list : <EOL> if queue . get ( '<STR_LIT:name>' ) == request . queue_name ( ) : <EOL> if '<STR_LIT>' in queue : <EOL> retry_params = queue [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in retry_params : <EOL> args [ '<STR_LIT>' ] = retry_params [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in retry_params : <EOL> args [ '<STR_LIT>' ] = retry_params [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in retry_params : <EOL> args [ '<STR_LIT>' ] = retry_params [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in retry_params : <EOL> args [ '<STR_LIT>' ] = retry_params [ '<STR_LIT>' ] <EOL> break <EOL> if request . has_retry_parameters ( ) : <EOL> retry_params = request . retry_parameters ( ) <EOL> if retry_params . has_retry_limit ( ) : <EOL> args [ '<STR_LIT>' ] = retry_params . retry_limit ( ) <EOL> if retry_params . has_min_backoff_sec ( ) : <EOL> args [ '<STR_LIT>' ] = request . retry_parameters ( ) . min_backoff_sec ( ) <EOL> if retry_params . has_max_backoff_sec ( ) : <EOL> args [ '<STR_LIT>' ] = request . retry_parameters ( ) . max_backoff_sec ( ) <EOL> if retry_params . has_max_doublings ( ) : <EOL> args [ '<STR_LIT>' ] = request . retry_parameters ( ) . max_doublings ( ) <EOL> return args <EOL> def get_task_headers ( self , request ) : <EOL> """<STR_LIT>""" <EOL> headers = { } <EOL> for header in request . header_list ( ) : <EOL> headers [ header . key ( ) ] = header . value ( ) <EOL> eta = self . __when_to_run ( request ) <EOL> secret = appscale_info . get_secret ( ) <EOL> secret_hash = hashlib . sha1 ( request . app_id ( ) + '<STR_LIT:/>' + secret ) . hexdigest ( ) <EOL> headers [ '<STR_LIT>' ] = secret_hash <EOL> headers [ '<STR_LIT>' ] = request . queue_name ( ) <EOL> headers [ '<STR_LIT>' ] = request . task_name ( ) <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> headers [ '<STR_LIT>' ] = str ( int ( eta . strftime ( "<STR_LIT:%s>" ) ) ) <EOL> return headers <EOL> def __when_to_run ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if request . has_eta_usec ( ) : <EOL> eta = request . eta_usec ( ) <EOL> return datetime . datetime . fromtimestamp ( eta / <NUM_LIT> ) <EOL> else : <EOL> return datetime . datetime . now ( ) <EOL> def __when_to_expire ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if request . has_retry_parameters ( ) and request . retry_parameters ( ) . has_age_limit_sec ( ) : <EOL> limit = request . retry_parameters ( ) . age_limit_sec ( ) <EOL> return datetime . datetime . now ( ) + datetime . timedelta ( seconds = limit ) <EOL> else : <EOL> return datetime . datetime . now ( ) + datetime . timedelta ( days = self . DEFAULT_EXPIRATION ) <EOL> def __validate_push_task ( self , request ) : <EOL> """<STR_LIT>""" <EOL> if not request . has_queue_name ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_QUEUE_NAME ) <EOL> if not request . has_task_name ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_TASK_NAME ) <EOL> if not request . has_app_id ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . UNKNOWN_QUEUE ) <EOL> if not request . has_url ( ) : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_URL ) <EOL> if request . has_mode ( ) and request . mode ( ) == taskqueue_service_pb . TaskQueueMode . PULL : <EOL> raise apiproxy_errors . ApplicationError ( <EOL> taskqueue_service_pb . TaskQueueServiceError . INVALID_QUEUE_MODE ) <EOL> def modify_task_lease ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueModifyTaskLeaseRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueModifyTaskLeaseResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def fetch_queue ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueFetchQueuesRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueFetchQueuesResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def query_tasks ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueQueryTasksRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueQueryTasksResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def fetch_task ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueFetchTaskRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueFetchTaskResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def force_run ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueForceRunRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueForceRunResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def pause_queue ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueuePauseQueueRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueuePauseQueueResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def delete_group ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueDeleteGroupRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueDeleteGroupResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def update_storage_limit ( self , app_id , http_data ) : <EOL> """<STR_LIT>""" <EOL> request = taskqueue_service_pb . TaskQueueUpdateStorageLimitRequest ( http_data ) <EOL> response = taskqueue_service_pb . TaskQueueUpdateStorageLimitResponse ( ) <EOL> return ( response . Encode ( ) , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def __cleanse ( self , str_input ) : <EOL> """<STR_LIT>""" <EOL> for char in "<STR_LIT>" : <EOL> str_input = str_input . replace ( char , "<STR_LIT:_>" ) <EOL> return str_input <EOL> def __is_localhost ( self , hostname ) : <EOL> """<STR_LIT>""" <EOL> if socket . gethostname ( ) == hostname : <EOL> return True <EOL> elif socket . gethostbyname ( socket . gethostname ( ) ) == hostname : <EOL> return True <EOL> else : <EOL> return False </s>
