관리-도구
편집 파일: accumulate.cpython-311.pyc
� IMg � � � d dl Z d dlZd dlZd dlZd dlmZ d dlmZ d dlm Z m Z d dl m Z m Z mZ d dlmZ ee� � Z G d� de e � � ZdS ) � N)� getLogger)� inactivity)� MessageType� Splittable)� MessageSink� MessageSource�expect)�recurring_checkc � � � e Zd Zej j Z eej � dd� � � � Z eej � dd� � � � Z e e f� fd� Z d� Zd� Zd� Zd � Z eej � � d � � � Zd � Z� xZS ) � Accumulate�%IMUNIFY360_AGGREGATE_MESSAGES_TIMEOUT�< �*IMUNIFY360_AGGREGATE_SHUTDOWN_SEND_TIMEOUT�2 c � �� t � � j di |�� || _ || _ t j t � � | _ d S )N� )�super�__init__�_period�_shutdown_timeout� collections� defaultdict�list�_data)�self�period�shutdown_timeout�kwargs� __class__s ��W/opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/plugins/accumulate.pyr zAccumulate.__init__ sG �� � �����"�"�6�"�"�"� �� �!1�� � �,�T�2�2�� � � � c � �� K � || _ || _ | j dk rd n<|� t | j � � | j � � � � � � | _ d S )Nr )�_loop�_sinkr � create_taskr �_flush�_task)r �loop�sinks r � create_sourcezAccumulate.create_source' sf � � � ��� ��� � �|�q� � � �D��!�!�"L�"?�/�$�,�"?�"?�� �"L�"L�"N�"N�O�O� � � � r! c � � K � || _ d S )N)r# )r r( s r � create_sinkzAccumulate.create_sink0 s � � � ��� � � r! c � � K � t j | � � � | j � � � d {V �� d S # t j $ r� t � d| j � � | j �a| j � � � t j t j � � 5 | j � d {V �� d d d � � Y d S # 1 swxY w Y Y d S Y d S w xY w)Nz5Timeout (%ss) sending messages to server on shutdown.) �asyncio�wait_for�stopr � TimeoutError�logger�errorr' �cancel� contextlib�suppress�CancelledError�r s r �shutdownzAccumulate.shutdown3 sC � � � � %��"�4�9�9�;�;��0F�G�G� G� G� G� G� G� G� G� G� G���#� %� %� %� �L�L�G��&� � � � �z�%�� �!�!�#�#�#� �(��)?�@�@� %� %� �*�$�$�$�$�$�$�$�%� %� %� %� %� %� %� %� %� %� %� %� %���� %� %� %� %� %� %� %� &�%�%� %���s4 �28 �A-C�%C�3 C�C �C�C � C�Cc � �l K � t � d� � | j �\| j � � � t j t j � � 5 | j � d {V �� d d d � � n # 1 swxY w Y t � d� � | � � � � d {V �� d S )Nz Accumulate.stop cancel _taskzAccumulate.stop wait lock) r2 �infor' r4 r5 r6 r. r7 r& r8 s r r0 zAccumulate.stopA s� � � � �� � �2�3�3�3� �:� !� �J� � � � � ��$�W�%;�<�<� !� !��j� � � � � � � � !� !� !� !� !� !� !� !� !� !� !���� !� !� !� !�� � �/�0�0�0��k�k�m�m���������s �A5�5A9�<A9c � �D K � t |j t � � r|j n|j f}|� � � r`t j � d� � 5 |D ]"}| j | � |� � �# d d d � � d S # 1 swxY w Y d S d S )N� accumulate) � isinstance� LIST_CLASS�tuple� do_accumulater �track�taskr �append)r �message� list_types� list_types r �collectzAccumulate.collectK s � � � � �'�,�e�4�4� '�G� � ��$�&� � � � � "� "� :��!�&�&�|�4�4� :� :�!+� :� :�I��J�y�)�0�0��9�9�9�9�:� :� :� :� :� :� :� :� :� :� :� :� :���� :� :� :� :� :� :� :� :s � &B�B�Bc � � K � | j }t j t � � | _ |� � � D ]�\ }}t |t � � r|� |� � n|f}|D ]�}t � d|j � dt |� � � d�� � | j � ||�� � � � � d {V �� �a# t $ r t � d||� � � w xY w��d S )NzPrepare z(<items=z>) for further processing)�itemsz%s, %s)r r r r rJ � issubclassr �batchedr2 r; �__name__�lenr$ �process_message� TypeErrorr3 )r � copy_datarG �messagesrL �batchs r r& zAccumulate._flushW sB � � � ��J� � �,�T�2�2�� �#,�?�?�#4�#4� � � �I�x� �i��4�4�!� �!�!�(�+�+�+� �[� � !� � ��� � �-�y� 1� -� -�3�u�:�:� -� -� -�� � �� �*�4�4�Y�Y�U�5K�5K�5K�L�L�L�L�L�L�L�L�L�L�� � � � ��L�L��9�e�<�<�<������ � � s �&*C�(C9)rM � __module__� __qualname__r �ProcessingOrder�POST_PROCESS_MESSAGE�PROCESSING_ORDER�int�os�environ�get�DEFAULT_AGGREGATE_TIMEOUT�SHUTDOWN_SEND_TIMEOUTr r* r, r9 r0 r r � AccumulatablerH r& � __classcell__)r s @r r r s �� � � � � �"�2�G�� #�� � ���>��C�C�!� !� � �C� � ���C�R�H�H� � �� )�.� 3� 3� 3� 3� 3� 3� � � �� � � %� %� %� � � � �V�K� %�&�&� :� :� '�&� :�� � � � � � r! r )r. r r5 rZ �loggingr �defence360agent.apir �"defence360agent.contracts.messagesr r �!defence360agent.contracts.pluginsr r r �defence360agent.utilsr rM r2 r r r! r �<module>rf s �� ���� � � � � � � � � � � � � � � � � � � *� *� *� *� *� *� F� F� F� F� F� F� F� F�� � � � � � � � � � 2� 1� 1� 1� 1� 1� ��8� � ��[� [� [� [� [��m� [� [� [� [� [r!