File: //home/arjun/projects/env/lib/python3.10/site-packages/flower/__pycache__/events.cpython-310.pyc
o
/weN � @ s� d dl Z d dlZd dlZd dlZd dlZd dl mZ d dlmZ d dlm Z d dl
m Z d dl mZ
d dl mZmZ d dlmZ d dlmZ e�e�Zdad d
� ZG d d � d �ZG d
d� de �ZG dd� dej�Z dS )� N)�Counter)�partial)�
EventReceiver)�State)�Gauge� Histogram)�PeriodicCallback)�optionsc C s t d u rt� a t S �N)�PROMETHEUS_METRICS�PrometheusMetrics� r
r
�F/home/arjun/projects/env/lib/python3.10/site-packages/flower/events.py�get_prometheus_metrics s r c @ s e Zd Zdd� ZdS )r c C sr t ddg d��| _tddddgtjd�| _td d
ddg�| _td d ddg�| _td
ddg�| _ tdddg�| _
d S )N�flower_events_totalzNumber of events)�worker�type�task�flower_task_runtime_secondsz Task runtimer r )�buckets�!flower_task_prefetch_time_secondszDThe time the task spent waiting at the celery worker to be executed.� flower_worker_prefetched_tasksz4Number of tasks of given type prefetched at a worker�flower_worker_onlinezWorker online status�1flower_worker_number_of_currently_executing_tasksz/Number of tasks currently executing at a worker) �PrometheusCounter�eventsr r �task_runtime_metric_buckets�runtimer �
prefetch_time�number_of_prefetched_tasks�
worker_online�*worker_number_of_currently_executing_tasks��selfr
r
r �__init__ s. ���
�zPrometheusMetrics.__init__N)�__name__�
__module__� __qualname__r$ r
r
r
r r s r c s( e Zd Z� fdd�Z� fdd�Z� ZS )� EventsStatec s* t � j|i |�� t�t�| _t� | _d S r
)�superr$ � collections� defaultdictr �counterr �metrics)r# �args�kwargs�� __class__r
r r$ < s zEventsState.__init__c s� t � �|� |d }|d }| j| | d7 < |�d�r�|d }| j�|�}|�dd�}|s:|| jv r:|jp9d}| jj� |||��
� |�dd �}|rX| jj � ||�� |� |j
}|j} |d
krq|jsq| rq| jj� ||��
� |d kr�|js�|r�| r�| jj� ||��|| � | jj� ||��� |d v r�|js�|r�| r�| jj� ||��d � |d
kr�| jj� |��d� |dkr�| jj� |��d� |�d�}
|
d ur�| jj� |��|
� |dkr�| jj� |��d � d S d S )N�hostnamer � ztask-�uuid�name� r r z
task-receivedz task-started)ztask-succeededz task-failedz
worker-onlinezworker-heartbeat�activezworker-offline)r) �eventr, �
startswith�tasks�getr5 r- r �labels�incr �observe�started�received�etar r �set�decr r! ) r# r8 � worker_name�
event_type�task_idr � task_namer � task_started�
task_received�num_executing_tasksr0 r
r r8 A sB
�zEventsState.event)r% r& r' r$ r8 �
__classcell__r
r
r0 r r( 9 s r( c @ sN e Zd ZdZ ddd�Zdd � Zd
d � Zd d
� Zdd� Zdd� Z dd� Z
dS )�Eventsi� NFTr c K s� t j�| � d| _|| _|| _|| _|| _|| _d | _ d | _
| jrBt � d| j� t
�| j�}|r5|d | _ |�� |rBt| j|�| _
| j sMtdi |��| _ t| j| j�| _d S )NTzLoading state from '%s'...r r
)� threading�Threadr$ �daemon�io_loop�capp�db�
persistent�
enable_events�state�state_save_timer�logger�debug�shelve�open�closer �
save_stater( �on_enable_events�events_enable_interval�timer) r# rQ rP rR rS rT �state_save_intervalr/ rU r
r
r r$ u s0
�
�zEvents.__init__c C sH t j�| � | jrt�d� | j�� | jr"t�d� | j�� d S d S )Nz Starting enable events timer...z Starting state save timer...)rM rN �startrT rW rX r_ rV r" r
r
r ra � s
�z Events.startc C sJ | j r
t�d� | j�� | jrt�d� | j�� | jr#| �� d S d S )Nz Stopping enable events timer...z Stopping state save timer...)rT rW rX r_ �stoprV rS r\ r" r
r
r rb � s
�z Events.stopc C s d} z5|d9 }| j �� �"}t|d| ji| j d�}d}t�d� |jd d dd� W d � n1 s2w Y W nG ttfy[ zdd l }W n
t
yT dd l }Y nw |� � Y n% t
y } zt�d ||� tj|dd
� t�|� W Y d }~nd }~ww q) Nr3 T� �*)�handlers�appzCapturing events...)�limit�timeout�wakeupr z;Failed to capture events: '%s', trying again in %s seconds.)�exc_info)rQ �
connectionr �on_eventrW rX �capture�KeyboardInterrupt�
SystemExit�_thread� ImportError�thread�interrupt_main� Exception�error�time�sleep)r# � try_interval�conn�recvrr �er
r
r �run� s< �
�� � ����z
Events.runc C s4 t �d| j� tj| jdd�}| j|d< |�� d S )NzSaving state to '%s'...�n)�flagr )rW rX rR rY rZ rU r[ )r# rU r
r
r r\ � s
zEvents.save_statec C s | j �d | jjj� d S r
)rP �run_in_executorrQ �controlrT r" r
r
r r] � s zEvents.on_enable_eventsc C s | j �t| jj|�� d S r
)rP � add_callbackr rU r8 )r# r8 r
r
r rl � s zEvents.on_event)NFTr ) r% r& r' r^ r$ ra rb r| r\ r] rl r
r
r
r rL q s
�
rL ) r* �loggingrY rM rv r � functoolsr �
celery.eventsr �celery.events.stater �prometheus_clientr r r �tornado.ioloopr �tornado.optionsr � getLoggerr% rW r r r r( rN rL r
r
r
r �<module> s&
8