U .eM @sdZddlZddlZddlmZmZmZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZdd lmZmZdd lmZd d lmZd dlmZdZdZee Z!edddddddddgZ"ddZ#ddZ$GdddeZ%d d!Z&d"d#Z'e'd$d%Z(e'd&d'd(efgd)dd+d,Z)d-d.Z*e'd/d0d1d2d3Z+ej,j-fd4d5Z.ej/j0ej1j0fd6d7Z2e&d8d0d1dd9d:Z3e&d;dZ4dd?d@Z5e&d8dAe6fgdBdCdDdEZ7e&dFe6fdGe6fgdHdIdJdGZ8e&dFe6fdKe9fdLe9fgdMdIddNdOZ:e'dPdQZ;e&ddRdSZe&dXdYZ?e'd*dZdd[d\Z@e'd]d^d_d`ZAe'dadbZBe'dcdddedfZCdgdhZDe'didddjdkZEe'dlddddmdnZFe'dodddpdqZGe'drdsdtduddvdwZHe'dxde6fdyeIfdzeIfgd{d|dddZJe'ddZKe'deIfgddIdddZLe&deIfgddIdddZMe&deIfgddIdddZNe&dddZOe&deIfdeIfgddIdddZPe&dddZQe&de6fde6fde6fde6fgddIdddZRe&de6fgddIddZSe'ddZTdS)z.Worker remote control command implementations.N)UserDict defaultdict namedtuple) TERM_SIGNAME) safe_repr)WorkerShutdown)signals) maybe_list) get_logger)jsonify strtobool)ratestate)Request)Panel)exchange routing_key rate_limitcontroller_info_taliastypevisibledefault_timeouthelp signatureargsvariadiccCsd|iS)Nokvaluer r 9/tmp/pip-unpacked-wheel-f4liivr4/celery/worker/control.pyrsrcCsd|iS)Nerrorr r!r r r#nok"sr%c @s2eZdZdZiZiZeddZed dd ZdS) rz+Global registry of remote control commands.cOs |r|jf||S|jf|SN) _register)clsrkwargsr r r#register,szPanel.registerNcontrolT?c s" f dd} | S)Nc s^p|j}p$|jpddd}|j|<t |j|<rZ|j<|S)N r)__name____doc__stripsplitdatarmeta)ZfunZ control_nameZ_help rrr(rrnamerrrrr r#_inner7s     zPanel._register.._innerr ) r(r6rrrrrrrrr7r r5r#r'2s zPanel._register) NNr+Tr,NNNN) r/ __module__ __qualname__r0r3r4 classmethodr*r'r r r r#r&s rcKstjfddi|S)Nrr+rr*r)r r r#control_commandDsr=cKstjfddi|S)Nrinspectr;r<r r r#inspect_commandHsr?cCst|jS)z6Information about Celery installation for bug reports.)rappZ bugreportrr r r#reportNsrAZ dump_confz[include_defaults=False] with_defaults)rrrFcKst|jjj|dttdS)zList configuration.)rB)Z keyfilterZunknown_type_filter)r r@conftable_wanted_config_keyr)rrBr)r r r#rCTsrCcCst|to|d S)N__) isinstancestr startswith)keyr r r#rE`srEidsz[id1 [id2 [... [idN]]]])rrcKsddtt|DS)z!Query for task information by id.cSs i|]}|jt||fqSr )id_state_of_taskinfo).0reqr r r# lszquery_task..)_find_requests_by_idr )rrKr)r r r# query_taskfs rSc cs2|D](}z||VWqtk r*YqXqdSr&)KeyError)rK get_requesttask_idr r r#rRrs rRcCs||r dS||rdSdS)Nactivereservedreadyr )requestZ is_active is_reservedr r r#rM{s rMrVcKsNtt|p gd}}t||||f|}t|tr>d|kr>|Std|dS)zRevoke task by task id (or list of ids). Keyword Arguments: terminate (bool): Also terminate the process if the task is active. signal (str): Name of signal to use for terminate (e.g., ``KILL``). Nrztasks z flagged as revoked)setr _revokerGdictr)rrV terminatesignalr)task_idsr r r#revokes rbheadersz/[key1=value1 [key2=value2 [... [keyN=valueN]]]]cKs.t|p t}t|tr&dd|D}|D]2\}}ttj |pFgtt|}|tj|<q.|svt d|dSttj } t t } | D]z} t| dr| jr|D]\\} } | | jkrt| } t| j| }t |t | @}|r| | || j|jj|dqq| st d|dSt d| dS) aRevoke task by header (or list of headers). Keyword Arguments: headers(dictionary): Dictionary that contains stamping scheme name as keys and stamps as values. If headers is a list, it will be converted to a dictionary. terminate (bool): Also terminate the process if the task is active. signal (str): Name of signal to use for terminate (e.g., ``KILL``). Sample headers input: {'mtask_id': [id1, id2, id3]} cSs&i|]}|dd|ddqS)=rr)r2)rOhr r r#rQs z-revoke_by_stamped_headers..zheaders z' flagged as revoked, but not terminatedstampsr`z were not terminatedz revoked)_signalssignumrrGlistitemsr worker_stateZrevoked_stampsgetractive_requestsrr\hasattrrfupdater_consumerpool)rrcr_r`r)riheaderrfZupdated_stampsrnZ#terminated_scheme_to_stamps_mappingrPZexpected_header_keyZexpected_header_valueZ actual_headerZmatching_stamps_for_requestr r r#revoke_by_stamped_headerss.     rtc Kst|}t}tj||rt|p(t}t|D]L}|j |kr4| |j t d|j ||j |jj|dt||kr4qq4|stdStdd|Sd|} t d| |S)NzTerminating %s (%s)rgzterminate: tasks unknownz terminate: {}z, zTasks flagged as revoked: %s)lenr\rlrevokedrprhrirrRrLaddloggerrNr_rqrrrformatjoin) rrar_r`r)sizeZ terminatedrirZZidstrr r r#r]s$       r]r`z [id1 [id2 [... [idN]]]])rrrcKst||d|dS)z+Terminate task by task id (or list of ids).T)r_r`)rb)rr`rVr)r r r#r_sr_ task_namerz0 )rrc Ksz t|Wn4tk r@}ztd|WYSd}~XYnXz||jj|_Wn,tk rtjd|ddtdYSX|j |st d|t dSt d ||t d S) zTell worker(s) to modify the rate limit for a task by type. See Also: :attr:`celery.app.task.Task.rate_limit`. Arguments: task_name (str): Type of task to set rate limit for. rate_limit (int, str): New rate limit. zInvalid rate limit string: Nz&Rate limit attempt for unknown task %sTexc_info unknown taskz)Rate limits disabled for tasks of type %sz rate limit disabled successfullyz(New rate limit for tasks of type %s: %s.znew rate limit set successfully) r ValueErrorr%r@tasksrrTrxr$rqZreset_rate_limitsrNr)rr|rr)excr r r#rs* $  softhardz# [hard_secs]cKsbz|jj|}Wn,tk r<tjd|ddtdYSX||_||_td|||t dS)zTell worker(s) to modify the time limit for task by type. Arguments: task_name (str): Name of task to change. hard (float): Hard time limit. soft (float): Soft time limit. z-Change time limit attempt for unknown task %sTr}rz5New time limits for tasks of type %s: soft=%s hard=%sztime limits set successfully) r@rrTrxr$r%Zsoft_time_limit time_limitrNr)rr|rrr)taskr r r#rs rcKsd|jjjiS)z Get current logical clock value.clock)r@rr"rr)r r r#r=srcKs|jjr|jj|||dS)zHold election. Arguments: id (str): Unique election id. topic (str): Election topic. action (str): Action to take for elected actor. N)rqZgossipelection)rrLZtopicactionr)r r r#rCs rcCs>|jj}|jr6d|jkr6|jdtdtdStdS)z+Tell worker(s) to send task-related events.rz)Events of group {task} enabled by remote.ztask events enabledztask events already enabled)rqevent_dispatchergroupsrwrxrNrr dispatcherr r r# enable_eventsPs   rcCs8|jj}d|jkr0|jdtdtdStdS)z3Tell worker(s) to stop sending task-related events.rz*Events of group {task} disabled by remote.ztask events disabledztask events already disabled)rqrrdiscardrxrNrrr r r#disable_events[s    rcCs,td|jj}|jddditjdS)z3Tell worker(s) to send event heartbeat immediately.zHeartbeat requested by remote.worker-heartbeatfreqN)r)rxdebugrqrsendrlZ SOFTWARE_INFOrr r r# heartbeatfs r)rcKsJ||jkrFtd||r&tj|tjtjj|jj dSdS)zRequest mingle sync-data.z sync with %s)rvrN) hostnamerxrNrlrvrppurge_datar@rZforward)rZ from_nodervr)r r r#hellops     rg?)rcKstdS)zPing worker(s).Zpong)rrr r r#pingsrcKs |jjS)z&Request worker statistics/information.)rq controllerstatsrr r r#rsrZ dump_schedule)rcKstt|jjS)z0List of currently scheduled ETA/countdown tasks.)rj_iter_schedule_requestsrqtimerrr r r# scheduledsrc csn|jjD]`}z|jjd}Wnttfk r8YqYqXt|tr|jrT|j nd|j | dVqdS)Nr)etapriorityrZ) schedulequeueentryr IndexError TypeErrorrGrr isoformatrrN)rZwaitingZarg0r r r#rs   rZ dump_reservedcKs.|tj|tj}|s gSdd|DS)zAList of currently reserved tasks, not including scheduled/active.cSsg|] }|qSr rNrOrZr r r# szreserved..)tsetrlreserved_requestsrn)rr)Zreserved_tasksr r r#rXs  rXZ dump_activec sfdd|tjDS)z'List of tasks currently being executed.csg|]}|jdqS)saferrrr r#rszactive..)rrlrn)rrr)r rr#rWs  rWZ dump_revokedcKs ttjS)zList of revoked task-ids.)rjrlrvrr r r#rvsrvZ dump_tasks taskinfoitemsz[attr1 [attr2 [... [attrN]]]])rrrc sJ|jjpt|rn ddD}fddfddt|DS)zList of registered tasks. Arguments: taskinfoitems (Sequence[str]): List of task attributes to include. Defaults to ``exchange,routing_key,rate_limit``. builtins (bool): Also include built-in tasks. css|]}|ds|VqdS)zcelery.N)rIrOrr r r# s zregistered..csBfddD}|r._extract_info..cSsg|]}d|qS)rd)rz)rOfr r r#rsz5registered.._extract_info..z{} [{}] )rkryr6rz)rfieldsrN)rrr# _extract_infos z!registered.._extract_infocsg|]}|qSr r r)rregr r#rszregistered..)r@rDEFAULT_TASK_INFO_ITEMSsorted)rrbuiltinsr)rr )rrrr# registereds  rgN@num max_depthz.[object_type=Request] [num=200 [max_depth=10]])rrr rc sz ddl}Wntk r(tdYnXtd|tjddddF}||d||j|fd d |jd d |jiW5QRSQRXdS) aCreate graph of uncollected objects (memory-leak debugging). Arguments: num (int): Max number of objects to graph. max_depth (int): Traverse at most n levels deep. type (str): Name of object to graph. Default is ``"Request"``. rNzRequires the objgraph libraryzDumping graph for type %rZcobjgz.pngF)prefixsuffixdeletecs|kSr&r )vZobjectsr r#zobjgraph..)rZ highlightfilenamer) objgraph ImportErrorrxrNtempfileNamedTemporaryFileZby_typeZ show_backrefsr6)rrrrZ _objgraphfhr rr#rs$    rcKsddlm}|S)z Sample current RSS memory usage.r) sample_mem)Zcelery.utils.debugr)rr)rr r r# memsamples rsamplesz[n_samples=10]cKs(ddlm}t}|j|d|S)z/Dump statistics of previous memsample requests.r)r)file)Z celery.utilsrioStringIOmemdumpgetvalue)rrr)routr r r#rs  rnz[N=1]cKs4|jjjrtdS|jj||j|tdS)z!Grow pool by n processes/threads.zJpool_grow is not supported with autoscale. Adjust autoscale range instead.zpool will grow)rqr autoscalerr%rrZgrow_update_prefetch_countrrrr)r r r# pool_grows   rcKs6|jjjrtdS|jj||j| tdS)z#Shrink pool by n processes/threads.zLpool_shrink is not supported with autoscale. Adjust autoscale range instead.zpool will shrink)rqrrr%rrshrinkrrrr r r# pool_shrinks  rcKs2|jjjr&|jjj|||dtdStddS)zRestart execution pool.)reloaderzreload startedzPool restarts not enabledN)r@rCZworker_pool_restartsrqrreloadrr)rmodulesrrr)r r r# pool_restart,s rmaxminz [max [min]]cCs>|jjj}|r2|||\}}td|d|StddS)zModify autoscale settings.zautoscale now max=z min=zAutoscale not enabledN)rqrrrprr)rrrrZmax_Zmin_r r r# autoscale6s  rGot shutdown from remotecKst|t|dS)zShutdown worker(s).N)rxwarningr)rmsgr)r r r#shutdownCs rrr exchange_typerz' [exchange [type [routing_key]]]cKs.|jj|jj|||pd|f|td|S)z2Tell worker(s) to consume from task queue by name.directz add consumer )rq call_soonZadd_task_queuer)rrrrroptionsr r r# add_consumerLs rzcKs |j|jj|td|S)z9Tell worker(s) to stop consuming from task queue by name.zno longer consuming from )rqrZcancel_task_queuer)rr_r r r#cancel_consumer^s rcCs |jjrdd|jjjDSgS)z:List the task queues a worker is currently consuming from.cSsg|]}t|jddqS)T)Zrecurse)r^Zas_dict)rOrr r r#rnsz!active_queues..)rqZ task_consumerZqueuesrr r r# active_queuesjs r)F)FN)FN)FN)NNN)N)N)F)NF)rrr)r)r)r)NFN)NN)r)NNN)Ur0rr collectionsrrrZbilliard.commonrZkombu.utils.encodingrZcelery.exceptionsrZcelery.platformsrrhZcelery.utils.functionalr Zcelery.utils.logr Zcelery.utils.serializationr r Zcelery.utils.timer r-rrlrZr__all__rr/rxrrr%rr=r?rArCrErSrequests __getitem__rRrn __contains__rrMrbrtr]rHr_rfloatrrrrrrrrrrrrXrWrvrintrrrrrrrrrrrr r r r#s<                6   $