z\ d dl Z d dlZd dlZd dlZd dlZd dlZd dlZd dlmZ d dl m
Z
m Z d dl m
Z
d dlmZmZ d dlmZ d dlmZmZmZ d dlmZ ej e Z e Z ej d d
d g Z G d d
e
Z G d de Z d Z! G d de" Z# G d de" Z$ G d de j% Z&dS ) N)
attrgetter)MessageReject)BaseMessageProcessor)Gen publisher)safe_cancel_task)DAY ServiceBase
rate_limit)gProcessingMessagemessage
start_timec 2 e Zd Zd Zd Zd Zd Zd Zd ZdS )TheSinkc t |t d | _ || _ t |t | j | _ | t _ d S )NPROCESSING_ORDER)key) sortedr _sinks_ordered_loop TaskManagerMessageProcessor
_task_managerr
sink)self sink_listloops W/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/internals/the_sink.py__init__zTheSink.__init__ s` $ :&899
( "4#677
c 8 | j j d| j j S )N.) __class__
__module____name__r s r __repr__zTheSink.__repr__% s .333T^5L5LMMr" c fd| j D }t | dk s
J d t t | d S )ze
introspection: decompose a specific role
:return classobj: instance or None
c 4 g | ]}t | |S )
isinstance).0r classobjs r
z%TheSink.decompose..- s8
JtX4N4N
r" zAmbiguous requestN)r lennextiter)r r/ optionss ` r decomposezTheSink.decompose( sf
! 0
7||q "5 DMM4(((r" c 8 | j dS )z
Make sure to run message processing bus only
when every MessageSource (or MessageSource+MessageSink mix)
got initialized
N)r startr( s r r8 z
TheSink.start3 s
"""""r" c X K t d | j t d | j d d {V t d | j d {V d S )Nzshutdown the sink startedzwait for current tasks timeoutzfinish wait task)loggerinfor should_stopwait_current_taskswaitr( s r shutdownzTheSink.shutdown; s /000 &&((( ,--- 33A3>>>>>>>>> &''' %%'''''''''''r" c J K | j | d {V d S N)r push_msg)r r s r process_messagezTheSink.process_messageC s5 ))'22222222222r" N) r' r& __qualname__r! r) r6 r8 rB rF r, r" r r r sq N N N ) ) )# # #( ( (3 3 3 3 3r" r c p e Zd ZdZdZdZ fdZd Zed Z d dZ
d Z d
Z e
d Z xZS )
r i r: c ~ t | t | j | _ | j | _ | j | _ || _ t j | _ t t t j | _ | t j | _ d S )N)maxsize)periodon_drop)superr! MessageQueueMAXSIZE_queue CONCURRENCY _concurrencyTIMEOUT_process_message_timeout_msg_processorweakrefWeakSettasksr r
r= warning_throttled_loggererrorthrottled_log_error)r r
msg_processorr% s r r! zTaskManager.__init__O s
"4<888 ,(, % + _&&
!+3!O!O!O #'#9#9&,#G#G r" c , K | j s/| j t | d{V dS | j j rd| j | j |f}n"d| j | j |f} | j | dS )z&Push message unless the queue is full.NzNMessage queue is full %s. Current processing messages: %s. Message ignored: %szZMessage queue is full. Queue size: %s Current processing messages: %s. Message ignored: %s) rQ fullputMessageComparabler[ should_be_calledcurrent_processing_messagesqsizer] )r msgargss r rE zTaskManager.push_msgY s { !! ,+//"3C"8"899 9 9 9 9 9 9 9 9 9%6
O K4O K%%''4
%D $d + + + +r" c > t d | j D S )Nc 3 K | ]R}| |j j t t j |j j z
d fV SdS ) N)doneprocessing_msgr roundtime monotonicr )r. tasks r z:TaskManager.current_processing_messages..{ sq
99;;
#+dn&&)<)GGKK
r" )tuplerY r( s r rd z'TaskManager.current_processing_messagest s6
r" Nc K | j rOd | j D }t d| t j | j | d {V d S d S )Nc j g | ]0\ }}| d | d |f1S )method
message_id)get)r. mlastings r r0 z2TaskManager.wait_current_tasks.. sI Aw x !%% "5"5w? r" z#Waiting for %r processing to finishr; )rY rd r= r> asynciorA )r r< msg_to_processs r r@ z TaskManager.wait_current_tasks s : < "&"B N
KK5
,tz7;;; ; ; ; ; ; ; ; ; ; < z"TaskManager._run.. s i.?.?.A.A r" z3There is still %s unprocessed messages in the queue Error during message processing:)rz BoundedSemaphorerS _should_stopr= debugrQ re _TaskManager__limit_concurrencyrw CancelledErrorr create_taskrV rf r rn ro rl add_done_callback_on_msg_processedrY addrZ exception)r msg_comparablet unprocessedr s @r _runzTaskManager._run s ,T->?? A'
" 5t{7H7H7J7JKKK 229=========+/;??+<+<%<%<%<%<%<% %r)filer zCMessage %r was not processed in the %r plugin in %ss; Traceback: %szError processing %r in %rz %s processed in %.4f secondszE%s message took longer to process than expected (%.4f sec > %.4f sec)) r report_reporter_gen_sinkrn ro r rz r rF r shieldTIMEOUT_TO_SINK_PROCESSr- r r r r r= r> strr ioStringIO print_stackseekr\ read Exceptionr PROCESSING_TIME_THRESHOLDr] ) r rf r8 r process_message_task processedr stackprocessing_times r r z MessageProcessor._call_unlocked sM 08MNNNN J * 8 * 8D)
8'.':((--( ($ #*"2
N#788 8
# # # D i11 $#C7 )
8 ""6777777777777
0#a&&#>>>2 ""67777777777771 '
$00e0<<<
1
$0JJLL
""6777777777777
!)klassrf r )formatr% r' rf r r( s r r) zMessageComparable.__repr__% s2 8??.)] @
r" )
r' r& rG __doc__r r r r r) r r s @r rb rb sm --
E \4 4 4
r" rb c 4 e Zd Z fdZdef fd Zd Z xZS )rO c t j |i | t j | _ d| j _ d| j _ d S )N2 i )rN r! reprlibRepr_repr maxstringmaxtuple)r rg kwargsr% s r r! zMessageQueue.__init__. sF $)&))) \^^
!
"
r" itemc V K t | d {V S rD )rN ra )r r r% s r ra zMessageQueue.put4 s/ WW[[&&&&&&&&&r" c t t j d | j D d d }d| j d| d| j | dS ) Nc 0 g | ]}|j j j S r, )rf r% rG )r. r s r r0 z(MessageQueue.__str__..; s IIIT#0IIIr" c | d S r r, )r s r r z&MessageQueue.__str__..= s
T!W r" T)r reversez) r collectionsCounterrQ itemsrK re r repr)r
msg_countss r __str__zMessageQueue.__str__7 s IIT[III
egg$$
r s$ > > > > > > > > B B B B B B M M M M M M M M 2 2 2 2 2 2 E E E E E E E E E E 4 4 4 4 4 4 8 $ $SUU *K*)\2
*3 *3 *3 *3 *3" *3 *3 *3ZwM wM wM wM wM+ wM wM wMt% % %
K K K K Kv K K K\
2
7(
r"