File manager - Edit - /opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/__pycache__/zeromq.cpython-310.pyc
Back
o ;j�� � @ s� d Z ddlZddlZddlZddlZddlZddlZddlZddlZddl Z ddl m Z ddl Z ddlZ ddlZ ddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddlZddl Zddl m Z ddl m Z m!Z! ddl m"Z"m#Z#m Z zddl$Z dZ%W n e&y� dZ%Y nw e�'e(�Z)d Z*e+� Z,dd d �Z-G d d � d ej.j/j0�Z0G dd� dej.j/j1�Z2dd� Z3G dd� d�Z4G dd� d�Z5G dd� dej.j/j6�Z7G dd� dej.j/j8�Z8dS )z Zeromq transport classes � N)�randint)� ipaddress)� SaltException�SaltReqTimeoutError)�LIBZMQ_VERSION_INFO�ZMQ_VERSION_INFO�zmqTF�< c C s� ddl m} dj|| �|d�}|s|rntdkr_tdkr_|r.|r.dj||�||| �|d�}|S |r@|s@d j||�|| �|d �}|S |r]|s]t�| �jd krNd n|d �}dj|||| �|d�}|S t� d� t� d� t� d� |S )aR Return the ZeroMQ URI to connect the Minion to the Master. It supports different source IP / port, given the ZeroMQ syntax: // Connecting using a IP address and bind to an IP address rc = zmq_connect(socket, "tcp://192.168.1.17:5555;192.168.1.1:5555"); assert (rc == 0); Source: http://api.zeromq.org/4-1:zmq-tcp r )� ip_bracketz tcp://{master_ip}:{master_port})� master_ip� master_port)� � � )� r r z9tcp://{source_ip}:{source_port};{master_ip}:{master_port})� source_ip� source_portr r z-tcp://{source_ip}:0;{master_ip}:{master_port})r r r r z0.0.0.0z::z6tcp://{ip_any}:{source_port};{master_ip}:{master_port})�ip_anyr r r zAUnable to connect to the Master using a specific source IP / portz9Consider upgrading to pyzmq >= 16.0.1 and libzmq >= 4.1.6zVSpecific source IP / port for connecting to master returner port: configuraion ignored) Zsalt.utils.networkr �formatr r r � ip_address�version�log�warning)r r r r r � master_urir � r �I/opt/saltstack/salt/lib/python3.10/site-packages/salt/transport/zeromq.py�_get_master_uri4 sV ���&�� ������� �r c s� e Zd ZdZdZ� fdd�Zdd� Zdd� Zd d � Ze j j j j dd d ��Zedd� �Ze j j j j dd� �Zedd� �Zdd� Ze j j j j dd� �Z� ZS )� PublishClientzw A transport channel backed by ZeroMQ for a Salt Publisher to use to publish commands to connected minions �zeromqc sv t � j||fi |�� || _|| _t�tjj� | jd ��� � | _ d| _ t �� | _| j�t j�| _| jd r^| j�t jd� | j�d�dkrO| j�t jd� n| j�t jtjj� | j �� n| j�t jd� | j�t jtjj� | jd �� tt d �r�| j�t j| jd � | j�t j| jd � | j�t j| jd � | j�t j| jd � | jd }| jd r�t | jd | jd | jd �}t � d| jd | jd | jd |� t � d|� | j�t j |� tt d��rt � d| jd | jd � | j�t j | jd � | jd du �sd| jd v �r tt d��r | j�t j!d� t"�r7| jd �r9t#| j�| _$| j$�%| j� d S d S d S ) N�idF� zmq_filtering� broadcast�__role�syndic� syndic� � TCP_KEEPALIVE� tcp_keepalive�tcp_keepalive_idle�tcp_keepalive_cnt�tcp_keepalive_intvlZ recon_defaultZrecon_randomizeZ recon_maxz?Generated random reconnect delay between '%sms' and '%sms' (%s)z#Setting zmq_reconnect_ivl to '%sms'�RECONNECT_IVL_MAXz'Setting zmq_reconnect_ivl_max to '%sms'�ipv6T�:r �IPV4ONLYr � zmq_monitor)&�super�__init__�opts�io_loop�hashlib�sha1�salt�utils� stringutils�to_bytes� hexdigest�hexid�_closingr �Context�context�socketZSUB�_socket� setsockoptZ SUBSCRIBE�get�IDENTITY�hasattrr& �TCP_KEEPALIVE_IDLE�TCP_KEEPALIVE_CNT�TCP_KEEPALIVE_INTVLr r �debugZ RECONNECT_IVLr+ r. �HAS_ZMQ_MONITOR�ZeroMQSocketMonitor�_monitor� start_io_loop)�selfr2 r3 �kwargsZ recon_delay�� __class__r r r1 x sz �� �� � � � �� �"� �zPublishClient.__init__c C s� | j du rd S d| _ t| d�r | jd ur | j�� d | _t| d�r(| j�d� n t| d�r3| j�d� t| d�rE| jjdu rG| j� � d S d S d S )NTrK �_streamr r@ r> F) r< rD rK �stoprQ �closer@ r> �closed�term�rM r r r rS � s �zPublishClient.closec C � | S �Nr rV r r r � __enter__� � zPublishClient.__enter__c C � | � � d S rX �rS �rM �exc_typeZexc_valZexc_tbr r r �__exit__� � zPublishClient.__exit__Nc C sP d| _ || _t�d| j� t�d| | j� | j�| j� |d ur&|d� d S d S )NTzCConnecting the Minion to the Master publish port, using the URI: %sz%r connecting to %s)�_connect_called� publish_portr rH � master_pubr@ �connect)rM rb Zconnect_callbackZdisconnect_callbackr r r rd � s � �zPublishClient.connectc C s( t | jd | j| j�d�| j�d�d�S )z0 Return the master publish port r r Zsource_publish_port�r r )r r2 rb rB rV r r r rc � s �zPublishClient.master_pubc C s� t |�}|dkrtj�|d �}nH|dkrPtjj�|d �}| j�d�dkr-|d| j fvs9| j�d�dkrG|dvrGt � d|� tj j j�d ��tj�|d �}n td �t |����tj j j�|��) z� Take the zmq messages, decrypt/decode them into a payload :param list messages: A list of messages to be decoded r r � r" r# � broadcast)rg r# z(Publish received for not this minion: %sNz@Invalid number of messages ({}) in zeromq pubmessage from master)�lenr6 �payload�loadsr7 r8 Zto_strr2 rB r; r rH �ext�tornado�gen�Return� Exceptionr )rM �messagesZ messages_lenri Zmessage_targetr r r �_decode_messages� s$ ��z PublishClient._decode_messagesc C s( t | d�stjjj| j| jd�| _| jS )zI Return the current zmqstream, creating one if necessary rQ �r3 )rD r � eventloop� zmqstream� ZMQStreamr@ r3 rQ rV r r r �stream s �zPublishClient.streamc s� � du r&z| j �d�W S ty% } zt|�dkr W Y d}~dS � d}~ww | j�� �fdd�}z| j �|�W S tyQ } zt|�dkrLW Y d}~dS � d}~ww )z� Register a callback for received messages (that we didn't initiate) :param func callback: A function which should be called when data is received NzStream is closedc s �j � g| �R i |�� d S rX ��spawn_callback)�argsrN ��callbackr3 r r � _dispatch9 s z(PublishClient.on_recv.<locals>._dispatch)rv �on_recv�OSError�strr3 )rM r{ �excr| r rz r r} s& �� ��zPublishClient.on_recvc C s | j j|dd� d S )NT)Znoblock)rv �send�rM �msgr r r r� C s zPublishClient.send�NN)�__name__� __module__� __qualname__�__doc__�ttyper1 rS rY r_ r6 rk rl rm � coroutinerd �propertyrc rq rv r} r� � __classcell__r r rO r r p s$ K ! $r c @ sp e Zd Zdd� Zdd� Zdd� Zdd� Zd d � Zd d � Ze j j j j d d� �Zdd� Zdd� Zdd� Zdd� ZdS )� RequestServerc C s || _ d| _d | _d | _d S )NF)r2 r< rK � _w_monitor�rM r2 r r r r1 I s zRequestServer.__init__c C s� | � � t�d�}djd i | j��| _|�tj�| _| j� tj d� t td�r/| j� tj d� | j� tj d� | j� tjd� | j� tjd� | j� tjd� | jd d u rct td �rc| j� tjd � | j� tj| j�d d�� | �� |�tj�| _| j� tj d� | jd r�tjj�� s�t� d| jd � t � | jd � | j�dd�dkr�d�| j�dd��| _ nd�t j �!| jd d��| _ t� d� t� d| j� | j�"| j� t� d| j � | j�"| j � | j�dd�dkr�t �#t j �!| jd d�d� | jj$�s| jj$�rn5z t�%tj&| j| j� W n' tj'�y. } z|j(t(j)k�r)W Y d }~q�� d }~w t*t+f�y9 Y nw q�|�,� d S ) zA Multiprocessing target for the zmq queue device r z tcp://{interface}:{ret_port}�� �ROUTER_HANDOVERr � � r, Tr. r � zmq_backlogZmworker_queue_nicenessz$setting mworker_queue niceness to %d�ipc_mode� �tcp�tcp://127.0.0.1:{}�tcp_master_workers� �ipc://{}�sock_dir� workers.ipcz*Setting up the master communication serverzReqServer clients %szReqServer workers %s� Nr )-� _RequestServer__setup_signalsr r= r r2 Zurir? ZROUTER�clientsrA �LINGERrD r� r& rE rG rF r. �BACKLOGrB �_start_zmq_monitorZDEALER�workersr6 r7 �platformZ is_windowsr �info�os�nice�w_uri�path�join�bind�chmodrT ZdeviceZQUEUE�ZMQError�errno�EINTR�KeyboardInterrupt� SystemExitrU )rM r> r� r r r � zmq_deviceO sh � �� ��� zRequestServer.zmq_devicec C s | j rdS t�dt�� � d| _ t| dd�dur | j�� d| _t| dd�dur0| j�� d| _t | d�r@| j j du r@| j � � t | d�rP| j j du rP| j � � t | d �rZ| j� � t | d �rj| jj du rj| j� � t | d �r|| jj du r~| j�� dS dS dS ) z4 Cleanly shutdown the router socket Nz$MWorkerQueue under PID %s is closingTrK r� r� Fr� rv r@ r> )r< r r� r� �getpid�getattrrK rR r� rD r� rT rS r� rv r@ r> rU rV r r r rS � s* �zRequestServer.closec C s |j | jdd� dS )z� Pre-fork we need to create the zmq router device :param func process_manager: An instance of salt.utils.process.ProcessManager Z MWorkerQueue)�nameN)� add_processr� �rM Zprocess_managerr r r �pre_fork� s zRequestServer.pre_forkc C sN t r#| jd r%t�d� t| j�| _tj| jj d�� � t�d� dS dS dS )zM Starts ZMQ monitor for debugging purposes. :return: r/ zStarting ZMQ monitor)�targetz$ZMQ monitor has been started startedN) rI r2 r rH rJ r@ r� � threading�Thread� start_poll�startrV r r r r� � s �z RequestServer._start_zmq_monitorc s t �d�}|�t j��_�j�t jd� ��� �j� dd�dkr-d� �j� dd���_ nd � t j ��jd d ���_ t�d �j � �j��j � �j� dd�dkrnt j �t j ��jd d ��rnt �t j ��jd d �d � t jjj�j� d��_|�_� �fdd�}�j�|� dS )ad After forking we need to create all of the local sockets to listen to the router :param func message_handler: A function to called to handle incoming payloads as they are picked up off the wire :param IOLoop io_loop: An instance of a Tornado IOLoop, to handle event scheduling r ���r� r� r� r� r� r� r� r� r� zWorker binding to socket %sr� rr c s � � �j| |� d S rX )rx �handle_message)rv ri �r3 rM r r �_dispatch_handle_message� s z9RequestServer.post_fork.<locals>._dispatch_handle_messageN)r r= r? ZREPr@ rA r� r� r2 rB r r� r� r� r� r r� rd �isfiler� rs rt ru rv �message_handlerZon_recv_stream)rM r� r3 r> r� r r� r � post_fork� s* ��� zRequestServer.post_forkc c sb � z| � |�}W n tjjy | j�| �ddi�� Y d S w | �|�V }| j�| �|�� d S )Nr� zbad load)�decode_payloadr6 � exceptionsZSaltDeserializationErrorrv r� �encode_payloadr� )rM rv ri Zreplyr r r r� � s �� z RequestServer.handle_messagec C s t j�|�S rX )r6 ri �dumps�rM ri r r r r� r` z RequestServer.encode_payloadc C s$ t � t j| j� t � t j| j� d S rX )�signal�SIGINT�_handle_signals�SIGTERMrV r r r Z__setup_signals s z RequestServer.__setup_signalsc C sb | j j� d�}|tjkr|d7 }n |tjkr|d7 }|d7 }t�|� | �� t� t j j j � d S )Nz received a r� r� z . Exiting)rP r� r� r� r� r rH rS �sys�exitr6 �defaultsZ exitcodes�EX_OK)rM �signumZsigframer� r r r r� s z RequestServer._handle_signalsc C s t j�|d �}|S )Nr )r6 ri rj r� r r r r� s z RequestServer.decode_payloadN)r� r� r� r1 r� rS r� r� r� r6 rk rl rm r� r� r� r� r� r� r r r r r� H s Q , r� c C s� t td�r=|r?d|v r| �tj|d � d|v r!| �tj|d � d|v r.| �tj|d � d|v rA| �tj|d � dS dS dS dS )a� Ensure that TCP keepalives are set as specified in "opts". Warning: Failure to set TCP keepalives on the salt-master can result in not detecting the loss of a minion when the connection is lost or when its host has been terminated without first closing the socket. Salt's Presence System depends on this connection status to know if a minion is "present". Warning: Failure to set TCP keepalives on minions can result in frequent or unexpected disconnects! r& r' r( r) r* N)rD r rA r& rE rF rG )Z zmq_socketr2 r r r �_set_tcp_keepalive s �r� c @ s� e Zd ZdZd dd�Zdd� Zdd � Zd d � Zd d � Zdd� Z dd� Z dd� Z e j jjjdd� �Ze j jjjd dd��Zdd� Ze j jjje j jjjfdd��ZdS ) �AsyncReqMessageClienta� This class wraps the underlying zeromq REQ socket and gives a future-based interface to sending and recieving messages. This works around the primary limitation of serialized send/recv on the underlying socket by queueing the message sends in this class. In the future if we decide to attempt to multiplex we can manage a pool of REQ/REP sockets-- but for now we'll just do them in serial r Nc C sh || _ || _|| _|du rtjjjj�� | _ n|| _ t j j � � | _d| _d| _tjjj�� | _d| _dS )a� Create an asynchronous message client :param dict opts: The salt opts dictionary :param str addr: The interface IP address to bind to :param int linger: The number of seconds to linger on a ZMQ socket. See http://api.zeromq.org/2-1:zmq-setsockopt [ZMQ_LINGER] :param IOLoop io_loop: A Tornado IOLoop event scheduler [tornado.ioloop.IOLoop] NF)r2 �addr�lingerr6 rk rl �ioloop�IOLoop�currentr3 r rs �futurer= r> r? �_closedZqueues�Queue�_queue�_send_recv_exit_future)rM r2 r� r� r3 r r r r1 ? s z AsyncReqMessageClient.__init__c C s8 | j d u r tjj�� | _ t| d�r| jrd S | �� d S )Nr? )r> r rs r� r= rD r? � _init_socketrV r r r rd V s z AsyncReqMessageClient.connectc C s$ | j �tj�| _ttd�r| j�tjd� | j�d�sO| j�d�p"d}zt � � }W n t y4 d}Y nw dj |t� � |t �� d d �}| j�tj|�d �� t| j| j� | j�d �rwttd �rj| j�tjd � n ttd�rw| j�tjd� | j�tj| j� | j�| j� | j�| j| j� d S )Nr+ i� r" r Zclirr z#salt-req/{role}/{host}/{uid}/{slot}� )�role�host�uidZslotzutf-8ztcp://[�IPV6r r. ) r> r? r ZREQrD rA r+ r2 rB r� �getuid�AttributeErrorr � gethostnamer� rC �encoder� r� � startswithr� r. r� r� rd r3 rx � _send_recv)rM r� r� �identityr r r r� _ s4 � � z"AsyncReqMessageClient._init_socketc C sX t | d�r| jdur| j�d� d| _| jdur(| jjdu r*| j�� d| _dS dS dS )zHClose socket and ZMQ context only (no IOLoop / _send_recv coordination).r? Nr F)rD r? rS r> rT rU rV r r r �_close_zmq_only� s �z%AsyncReqMessageClient._close_zmq_onlyc C s | � � | jS )uW Return a ``Future`` that completes after ZMQ resources are released. Coroutine callers on the owning I/O loop thread should ``yield`` this future (via ``salt.ext.tornado.gen.convert_yielded``) before allocating another client on that loop—``close()`` schedules teardown asynchronously in that case. )�_initiate_async_req_close�_close_completed_futurerV r r r � close_future� s z"AsyncReqMessageClient.close_futurec C s$ | � � }|dur|jdd� dS dS )u2 Stop the send/recv coroutine and close ZMQ resources. Safe to call more than once. When no I/O loop iteration is active, uses ``run_sync`` around the graceful shutdown coroutine (LocalClient/SyncWrapper and similar). When the owning ``io_loop`` is already running—including on the master event loop thread while short-lived REQ clients are torn down—you cannot call ``run_sync``. In that case the shutdown future is queued with ``add_callback`` / ``add_future``. If ``close()`` is invoked from another thread while the loop is running, we block until shutdown completes. On the owning I/O loop thread, prefer ``yield``-ing ``close_future()`` instead of ``close()`` when you recreate clients immediately afterward. N� ��timeout)r� �wait)rM �cross_thread_evtr r r rS � s �zAsyncReqMessageClient.closec C s2 t | dd �}|d ur|�� s|�d � d S d S d S )Nr� )r� �done� set_result�rM Zfutr r r �_mark_teardown_finished� s �z-AsyncReqMessageClient._mark_teardown_finishedc s t �dd�dur dS tjjj�� �_d� � �fdd���fdd�}�jr)�� dS d�_�jdu r6�� dS tjjj�� �_ zt �j dd �sT�j j |d d � �� W dS W nT t y� } z d t |��� vrwtjd |dd� �� W Y d}~dS W Y d}~n-d}~w tjjjjy� t�d� �� Y dS ty� tjddd� �� Y dS w z tjjj���� ��W n ty� tjddd� �� Y dS w t �j dd�}|t�� k}|r�dnt�� � �fdd�����fdd�}z �j �|� W � S t�y tjddd� �� Y dS w )z� Start teardown once. Sets ``self._close_completed_future`` and returns ``None``, or an ``threading.Event`` that unblocks after ``finalize()`` when the caller needs to wait from a non-I/O-loop thread. r� Nc s. d �_ ��� ��� � d ur� �� d S d S rX )r� r� r� �setr )r� rM r r �finalize� s �zAAsyncReqMessageClient._initiate_async_req_close.<locals>.finalizec s � � � S rX )�_graceful_shutdown_coror rV r r � run_shutdown� s zEAsyncReqMessageClient._initiate_async_req_close.<locals>.run_shutdownTZ_runningFr� r� zalready runningz)REQ client shutdown: run_sync aborted: %s��exc_infoz>Graceful REQ message client shutdown timed out during run_syncz;Graceful REQ message client shutdown failed during run_syncz?Could not schedule REQ shutdown coroutine while loop is runningZ _thread_identc sP z#z| � � W n ty tjddd� Y nw W � � d S W � � d S � � w )NzAGraceful REQ message client shutdown failed during async teardownTr� )�resultro r rH )r� )r� r r �on_done s ��� �z@AsyncReqMessageClient._initiate_async_req_close.<locals>.on_donec s �j ��� � d S rX )r3 Z add_futurer )r rM �shutdown_futurer r �schedule s zAAsyncReqMessageClient._initiate_async_req_close.<locals>.schedulez?Graceful REQ message client shutdown failed scheduling callback)r� r6 rk rl � concurrent�Futurer� r� r? r� r3 Zrun_sync� RuntimeErrorr �lowerr rH r� � TimeoutErrorro rm Zconvert_yieldedr� r� � get_ident�Event� add_callback)rM r� r� Z ioloop_threadZ same_threadr r )r� r� r rM r r r� � s� �� �� �� � �� ���z/AsyncReqMessageClient._initiate_async_req_closec c st � z| j �tjjj�� tf� W n# ty3 t j ddd� | j r,| j � � s,| j � d � tjjj�� �w | j V d S )Nz!Could not queue shutdown sentinelTr� )r� � put_nowaitr6 rk rl r r �_REQ_QUEUE_SHUTDOWNro r rH r� r� r� rm rn rV r r r r� , s �� � z-AsyncReqMessageClient._graceful_shutdown_coroc # s� � t jjj�� }t j�|�}�j�||f� d}� dur(� �fdd�}|� |� �j � d�du r2d}|dur?�j � |�j|�}z|V }W |durN�j �|� n |durZ�j �|� w w t jjj�|��)zY Return a future which will be completed when the message has a response Nc s | � � }�j�� |� d S rX )r r3 r )r� �response�r{ rM r r � handle_futureH s z1AsyncReqMessageClient.send.<locals>.handle_futureZ detect_modeTr )r6 rk rl r r ri r� r� r Zadd_done_callbackr2 rB r3 Z call_later�_timeout_messageZremove_timeoutrm rn )rM �messager� r{ r� Z send_timeoutr �recvr r r r� 9 s. � � ���zAsyncReqMessageClient.sendc C s |� � s |�td�� d S d S )NzMessage timed out)r� � set_exceptionr )rM r� r r r r ^ s �z&AsyncReqMessageClient._timeout_messagec c s� � d}�z2|�r#z| j jtjdd�d�V \}}W nF |y^ z |�dtj�}|�� W n/ |y5 Y n' tjj j yI t � d� d}Y Y �n� tj y[ t � d� d}Y Y �n�w Y qw |tu rgd}�n�z|�|�V W n{ tjj j y� } zt � d � d}|�� s�|�|� W Y d }~�n�d }~w tj y� } zI|jtjtjtjjfv r�t � d � d}|�� s�|�|� n&|jtjkr�t �d � d}|�� s�|�|� nt �d |� |�� s�|�|� W Y d }~nd }~ww |�� �r|�� }t|t�r�t � d� nt � d� | jd u}| � � |�r| � � d}�n d} d} |�� �r#n�z |�dtj �V } W nR tjj j �y[ } z t j ddd� t �d� d}|�� �sQ|� d � W Y d }~n)d }~w tj �y } zt � d� d}|�� �su|�|� W Y d }~nd }~ww | �r�z |� � V } d} W nK tjj j �y� } zt � d� d}|�� �s�|�|� W Y d }~n)d }~w tj �y� } zt � d� d}|�� �s�|�|� W Y d }~nd }~ww n|�� �r�n�q |�� �r|�� }t|t��r�t � d� nt � d� | jd u}| � � |�r | � � d}n| �r!t!j"�#| �} |�� �s!|� | � |sW | j$d u�r6| j$�� �s6| j$� d � n| j$d u�rJ| j$�� �sK| j$� d � w w w t � d|� d S )at Long-running send/receive coroutine. This should be started once for each socket created. Once started, the coroutine will run until the socket is closed. A future and message are pulled from the queue. The message is sent and the reply socket is polled for a response while checking the future to see if it was timed out. Ti, )Z millisecondsr� r z&Loop closed while polling send socket.Fz!Send socket closed while polling.zLoop closed while sending.Nz!Send socket closed while sending.z"Socket was found in invalid state.z/Unhandled Zeromq error durring send/receive: %sz.Request timed out while sending. reconnecting.z<The request ended with an error while sending. reconnecting.z)Loop closed while polling receive socket.r� z-Master is unavailable (Connection Cancelled).z$Receive socket closed while polling.z Loop closed while receiving.z&Receive socket closed while receiving.z=Request timed out while waiting for a response. reconnecting.z.The request ended with an error. reconnecting.z$Send and receive coroutine ending %s)%r� rB �datetime� timedelta�pollr �POLLOUTr rs r� ZCancelledErrorr �tracer� r r� r� r r� �ENOTSOCKZETERM�errorr� ZEFSM� exception� isinstancer r> r� rd �POLLINr� r r6 ri rj r� ) rM r? Z _TimeoutErrorZsend_recv_runningr� r Z poll_futurer� Z_should_reconnectZreceived�readyr �datar r r r� b s* � � �� �� � �� �� � � � �� � �� � & � ��� � � ���z AsyncReqMessageClient._send_recv)r Nr� )r� r� r� r� r1 rd r� r� r� rS r� r� r6 rk rl rm r� r� r� r r r� r r r r r� 6 s" 4 k $ r� c @ sD e Zd ZdZdd� Zdd� Zdd� Zedd � �Zd d � Z d d � Z dS )rJ Nc C s || _ | j �� | _d| _dS )z� Create ZMQ monitor sockets More information: http://api.zeromq.org/4-0:zmq-socket-monitor N)r@ Zget_monitor_socket�_monitor_socket�_monitor_stream)rM r? r r r r1 s z ZeroMQSocketMonitor.__init__c C s2 t �d� tjjj| j|d�| _| j�| j � d S )N�Event monitor start!rr ) r r r rs rt ru r! r" r} �monitor_callback)rM r3 r r r rL s �z!ZeroMQSocketMonitor.start_io_loopc C s| t �d� z*| jd ur'| j�� r-| j�� }| �|� | jd ur*| j�� sW d S W d S W d S W d S ttjj fy= Y d S w )Nr# ) r r r! r Zrecv_multipartr$ r� r r ZContextTerminatedr� r r r r� s ,��z ZeroMQSocketMonitor.start_pollc C sD t jd u r i }tt�D ]}|�d�rtt|�}|||< q |t _t jS )NZEVENT_)rJ � _ZeroMQSocketMonitor__EVENT_MAP�dirr r� r� )rM � event_mapr� �valuer r r r' # s �z ZeroMQSocketMonitor.event_mapc C sJ t jj�|�}| j|d |d< t�d|� |d t jkr#| �� d S d S )N�event� descriptionzZeroMQ event: %s) r r7 �monitorZparse_monitor_messager' r rH ZEVENT_MONITOR_STOPPEDrR )rM r� Zevtr r r r$ . s �z$ZeroMQSocketMonitor.monitor_callbackc C sz | j d u rd S z| j �� W n tjy Y nw d | _ | jd ur)| j�� d | _| jd ur6| j�� d | _t�d� d S )NzEvent monitor done!) r@ Zdisable_monitorr �Errorr" rS r! r r rV r r r rR 5 s � zZeroMQSocketMonitor.stop) r� r� r� r% r1 rL r� r� r' r$ rR r r r r rJ s rJ c @ s� e Zd ZdZe�� Zdd� Zdd� Z d!dd�Z e d d � �Z e d d � �Z e jjjjd"d d��Zdd� Ze dd� �Zdd� Zdd� Zdd� Ze dd� �Zdd � Zd d � Zd d � ZdS )#� PublishServerzD Encapsulate synchronous operations for a publisher channel c C s || _ d S rX �r2 r� r r r r1 M s zPublishServer.__init__c C s t jjj�d�S )N� )r6 rk rl rm �sleeprV r r r rd P s zPublishServer.connectNc sH t jjj�� � � �� � | _t�d�}|� tj �}t |�}|� � � t || j� tjj�|� | _}z|�tj| j�dd�� W n% ttjjfye |�tj| j�dd�� |�tj| j�dd�� Y nw | jd du ryttd�ry|�tj d� |�tj | j�dd�� |�tj d � |� tj �}tjj�|�}|�tj d � t j j!�"| j#� t$�%d | j&� |�'| j&� t$�%d | j#� t j j(�)d �� |�'| j#� W d � n1 s�w Y t jjj*j+�fdd���� �fdd�}|�,|� z)z� �-� W n t.t/f�y Y n w W |�0� |�0� d S W |�0� |�0� d S |�0� |�0� w )z� This method represents the Publish Daemon process. It is intended to be run in a thread or process as it creates and runs its own ioloop. r Zpub_hwmr� r, Tr. r r� r� z!Starting the Salt Publisher on %sz Starting the Salt Puller on %s� Nc 3 sb � z| D ] }t j�|�}� |�V qW d S ty0 } ztjd|tjd� W Y d }~d S d }~ww )Nz Un-handled error in publisher %s)Zexc_info_on_loglevel)r6 ri rj ro r r �logging�DEBUG)�packages�packageri r� )�publish_payloadr r r} � s � ����z-PublishServer.publish_daemon.<locals>.on_recvc s � � �| � d S rX rw )r4 )r� r} r r �_dispatch_on_recv� s z7PublishServer.publish_daemon.<locals>._dispatch_on_recv)1r6 rk rl r� r� Z make_currentr3 r r= r? ZPUBrJ rL r� r2 rs rt ru � dpub_sockrA ZHWMrB r� r r� ZSNDHWMZRCVHWMrD r. r� r� ZPULLr7 r Zcheck_ipc_path_max_len�pull_urir r� �pub_urir� �filesZ set_umaskrm r� r} r� r� r� rS ) rM r6 Zpresence_callbackZremove_presence_callbackr> �pub_sockr+ Z pull_sockr7 r )r� r} r6 r �publish_daemonS s\ � � �� � � z PublishServer.publish_daemonc C sH | j �dd�dkrd�| j �dd��}|S d�tj�| j d d ��}|S ) Nr� r� r� r� �tcp_master_publish_pull� r� r� �publish_pull.ipc)r2 rB r r� r� r� )rM r9 r r r r9 � s ���zPublishServer.pull_uric C s dj di | j��S )Nz tcp://{interface}:{publish_port}r )r r2 rV r r r r: � s zPublishServer.pub_uric c s � t j�|�}| jd rm|rV|D ](}t�d| j� t jj� t � t jj� |��� � �}| j �||g�V t�d� q| j�d�rTt�d� | j �d|g�V t�d� d S d S t�d| j� | j �d |g�V t�d � d S t�d | j� | j �|�V t�d � d S ) Nr z'Sending filtered data over publisher %szFiltered data has been sentZ order_mastersz Sending filtered data to syndicr$ z%Filtered data has been sent to syndicz*Sending broadcasted data over publisher %sr! z Broadcasted data has been sentz-Sending ZMQ-unfiltered data over publisher %sz Unfiltered data has been sent)r6 ri r� r2 r r r: r7 r8 r9 r4 r5 r: r8 Zsend_multipartrB r� )rM ri Z topic_listZtopicZhtopicr r r r6 � s, � � �z PublishServer.publish_payloadc C s |j | j| jfd� dS )a. Do anything necessary pre-fork. Since this is on the master side this will primarily be used to create IPC channels and create our daemon process to do the actual publishing :param func process_manager: A ProcessManager, from salt.utils.process.ProcessManager )ry N)r� r= r6 r� r r r r� � s �zPublishServer.pre_forkc C s z| j jW S ty Y dS w )z� This thread's zmq publisher socket. This socket is stored on the class so that multiple instantiations in the same thread will re-use a single zmq socket. N)� _sock_data�sockr� rV r r r r< � s �zPublishServer.pub_sockc C s� | j r| �� t�� }|�tj�| j_| j �tj d� | j � dd�dkr/d� | j � dd��}n d� t j�| j d d ��}t�d |� | j �|� | jjS ) z� Create and connect this thread's zmq socket. If a publisher socket already exists "pub_close" is called before creating and connecting a new socket. r� r� r� r� r� r> r? r� r� r@ z Connecting to pub server: %s)r< � pub_closer r= r? ZPUSHrA rB rA r� r2 rB r r� r� r� r rH rd )rM �ctxr9 r r r � pub_connect� s �� zPublishServer.pub_connectc C s, t | jd�r| jj�� t| jd� dS dS )zn Disconnect an existing publisher socket and remove it from the local thread's cache. rB N)rD rA rB rS �delattrrV r r r rC � s �zPublishServer.pub_closec K s4 | j s| �� tj�|�}| j �|� t�d� dS )z� Publish "load" to minions. This send the load to the publisher daemon process with does the actual sending to minions. :param dict load: A load to be sent across the wire to minions z Sent payload to publish daemon.N)r< rE r6 ri r� r� r rH )rM ri rN Z serializedr r r �publish s zPublishServer.publishc C s | j �dd�S )Nr F)r2 rB rV r r r � topic_support s zPublishServer.topic_supportc C r[ rX )rC rV r r r rS r` zPublishServer.closec C rW rX r rV r r r rY rZ zPublishServer.__enter__c C r[ rX r\ r] r r r r_ r` zPublishServer.__exit__r� rX ) r� r� r� r� r� �localrA r1 rd r= r� r9 r: r6 rk rl rm r� r6 r� r< rE rC rG rH rS rY r_ r r r r r- F s2 �J r- c sh e Zd ZdZ� fdd�Zejjjj dd� �Z ejjjj e fdd��Z dd � Z d d � Zed d � �Z� ZS )� RequestClientr c s@ t � �||� || _| �|�}t| j||d�| _d| _d| _d S )Nrr F)r0 r1 r2 �get_master_urir� �message_clientr< ra )rM r2 r3 r rO r r r1 # s � zRequestClient.__init__c C s d| _ | j�� d S �NT)ra rL rd rV r r r rd / s zRequestClient.connectc c s. � | � � V | jj||d�V }tjjj�|��)Nr� )rd rL r� r6 rk rl rm rn )rM �loadr� �retr r r r� 4 s � zRequestClient.sendc C s | j �� }d| _|S rM )rL r� r< r� r r r r� : s zRequestClient.close_futurec C s | j rd S d| _ | j�� d S rM )r< rL rS rV r r r rS ? s zRequestClient.closec C sD d| v r| d S d| v r t | d | d | �d�| �d�d�S td��)Nr r r r Zsource_ret_portre z5ReqChannel: missing master_uri/master_ip in self.opts)r rB r r. r r r rK E s �z RequestClient.get_master_uri)r� r� r� r� r1 r6 rk rl rm r� rd �REQUEST_TIMEOUTr� r� rS � staticmethodrK r� r r rO r rJ s rJ r� )9r� r r� r4 r2 r� r� r? r� r� �randomr Z zmq.errorr Zzmq.eventloop.futureZzmq.eventloop.zmqstreamZsalt.ext.tornador6 Zsalt.ext.tornado.concurrentZsalt.ext.tornado.genZsalt.ext.tornado.ioloopZsalt.ext.tornado.locksZsalt.ext.tornado.queuesZ salt.payloadZsalt.transport.baseZsalt.utils.filesZsalt.utils.processZsalt.utils.stringutilsZsalt.utils.zeromqZ salt._compatr Zsalt.exceptionsr r r r Zzmq.utils.monitorrI � ImportError� getLoggerr� r rP �objectr r Z transport�baser ZDaemonizedRequestServerr� r� r� rJ ZDaemonizedPublishServerr- rJ r r r r �<module> sh � < Y V PC Z
| ver. 1.4 |
Github
|
.
| PHP 8.2.30 | Generation time: 0 |
proxy
|
phpinfo
|
Settings