9Mj d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl
Z
ddl Z ddl Z ddl
mZmZmZmZmZ ddl mZmZmZmZ ddlmZ ddlZddlmZ ddl m Z dd l m Z dd
l!m"Z"m#Z#m$Z$ dd l%m&Z&m'Z' dd l(m)Z) dd
l*m+Z+m,Z,m-Z- ddl.m/Z/ ddl0m1Z1 ddl2m3Z3m4Z4 e j5 r
ddl6m7Z7m8Z8m9Z9 e G d d Z: ej; e< Z= G d d Z>dS )zJ
This module contains contains classes implementing X-Ray Agent behaviour
N)Threadcurrent_threadLockEventTimer)AnyOptionalDictTuple) dataclass )BoundedThreadExecutor)gettext)
get_client)local_tasks_storagetasks_base_storagesafe_id_pattern) XRayError XRayAPIError)
FaultDetector)open_local_storageflush_memory_storage get_task_ids)Task)
extract_creds) dbm_storage get_current_cpu_throttling_time)
SendClientSmartAdviceAPIClient APIClientc $ e Zd ZU ded< eed< dS )APIDataContainerr clienttaskN)__name__
__module__ __qualname____annotations__r H/opt/cloudlinux/venv/lib64/python3.11/site-packages/xray/agent/daemon.pyr" r" - s%
JJJJJr* r" c
| e Zd ZdZdZdZdZdZ ej dz ej dz ej dz d d d fd
Z
d Z d Z dd.Z*d'J # #P_cdPd
$ +[ 8( /?.J**PSTUWbfgWgPhPh 0C/N++T_
v /&.1ABBB =Gy">">D % % ' ' '
"d.>TM^ _ _ _ co ! 0
" ,%,,,&*kkmm#
G'$++D,KZYeffff z ' ' 'LLW $$&&&&& ' $QJE !,%,,, sZ E%5 E
C10E
1=D1.E
0D11E
E%
E E%E E%%E),E)c D | || \ |d< |d< dS )zj
Calculate throttling values and update given data
with resulting throttling stat
hitting_limitsthrottled_timeN)rK )rV datat_key cpu_values r+ add_limit_faultszAgent.add_limit_faults s1
:>9L9LUT]9^9^6
%5 6 6 6r* php_pid cpu_usage
caller_uidc t d||| | j ||| | j dS )z
Called when php request starts and sends us welcome
request meaning that request started on the php side.
z=Received request init trigger from php=%s uid=%s cpu_usage=%s)uidN)rp r_ rK saveflush)rV r r r s r+ _handle_request_initzAgent._handle_request_init s[
SU\^hjsttt
) DDD !!#####r* request_datac 2 | d | ||| t dt j | d | d | || d S )Nr z&[%s] Processing trace for task %s (%s)tracing_task_idurl)getr rp r_ r nameprocess_request_data)rV r r r r s r+ _handle_request_endzAgent._handle_request_end s , - - 5 ! !, C C C 4 ! . / / U # #
!!,
;;;;;r* i @ g @g >@r r c J t | \ }}}n?# t $ r2 t d | Y dS w xY w| j 5 | j |d | j k r=t d| | ddd dS | j |xx dz
cc< ddd n # 1 swxY w Y | |||| | j 5 | j |xx dz cc< | j | dk r| j |= ddd dS # 1 swxY w Y dS # | j 5 | j |xx dz cc< | j | dk r| j |= ddd w # 1 swxY w Y w xY w)a
Handle incoming connection
:param connection: socket object usable to
send and receive data on the connection
:param workers_pool: pool where we can place
tasks for the futher processing
z1Failed to extract credentials, closing connectionNr z6Too many concurrent connections from uid=%s, rejectingr
)
r rd rp re r rP rO r r< _process_connection)rV r r _pid_uid_gids r+ r z Agent.handle_incoming_connection= s ,Z 8 8 D$ NNN O O O FF
' - -$((q11T5QQQ WY]^^^ """ - - - - - - - -
!$ ' ' '1 , ' ' ' - - - - - - - - - - - - - - - 4 $ $Z tT J J J+
4
4%d+++q0+++(.!33 -d3
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4+
4
4%d+++q0+++(.!33 -d3
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4
4sj 8AA AC >C C#&C#+E
/EE
E
F"/F
F"F F" F F"r r c t | } | j d }t fd}t | j | }| | | } n# t j
$ rp}
t dt |
dt j i Y d }
~
| d S d }
~
wt$ t& j f$ rH t d|| Y | d S w xY w | n-# | w xY w r$t d| j || d S | "| ||| j ||| d S | ||| j ||| |
d S ) Nignore)errorsc t j d S # t $ r Y d S w xY wN)setshutdownsocket SHUT_RDWRrd )r deadline_exceededs r+ _abort_on_deadlinez5Agent._process_connection.._abort_on_deadlinei sY ! ! # # #
##F$455555
s 8
AAzJSON decode failed: %st_nameextraz5Connection read timed out or failed for pid=%s uid=%szAConnection read exceeded %ss deadline for pid=%s uid=%s, dropping)r r r )r r r r )r
settimeout
_READ_TIMEOUTmakefiler r _MAX_READ_DURATIONr
read_inputjsonJSONDecodeErrorrp r strr r cancelr rd r timeoutre is_set_submit_uid_jobr r ) rV r r r r current_cpufileobjr watchdog
input_dataer s ` @r+ r zAgent._process_connection[ s 6d;; d 0111!+!4!4H!4!E!E "GG 02DEE 11JJ# LL13q66(NL\L\LaAbL c c c FFF
OO
( NNRTXZ^ _ _ _ OO
OO
OO
# # % %
NNSUYUlnrtx
F dD$=tWbos
!
( % '
!
s< B F E1%=D"F 5E1F 0E11F *Gc t | j || j v rdndz }t | j | j |z }| j | j z
}t | j t
|| S )a Dynamic max-min fair-share cap for one uid, in outstanding jobs.
MUST be called while holding _uid_jobs_lock (it reads _uid_jobs).
active = number of currently-active uids (those with > 0 outstanding
jobs). _uid_jobs is kept free of zero-count entries, so its size is the
active set; _uid itself is counted even on its first job (it is about
to become active), so a flood never under-counts itself out of a share.
The cap is:
max(min_floor, min(capacity - reserve, max(min_floor, capacity // active)))
* active == 1 -> capacity // 1 == capacity, clamped to capacity-reserve
(e.g. 80-16=64): a lone user is bounded only by the real queue.Full,
no regression, yet still leaves `reserve` slots for a new tenant.
* active == N -> capacity // N: a fair split; a flooder is held to its
share going forward and converges to it as its jobs drain.
* min_floor stops the cap collapsing to 0 with very many active uids.
* the OUTER max(min_floor, ...) keeps the cap >= min_floor even when the
capacity - reserve clamp is <= 0 (e.g. maxqueuesize == 0, so
capacity == max_workers == reserve): without it the cap would be 0 and
_submit_uid_job would drop EVERY request from EVERY uid (total
self-DoS). On such a tiny pool the per-uid cap is min_floor and the
executor's real queue.Full (BoundedSemaphore = maxqueuesize +
max_workers) is the backstop.
r r
)lenrQ r> r? r= r@ min)rV r active
fair_share
reserve_clamps r+ _effective_uid_capzAgent._effective_uid_cap sl 6 T^$$TT^-C-CK $*=*GHH
+dm;
4?C
z$B$BCCCr* c
j 5 } j d |k rqt d j d |t
j j v rdndz j j ddd dS j xx dz
cc< xj dz
c_ ddd n # 1 swxY w Y fd} |
| dS # t j $ r3
t d Y dS t $ r
w xY w)uM
Submit a worker job into the shared workers_pool, enforcing a DYNAMIC
max-min per-UID fair share of the real pool capacity, UNCONDITIONALLY.
At submit time the per-UID cap is recomputed from the number of
currently-active uids (see _effective_uid_cap): a sole active uid gets
the whole capacity minus a reserved headroom (no regression — bounded
only by the real queue.Full), while with N active uids each gets
capacity // N (fair split). There is NO pressure gate: the cap alone
provides both no-regression for a lone user and anti-monopolization
under contention, and a flooder cannot defeat it by supplying its own
load. The reserved headroom guarantees a newly-arriving tenant always
finds free slots even when one uid has filled its share.
r zPer-UID fair-share drop: uid=%s holds %s outstanding jobs (dynamic fair-share cap %s for %s active uid(s), pool capacity %s, reserve %s); dropping this request to protect other tenantsr
Nc t di S # w xY w)Nr) )_release_uid_job)r fnkwargsrV s r+ _jobz#Agent._submit_uid_job.._job sH
,r||F||%%d++++%%d++++s 7zRequest %s was rejected because our workers thread pool is full of tasks. Increase queuemaxsize or max_threads in configuration.)rS r rQ r rp re r r= r@ rR r r r r r
BaseException)rV r r r r
effective_capr s ` ``` r+ r zAgent._submit_uid_job s%
& & 3 3D 9 9M~!!$**m;;E N&&tQ//!''0F0F11AN'M & & & & & & & &
N4 A % A % # & & & & & & & & & & & & & & &, , , , , , , , , % % % % %z
! !$ ' ' ' LL[
FF ! !$ ' ' ' s* BC#2%C##C'*C'7D >E/ E/c | j 5 | j |xx dz cc< | j | dk r| j |= | j dk r| xj dz c_ ddd dS # 1 swxY w Y dS )a* Decrement the per-UID and total outstanding-job counters under the
lock. The total counter is floored at 0 so it can never underflow if a
release is somehow called more often than expected, and zero-count
per-UID entries are deleted to keep _uid_jobs from growing unbounded.r
r N)rS rQ rR )rV r s r+ r zAgent._release_uid_job s
* * N4 A % ~d#q((N4(#a''$$)$$ * * * * * * * * * * * * * * * * * *s A
A A#&A#fileioc $ | | j }t dt | t | dk rdS t
j | d S )zy
Read input data and return decoded json
:param fileio: a file-like object providing read method
zReceived %d bytesr NF)strict)read_MAX_READ_SIZErp debugr stripr loads)rV r r s r+ r zAgent.read_input sp
{{4.// (#d))444 tzz|| ! ! Fz$**,,u5555r* fake_task_idr c t t 5 } || }nA# t $ r4 t t d || d w xY w ddd n # 1 swxY w Y n># t $ r1}t t d t | z d}~ww xY w| | j
| S )z<
Initialize MongoDB client for current task
z9Cannot resolve tracing_task_id: no match found in storage)idall_idsr Nz"Cannot resolve tracing_task_id: %s)r6 r ) r r decodeKeyErrorr _keys RuntimeErrorr rG r8 )rV r task_storagereal_idr s r+ instantiate_mongo_clientz Agent.instantiate_mongo_client sE
N 011
\ *< 8 ? ? A AGG #UVV%1l>O>O>Q>Q R R
N N NA B C Cc!ff LMM M N &&g&VVVsN B
B3B>A11B5 B
BB
B B
C,CCc t d| j | j | }||j |j fS | | }| }t d||| | j 5 | j | }||j |j fcddd S t | j | j
k rit t | j }t
dt | j | | j | t || | j |< ddd n # 1 swxY w Y ||fS )z
Returns a client and task from cache of API data or
initialize client and GET task from MongoDB and add to cache
zCached API data: %sNz,Adding new container in cache: %s --> %s, %sz9api_data_cache at capacity (%d), evicting oldest entry %s)r# r$ )rp r rJ r r# r$ r get_taskrH r _MAX_CACHE_ENTRIESnextiterr_ popr" )rV r cached_datary _t
oldest_keys r+ rq zAgent.get_cached_or_load s
*D,?@@@ )--l;; " %{'77 711,??
!
! C\SUW`aaa
% \ \ -11,??K&");+;; \ \ \ \ \ \ \ \ 4&''4+BBB !$t':";"; < <
OQTUYUhQiQiku #''
3330@ XZ0[0[0[D - \ \ \ \ \ \ \ \ \ \ \ \ \ \ \ "} s *E:B E::E>E>c t t 5 }d | D }ddd n # 1 swxY w Y n+# t $ r t d Y dS w xY wt
| j D ]c}| j 5 || j v r9||vr5t d| | j
| ddd n # 1 swxY w Y ddS )z
Cleanup an API data im-memory cache dict in order not store
inactive (stopped, already completed) tasks there
c 6 g | ]}| S r) )r ).0ks r+
z0Agent.cleanup_api_data_cache..H s H H Hq
H H Hr* Nz,Unable to cleanup cache, storage unavailablez!Cleaning up inactive container %s) r r r r rp re listrJ rH r_ r )rV r active_tasks_tasks r+ r z Agent.cleanup_api_data_cacheA s
011
I\ H HL4E4E4G4G H H H
I
I
I
I
I
I
I
I
I
I
I
I
I
I
I NNI J J J FF $-224455 3 3E)
3
3D ///E 4M4MKK CUKKK'++E222
3
3
3
3
3
3
3
3
3
3
3
3
3
3
3 3 3sF A
A A
AA
A A
$A54A5'AC66C: =C: c |d }|t j | s t d| dS t j t t | | }t j | s t d|| dS |
| \ }}t d|j t |d |j 5 }| |j |j dk r(t d ddd dS | || }t d
| | t d | | n# t* $ r Y nw xY w|xj d z
c_ | |j ddd n # 1 swxY w Y |j r| |j |j d
k r"|j dk r| | dS dS dS )aK
Increment request ID in /usr/share/alt-php-xray/requests/{tracing_task_id} file
Substitute request_id and tracing_task_id in request_data.
Send request_data to ClickHouse
:param request_data: original request data
:param caller_uid: UID of the connecting process (from SO_PEERCRED)
r Nz*Rejecting request with invalid task id: %szSRejecting request from uid=%s for task %s: no matching task file in per-UID storagezProcessing task: %s)r rm r z*Tracing count is 0, nothing should be donezSending to ClickHousezSending to SmartAdvicer
ro )r matchrp re rb pathjoinr r isfilerq r_ rw r is_manualrr rn rv update_request_datarE rF r ra fake_idrs
complete_task) rV r r r owner_task_filer r$ rz updated_request_datas r+ r zAgent.process_request_dataS s $$56 !"( 66
K\ZZZ gll+=s:P\ ] ]O7>>/22
i
)),774 )4<888
-> ?t~
V
V
V QZa ' '8O ' P P P!Q&& HIII
Q Q Q Q Q Q Q Q $(#;#;L$#O#O KK/ 0 0 0 1 2 2 2
4555##$89999
# #q ( # # ' '8O ' P P P- Q Q Q Q Q Q Q Q Q Q Q Q Q Q Q0 > 7 ( ( 6 6 6 ?f $ $);q)@)@ t $ $ $ $ $ % $)@)@s> AH AH /GH
GH G.H HHr r$ c |j dz |d< |j |d< |d D ]C}|j dz |d< |j |d< |d dk r | |d |d< Dt d| d | d t d t | dg | d |S )
z
Substitute request_id and tracing_task_id
:param data: original input
:param task: a Task instance
:return: updated input
r
request_idr r type mysql_queryqueryz5Input updated: tracing_task_id = %s & request_id = %sz2Updated input: %d item(s) for tracing_task_id = %s)ru rw hide_symbolsrp r_ r r r )rV r r$ items r+ r
zAgent.update_request_data s " /! 3\"&,
L A AD!%!3a!7D &*lD" #F|},, $ 1 1$w- @ @W
CTXXN_E`E`bfbjbjkwbxbx
@#dhhvWYFZFZB[B[]a]e]efw]x]x
r* r# ru rv c 4 | || dS )z:
Update task counters in mongodb instance
)rv ru N)update_counts_only)rV r# ru rv s r+ rt zAgent.update_counts s# !!
]![[[[[r* r c t d|j t j d | |j dS )z_
Stop and complete request_qty task
:param _task: tracing task to stop
zTask %s should be completedr
N)rp r_ rw ro r} _run_complete_task_cmd)rV r s r+ r zAgent.complete_task sD
15=AAA
1
##EM22222r* c t j ddd| j d|gdt j d S # t j t j f$ r'}t d|| Y d }~d S d }~ww xY w)Nzcloudlinux-xray-managerstopz --system_idz--tracing_task_idx )r stderrz Failed to complete task %s: %s)
subprocess check_outputr8 PIPETimeoutExpiredCalledProcessErrorrp r )rV rw r s r+ r z Agent._run_complete_task_cmd s G #*FM4;Pcelm!
):+HI G G G LL97A F F F F F F F F F Gs ,0 A2 A--A2r c \ d }t j d }t j ||| S )zA
Sanitize data in single quotes from MySQL query
c | d }| d }d }t j d|| d }| | |p| S )zL
Works with whole string in single or double quotes
quotetruncc b | d rdS | d rdS dS )zE
Works with characters inside quotes
digit0symbolxN)group)inner_ms r+
inner_replz8Agent.hide_symbols..replacer..inner_repl s@ ==)) 3 ]]8,, 3 r* z.((?P\d)|(?P[^0-9_:;\-/',. \\]))in_quote)r/ resub)mqtr1 sanitizeds r+ replacerz$Agent.hide_symbols..replacer sx A A
PR\^_^e^efp^q^qrrI ,,AF,, ,r* zH(?P['"])(?P.*?)((?\.{3})))r3 compiler4 )r r9 patterns r+ r zAgent.hide_symbols s7 - - -0 * lmmvgx555r* r )FN)r N)2r% r& r' __doc__r~ r r r psutil cpu_countr[ rg rk ra r r boolr rN r r r rI r r r r r r r r r r io
TextIOBaser r r r r r rq r r r
rt r r staticmethodr r) r* r+ r- r- 6 s ! )(**Q.$F$&&*%V%''", $ Y5 Y5 Y5 Y5v U U U Z Z Z V V V V23 3 3*% % &- % T % RZ[^R_ % ko % % % % N_ _ _
$
$C
$C
$XVY]
$
$
$
$
<3
<3
->7L>TW>_b>
> > > >@ Ds Ds D D D D@:,A : :_c : : : :x
*S
*T
*
*
*
* 6 63 6 6 6 6WS W[ W W W W$ s u[$=N7O B3 3 3 3$8% 8% 8%3 8%RV 8% 8% 8% 8%t D T .\ \K \ \T\]`Ta \mq \ \ \ \ 34 3D 3 3 3 3G G G 6# 6# 6 6 6 \ 6 6 6r* r- )?r< rL r@ r r^ rb r r3 rA r r! ro typing threadingr r r r r r r r
r dataclassesr r= executorr xrayr r xray.apiclientr xray.internal.constantsr r r xray.internal.exceptionsr r rK r xray.internal.local_countersr r r xray.internal.typesr xray.internal.user_plugin_utilsr xray.internal.utilsr r
TYPE_CHECKINGxray.apiclient.api_clientr r r r" getLoggerr% rp r- r) r* r+ rR sr
@ @ @ @ @ @ @ @ @ @ @ @ @ @ - - - - - - - - - - - - ! ! ! ! ! !
+ + + + + + % % % % % % \ \ \ \ \ \ \ \ \ \ < < < < < < < < ) ) ) ) ) ) _ _ _ _ _ _ _ _ _ _ $ $ $ $ $ $ 9 9 9 9 9 9 L L L L L L L L VUUUUUUUUUU
8 $ $g
6 g
6 g
6 g
6 g
6 g
6 g
6 g
6 g
6 g
6r*