$Fi$ " d Z ddlZddlZddlZddlZddlZddlmZmZmZ ddl m
Z
m Z m Z ddl
mZmZ ddlmZmZ ddlmZ dd lmZ dd
lmZ dd lmZ dd lm Z d
dl m Z m Z m Z m!Z!m"Z"m#Z# d
dl$m%Z% d
dl&m'Z'm(Z(m)Z) e' G d de Z*dS )z-
This module contains RequestProcessor class
N)datetime timedeltatimezone)ThreadRLockcurrent_thread)CallableAny)QueueEmpty)OperationalError )
AutoTracer)Common)
DecisionMaker)StatisticsSender )
session_scopesetup_database
RequestResultcleanup_old_datarestore_databaseis_malformed_database)SSAError) singleton url_splitswitch_schedstatsc e Zd ZdZdZdZd fd Zed Zde fdZ
d d
Z d d Z d d Z
d
ed efdZd defdZd Zd Zeded efd Zd Zd Zd Zde d dfdZd Zd Z xZ S ) RequestProcessorz^
SSA Request processor implementation.
Only one instance is allowed to be created
d Nc t t j d | _ | j dt t d |r|n
t | _ t | _ t | j | _
t | _ t! | j | _ t% | _ g | _ | | t/ j t. j | j t/ j t. j | j d S )N
req_processorzProcessor enabled: %sTenabled)engine)super__init__logging getLoggerloggerinfo __package__r r r& r _lockr decision_makerr senderr auto_tracerr _queue_bufferstart_background_routinestart_flush_workersignalSIGTERMshutdown_handlerSIGINT)selfr& __class__s L/opt/cloudlinux/venv/lib64/python3.11/site-packages/ssa/modules/processor.pyr( zRequestProcessor.__init__- s
'88 0+>>>$'''' &>,T[99 9 (A(A$+(N(NOOO (( n * , , 149 9 9 K$$L&((-t5 5 5 OOD$@AAAK$$G&((-t5 5 5 OOD$4555K$$J&((-t5 5 5 "__T-@ A AFK$$D&((-t5 5 5 %T[111OODOV<<< &d3333""2&&&&**4;777A 8r= actionc
|| S # t $ rZ t j \ }}}| j dt | ||dt
j i Y dS w xY w)zACall requested Callable with given args and capture any exceptionz %s failed with exception %s, %sorig_traceback)extraN) Exceptionsysexc_infor+ exceptionrF traceback
format_exc)r: ra argsetev_s r< r[ zRequestProcessor._safe_exec s T64= T T T IBA K ! !"C"%f++r2)99;O;Q;Q(R
"
T
T
T
T
T
T Ts A A+*A+ to_sleepc | j dt j | t j | dS )zP
Log and sleep given number of seconds or 15 minutes by default
z%[%s] Routine thread sleeping for (%s)N)r+ r, r rP timesleep)r: rp s r< r^ z RequestProcessor._simple_sleep sE
@')). : : :
8 r= c t d t | z
}| t | dS )zF
Sleep the number of minutes remaining till next hour
r )hours)minutesN)r
total_secondsr^ int)r: start_minute sleep_fors r< r_ z&RequestProcessor._sleep_till_next_hour sY Q''') +" +" +" "#0=?? 3y>>*****r= c d | j 5 t | d d d d S # 1 swxY w Y d S N)r. r )r: r& s r< r\ z%RequestProcessor.restore_db_with_lock s|
Z % % V $ $ $ % % % % % % % % % % % % % % % % % %s %)) timestampc J t j | t j j S )zN
Takes an hour of a day, to which the given timestamp belongs
)r
fromtimestampr rX rZ )r} s r< get_interval_forz!RequestProcessor.get_interval_for s
%i >>CCr= c @ | j }| j | t | j | j k r!| | j g | _ n*# t $ r | j d Y nw xY w)zK
Continuously flush queued request results to the database
TzFlush worker failed)
r2 getr3 appendlen BUFFER_SIZE_flush_objectsre r+ rh r: objs r< rT z RequestProcessor.flush_worker s =
=koo'' ##C(((t|$$(888'' 555#%DL
=
=
= %%&;<<<<<
= =s A1A5 5$B B c | j r[ | | j n*# t $ r | j d Y nw xY wg | _ dS # g | _ w xY wdS )zq
Flush all remaining objects even if less than BUFFER_SIZE.
Should be called once a day.
z Flush remaining objects failedN)r3 r re r+ rh r@ s r< r] z(RequestProcessor.flush_remaining_objects s
< "
"##DL1111
H
H
H %%&FGGGGG
H " r !!!!
" "s& $ A $A A
A A A c t | j D ]} t | j 5 }| | d d d n # 1 swxY w Y d S # t
$ rU}dt
| v r9| j d|dz | j t j
d n Y d }~d }~ww xY wd S )Nzdatabase is lockedz-Database is locked, retrying attempt %s/%s...r g?) rangeDB_ACCESS_RETRIESr r& bulk_save_objectsr
rF r+ warningrr rs )r: objectsattemptdbrG s r< r z RequestProcessor._flush_objects s T 344
G
"4;// 22((1112 2 2 2 2 2 2 2 2 2 2 2 2 2 2#
'3q6611K''(WY`cdYdfjf|}}}JsOOOO $OOOO
s; A A A A A A A
B;&A B66B;datac |s.| j dt j dS | d }| | r | j d| dS t | \ }}t |||d |d |d | j k |d |d |d |d
}| j
| dS ) z,
Process given request data
z [%s] has empty request, skippingNurlz
%s ignoredr} durationhitting_limitsthrottled_timeio_throttled_time wordpress) domainpathr} r is_slow_requestr r r r ) r+ r, r rP r
is_ignoreddebugr r rA r2 put)r: r r r urirequests r< handlezRequestProcessor.handle s K ? AQAQAV W W W Fhhuoo ??3 K lC 0 0 0 F nn ;'*% ,t/G G 01 01 "#6 7;'
r= c D | j d| d | | n*# t $ r | j d Y nw xY wt
j d dS # t
j d w xY w)zQ
Handle shutdown signals to flush remaining objects before exit.
zReceived shutdown signal z+, flushing queue and buffer before shutdownz1Failed to flush remaining objects during shutdownr N)r+ r, _drain_queue_to_bufferr] re rh rf exit)r: signumframes r< r8 z!RequestProcessor.shutdown_handler s
hVhhhiii ' ' ) ) ) ( ( * * * * W W W K ! !"U V V V V V W
HQKKKKKCHQKKKKs) (A B $A0-B /A00B B c | j sa | j }| j | n# t
$ r Y dS w xY w| j _dS dS )zC
Drain all remaining objects from queue to buffer.
N)r2 empty
get_nowaitr3 r r r s r< r z'RequestProcessor._drain_queue_to_buffer s +##%%
k,,.. ##C((((
+##%% s 3A
A A r| )rI N)ro ) __name__
__module__ __qualname____doc__r r r( propertyrA dictrH r4 r5 rN r r
r[ rx r^ r_ r\ staticmethodr rT r] r r r8 r
__classcell__)r; s @r< r r # s
K< < < < < <* 0 0 X0E E E E E9 9 9 9> > > >&8 &8 &8 &8PT TS T T T T c + + +% % % DC DC D D D \D = = = " " " !4 !D ! ! ! !6 r= r )+r r) rf rr r6 ri r r r threadingr r r typingr r
queuer r sqlalchemy.excr
autotracerr commonr r/ r stat_senderr r r r r r r r internal.exceptionsr internal.utilsr r r r r= r< r s
2 2 2 2 2 2 2 2 2 2 3 3 3 3 3 3 3 3 3 3 + + + + + + " " " " " " ) ) ) ) ) ) ) ) ) ) ) ) x x x x x x x x x x x x x x x x * * * * * * e e e e ev e e e e er=