0x1998 - MANAGER
Düzenlenen Dosya: agent.cpython-37.pyc
B �V�d�D � @ s d Z ddlZddlZddlZddlZddlZddlZddlZddlm Z m Z m Z ddl m Z mZmZmZ ddlmZ ddlmZ ddlmZmZ dd lmZmZ dd lm Z dd l m Z m Z m Z dd l!m"Z" dd l#m$Z$ ddl%m&Z&m'Z'm(Z(m)Z)m*Z* eG dd� d��Z+G dd� d�Z,dS )zJ This module contains contains classes implementing X-Ray Agent behaviour � N)�Thread�current_thread�Lock)�Any�Optional�Dict�Tuple)� dataclass� )� get_client)� agent_sock�local_tasks_storage)� XRayError� XRayAPIError)� FaultDetector)�open_local_storage�flush_memory_storage� get_task_ids)�Task)� extract_creds)� read_sys_id� create_socket� write_sys_id� dbm_storage� get_current_cpu_throttling_timec @ s e Zd ZU ded< eed< dS )�APIDataContainer� APIClient�client�taskN)�__name__� __module__� __qualname__�__annotations__r � r# r# �;/opt/alt/python37/lib/python3.7/site-packages/xray/agent.pyr * s r c @ s\ e Zd ZdZdZdZdZdZdDdd�Zd d � Z d d � Z dEdd�Z dd� Z dd� Z d d�dd�Zdd� Zdd d�dd �Zd ed �d d �Zed d!�d"d#�Zed$d%�d&d'�Zeeed(�d)d*��Zeed+�d,d-�Zee d.ef d%�d/d0�Z d d�d1d2�Z ed d3�d4d5�Z eeed6�d7d8�Z dFd$e!e"e! d d9�d:d;�Z#ed d<�d=d>�Z$d?d@� Z%eeedA�dBdC��Z&d S )G�Agentz X-Ray Agent class z/var/run/xray-agent.pid� �< i�� Fc C s� t �t jt jh� t�d�| _t� | _t | j� t d�}|| jd�| _ t d�}|� | _ t � | _ t� | _t� | _t� | _|r�t| jd�| _| j�� t| jd�| _| j�� |r�| �� | �� d S )N�agentZapi_req)� system_idZadviser)�target) �signal�pthread_sigmask� SIG_BLOCK�SIGUSR2�loggingZ getLogger�loggerr �sys_idr r � send_client�adviser_client�task_client_objectr �api_data_cache_lock�dict�api_data_cacher �fault_detectorr �_setup_signal_handlerZsignal_handler_thread�start�_flusherZflusher_thread� _save_pid�listen)�self�save_pid�background_routineZclickhouse_client_objectZadviser_client_objectr# r# r$ �__init__; s( zAgent.__init__c C sl t �t jh�}t�d|j� | �� t� t�d� yt� |jt j� W n t k rf t� d� Y nX d S )Nz?Received SIGUSR2 from pid=%s, flushing database storage on diskz;Sending signal back to process that requested storage flushz5Process that requested storage flush no longer exists) r+ � sigwaitinfor. r/ �info�si_pid�_flush_mongodb_countersr �os�kill�OSError�warning)r> Zsiginfor# r# r$ �_wait_for_sigusr2_ s zAgent._wait_for_sigusr2c C sH t �t jt jh� x0y | �� W q tk r> t�d� Y qX qW dS )z� Setup SIGUSR2 handler that starts in-memory storage flush when received. When flushed, send SIGUSR2 back to the process that send signal. z4Unable to process signal, see traceback for details.N)r+ r, r- r. rJ � Exceptionr/ Z exception)r> r# r# r$ r9 m s zAgent._setup_signal_handlerNc C s� |d k r|gnt � }x�|D ]�}| j�d|� y| �|�\}}W n" tk r` t�d|� wY nX t|��}|j|j d� W d Q R X |j dkr�| � ||j � q| � ||j |j � qW d S )NzFlushing task to mongo %sz Unable to get client and task %s)�next_request_id�time)r r0 rC �get_cached_or_loadr r/ rI r �update_with_local_datarL � tracing_by� update_counts� request_count� tracing_count)r> �task_idZtasks_to_flush� apiclientr �storager# r# r$ rE { s z Agent._flush_mongodb_countersc C s. t | jd��}|�tt�� �� W dQ R X dS )zJ Save agent process pid to file that other tools can use. �wN)�open�PIDFILE�write�strrF �getpid)r> �fr# r# r$ r< � s zAgent._save_pidc C st d}d}xft �| j� t � � | | jkrb| �� t� t � � }t � � | | jkrl| �� t � � }q tdd� q W dS )z] This method flushes data from memory to local storage periodically. r F)�removeN)rM �sleep�COUNTERS_FLUSH_INTERVAL�MONGO_FLUSH_INTERVALrE r �CLEANUP_INTERVAL�cleanup_api_data_cache)r> Zlast_mongo_flush_timeZlast_api_data_cache_cleanupr# r# r$ r; � s zAgent._flusher)�returnc C sH t t�}x:|�� \}}t| j|fd�}|�� | j�d|�� � q W dS )z( Start listening socket )r* �argsz [%s] StartedN) r r Zacceptr �handler: r0 �debug�getName)r> Z_socket� connectionZaddress�tr# r# r$ r= � s z Agent.listenc C s | � ||�\|d< |d< dS )zj Calculate throttling values and update given data with resulting throttling stat �hitting_limitsZthrottled_timeN)r8 )r> �dataZt_keyZ cpu_valuer# r# r$ �add_limit_faults� s zAgent.add_limit_faultsz socket object)ri rd c C s� t |�\}}}t|�}|jdd�}y| �|�}W nJ tjk rz } z*| jjdt|�dt � j id� |� � dS d}~X Y nX |dkr�| j � ||� | j �� |� � dS zDy*|�d�dkr�| �|||� | �|� W n tk r� Y nX W d|� � X dS )z� Handle incoming connection :param connection: socket object usable to send and receive data on the connection �ignore)�errorszJSON decode failed: %sZt_name)�extraNrk )r r Zmakefile� read_input�jsonZJSONDecodeErrorr0 �errorr[ r �name�closer8 Zsave�flush�getrm �__call__r ) r> ri Z_pidZ_uidZ_gidZ current_cpuZfileobjZ input_data�er# r# r$ rf � s, z Agent.handlez file object)�fileiord c C sb |� � }| j�dt� �� t|�� �� | j�dt� �� |� t|�� �dkrPdS t j |�� dd�S )zy Read input data and return decoded json :param fileio: a file-like object providing read method z [%s] I received %i bytes tracez[%s] I received data: %sr NF)�strict) �readr0 rC r rh �len�encoderg �striprr �loads)r> rz rl r# r# r$ rq � s zAgent.read_input)� decoded_datard c C s2 | j �dt� �� |�d�|�d�� | �|� dS )z' Process received data z&[%s] Processing trace for task %s (%s)�tracing_task_idZurlN)r0 rC r rh rw �process_request_data)r> r� r# r# r$ rx � s zAgent.__call__zAPIClient instance)� fake_task_idrd c C s� yTt t��B}y|| �� }W n* tk rF td||�� d�d��Y nX W dQ R X W n4 tk r� } ztdt|�� ���W dd}~X Y nX | j| j |d�S )z< Initialize MongoDB client for current task z9Cannot resolve tracing_task_id: no match found in storage)�idZall_ids)rp Nz Cannot resolve tracing_task_id: )r) r� ) r r �decode�KeyErrorr �keys� RuntimeErrorr[ r4 r1 )r> r� � task_storageZreal_idry r# r# r$ �instantiate_mongo_client s $$z Agent.instantiate_mongo_client)rl rd c C s | d S )zj Retrieve tracing task ID from the input data :return: tracing task ID (fake one) r� r# )rl r# r# r$ � get_task_id s zAgent.get_task_id)r rd c C s" |j r | j�dt|j�� dS dS )z- Check if task is continuous z'Task with id: %s detected as continuousTF)Z is_continuousr0 rC r[ rT )r> r r# r# r$ �is_continuous_task s zAgent.is_continuous_taskr c C sn | j �d| j� | j�|�}|dk r0|j|jfS | �|�}|�� }| j �d|||� t||d�| j|< ||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, %s)r r ) r0 rg r7 rw r r r� Zget_taskr )r> r� Z cached_datarU Z_tr# r# r$ rN ( s zAgent.get_cached_or_loadc C s� y*t t��}dd� |�� D �}W dQ R X W n tk rJ | j�d� dS X xTt| j�� �D ]B}| j�2 || jkr�||kr�| j� d|� | j� |� W dQ R X q\W dS )z� Cleanup an API data im-memory cache dict in order not store inactive (stopped, already completed) tasks there c S s g | ] }|� � �qS r# )r� )�.0�kr# r# r$ � <listcomp>B s z0Agent.cleanup_api_data_cache.<locals>.<listcomp>Nz,Unable to cleanup cache, storage unavailablez!Cleaning up inactive container %s) r r r� r� r0 rI �listr7 r5 rC �pop)r> r� Z active_tasks�_taskr# r# r$ rc ; s z Agent.cleanup_api_data_cache)� request_datard c C s t � �� }| �| �|��\}}t�d|j� t| �|�|jd���}|j|j d� |j dkrj| j �d� dS | � ||�}| j �d|� | � |� y | j �d|� | �|� W n tk r� Y nX | j d 7 _ |j|j d� W dQ R X |jr�| �|j� |jd k�r|j dk�r| �|� dS ) z� 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 z Is Manual: %s)rv )rL r z*Tracing count is 0, nothing should be doneNz[%s] Sending to ClickHousez[%s] Sending to SmartAdvicer rM )r rh rN r� r/ rC Z is_manualr rO rL rS r0 �update_request_datar2 r3 r rE Zfake_idrP � complete_task)r> r� rj �_r rV Zupdated_request_datar# r# r$ r� N s0 zAgent.process_request_data)rl r rd c C s� |j d |d< |j|d< xF|d D ]:}|j d |d< |j|d< |d dkr"| �|d �|d< q"W | j�dt� �� |�d�|�d�� | j�d t� �� |� |S ) z� Substitute request_id and tracing_task_id :param data: original input :param task: a Task instance :return: updated input r Z request_idr� rl �type� mysql_queryZqueryz [%s] Input updated => %s & %sz[%s] Full updated input %s) rR rT � hide_symbolsr0 rC r rh rw rg )r> rl r �itemr# r# r$ r� w s zAgent.update_request_data)r rR rS rd c C s |j ||d� dS )z: Update task counters in mongodb instance )rS rR N)Zupdate_counts_only)r> r rR rS r# r# r$ rQ � s zAgent.update_counts)r� rd c C s2 | j �dt� �� |j� t�d� | �|j� dS )z_ Stop and complete request_qty task :param _task: tracing task to stop z [%s] Task %s should be completedr N)r0 rC r rh rT rM r_ �_run_complete_task_cmd)r> r� r# r# r$ r� � s zAgent.complete_taskc C s t �ddd| jd|g� d S )Nzcloudlinux-xray-manager�stopz --system_idz--tracing_task_id)� subprocessZ check_outputr1 )r> rT r# r# r$ r� � s z Agent._run_complete_task_cmd)r� rd c C s dd� }t �d�}t �||| �S )zA Sanitize data in single quotes from MySQL query c S sD | � d�}| � d�}dd� }t�d|| � d��}|� |� |p>|� �S )zL Works with whole string in single or double quotes ZquoteZtruncc S s | � d�rdS | � d�r dS dS )zE Works with characters inside quotes Zdigit�0Zsymbol�xN)�group)Zinner_mr# r# r$ � inner_repl� s z8Agent.hide_symbols.<locals>.replacer.<locals>.inner_replz.((?P<digit>\d)|(?P<symbol>[^0-9_:;\-/',. \\]))Zin_quote)r� �re�sub)�m�qrj r� Z sanitizedr# r# r$ �replacer� s z$Agent.hide_symbols.<locals>.replacerzH(?P<quote>['"])(?P<in_quote>.*?)((?<![\\|/])(?P=quote)|(?P<trunc>\.{3})))r� �compiler� )r� r� Zpatternr# r# r$ r� � s zAgent.hide_symbols)FF)N)N)'r r r! �__doc__rY r` ra rb rA rJ r9 rE r< r; r= rm rf r rq r6 rx r[ r� � staticmethodr� r �boolr� r rN rc r� r� �intr rQ r� r� r� r# r# r# r$ r% 0 s: $ & ) r% )-r� rr r/ rF r� r+ r� rM Z threadingr r r Ztypingr r r r Z dataclassesr rU r Zinternal.constantsr r Zinternal.exceptionsr r Zinternal.fault_detectorr Zinternal.local_countersr r r Zinternal.typesr Zinternal.user_plugin_utilsr Zinternal.utilsr r r r r r r% r# r# r# r$ �<module> s(
geri dön