
    :jv*                     
   d Z ddlZddlZddlmZmZ ddlmZ ddlm	Z	m
Z
mZmZ ddlmZ ddlmZmZ ddlmZ dd	lmZ dd
lmZ ddlmZ ddlmZmZmZmZmZm Z m!Z!m"Z"  ed      Z# G d d      Z$ G d de      Z%dede%fdZ&y)z~
Monitoring system for multi-provider pipeline processing.
Pure data aggregation and caching without scheduling dependencies.
    N)OrderedDictdeque)MappingProxyType)CallableDictListOptional)MonitoringConfig)MONITORING_THRESHOLDSAlertConfig)ALERT_COOLDOWN_SECONDS)AlertKeyType)
get_logger   )
AlertLevel)AlertIMonitorProviderMonitoringSnapshotMonitoringSummaryPerformanceMetricsPipelineStatusProviderStatusSystemStatusstatec                   ~    e Zd ZdZdefdZdeegdf   ddfdZde	e
ef   d	eddfd
ZdeddfdZdeddfdZddZy)AlertManagerz4Manages alerts and notifications for pipeline issuesconfigc                    g | _         t        t        j                  dt        j
                              | _        t               | _        t        j                         | _        t        j                         | _        |j                  | _        |j                  | _        |j                   | _        t        d   | _        t$        j'                  d       y )Nmax_alert_historymaxlenmin_sample_sizezAlert manager initialized)alert_handlersr   r   getr   DEFAULT_HISTORY_SIZEalert_historyr   alert_ttl_index	threadingLocklocktimelast_cleanup_timeerror_thresholdqueue_thresholdmemory_thresholdr"   loggerinfoselfr   s     4/root/.openclaw/workspace/harvester/state/monitor.py__init__zAlertManager.__init__&   s    =?$)(,,-@+BbBbc%
 9DNN$	 "&  &55%55 & 7 745FG/0    handlerNreturnc                 :    | j                   j                  |       y)zAdd alert handler functionN)r#   append)r3   r7   s     r4   add_handlerzAlertManager.add_handler:   s    ""7+r6   statusespipeline_statusc                 l   |j                         D ]  \  }}|j                  }|| j                  kD  s"|j                  |z  }|| j                  kD  sAt        t        j                  t        j                  d| d|dt        j                         ||t        |      d      }| j                  |        |j                         }|| j                  kD  r_t        t        j                  t        j                  d| t        j                         ddt        |      i      }| j                  |       y	y	)
z4Check for alert conditions and trigger notificationsz	Provider z has high error rate: z.2%)provider
error_rate)typelevelmessage	timestampsourcecontextzTotal queue size is high: queue_monitor
queue_sizeN)itemscallsr"   errorsr-   r   r   PERFORMANCEr   WARNINGr+   str_trigger_alertrH   r.   SYSTEM)	r3   r<   r=   namestatustotal_callsr@   alerttotal_queue_sizes	            r4   check_alertszAlertManager.check_alerts>   s     %NN, 	/LD& ,,KT111#]][8
 4 44!)55(00"+D61G
SVGW X"&))+#-1Z QE ''.	/  +557d222!(( ((45E4FG))+&%s+;'<=E & 3r6   rT   c                 &   | j                   5  |j                  j                   d|j                   }t	        j                         }| j                  |       || j                  v r%| j                  |   }||z
  t        k  r
	 ddd       y|| j                  |<   | j                  j                  |       | j                  D ]  }	  ||        	 ddd       y# t        $ r"}t        j                  d|        Y d}~>d}~ww xY w# 1 sw Y   yxY w)z7Trigger an alert with efficient TTL-based deduplication:NzAlert handler error: )r*   rA   valuerE   r+   _cleanup_expired_alertsr'   r   r&   r:   r#   	Exceptionr0   error)r3   rT   	alert_keycurrent_time	last_sentr7   es          r4   rO   zAlertManager._trigger_alert_   s
   YY 	> ::++,Aell^<I99;L ((6 D000 00;	)+.DD	> 	> /;D  +%%e,  .. >>EN>'	> 	>, ! >LL#8!<==>-	> 	>s<   A5D9DCD	D"C?:D?DDDr^   c                     | j                   j                         D cg c]  \  }}||z
  t        k\  s| }}}|D ]  }| j                   |=  yc c}}w )z%Remove expired entries from TTL indexN)r'   rI   r   )r3   r^   keyrD   expired_keyss        r4   rZ   z$AlertManager._cleanup_expired_alertsz   sc     '+&:&:&@&@&B
"ClU^F^bxFxC
 
   	*C$$S)	*
s
   AAc                 &   t        j                          }| j                  5  | j                  |       || _        t	        | j
                        dkD  r,t        j                  dt	        | j
                         d       ddd       y# 1 sw Y   yxY w)z!Perform cleanup of expired alertsr   zAlert TTL cleanup: z active alert keys remainingN)r+   r*   rZ   r,   lenr'   r0   debug)r3   r^   s     r4   cleanupzAlertManager.cleanup   s{    yy{YY 	l((6%1D" 4''(1,23t7K7K3L2MMijk	l 	l 	ls   ABB)r8   N)__name__
__module____qualname____doc__r
   r5   r   r   r;   r   rN   r   r   rV   rO   floatrZ   rg    r6   r4   r   r   #   s    >1/ 1(,8UGTM#: ,t ,'T#~*=%> 'Q_ 'dh 'B>E >d >6*E *d *	lr6   r   c                       e Zd ZdZdefdZdeddfdZded	e	ddfd
Z
deddfdZdefdZdeee	f   fdZdee   fdZdefdZdefdZdeddfdZy)ProviderMonitoringz?Pure monitoring data aggregator without scheduling dependenciesr   c                 \   i | _         t               | _        t        |      | _        t        t        j                  dt        j                              | _
        t        j                         | _        | j                  j                  | j                         t         j#                  d       y )Nmax_stats_historyr    z7Provider monitoring initialized as pure data aggregator)provider_statusr   r=   r   alert_managerr   r   r$   r   r%   stats_historyr(   r)   r*   r;   _console_alert_handlerr0   r1   r2   s     r4   r5   zProviderMonitoring.__init__   s    :<-/)&1 %*(,,-@+BbBbc%
 NN$	 	&&t'B'BCMNr6   system_statsr8   Nc                    | j                   5  |j                  r4|j                  j                         D ]  \  }}| j                  ||        |j                  r| j                  |j                         t        j                         |j                  z
  | j                  _
        | j                  j                  | j                  | j                         ddd       y# 1 sw Y   yxY w)z1Ingest task statistics for monitoring aggregationN)r*   	providersrI   _update_providerpipeline_update_pipeliner+   	monotonicruntimer=   startrs   rV   rr   )r3   rv   rQ   rR   s       r4   ingestzProviderMonitoring.ingest   s    YY 	X%%$0$:$:$@$@$B 8LD&))$78 $$%%l&;&;< *.)9L<P<P)PD  & ++D,@,@$BVBVW	X 	X 	Xs   CCC'rQ   rR   c                    || j                   vrt        |      | j                   |<   | j                   |   }|j                  |_        |j                  |_        |j                  |_        |j
                  |_        |j                  |_        |j                  |_        |j                  |_        |j                  |_	        |j                  |_
        y)z,Update provider stats from task manager data)rQ   N)rr   r   r   enabledresourcerJ   rK   
searchable
gatherable	checkableinspectable)r3   rQ   rR   targets       r4   ry   z#ProviderMonitoring._update_provider   s    t+++)7T)BD  & %%d+|| //|| #--"--!++#//r6   r=   c                     |r|| _         yy)z,Update pipeline stats from task manager dataN)r=   )r3   r=   s     r4   r{   z#ProviderMonitoring._update_pipeline   s     #2D  r6   c                 X    t        j                         | j                  j                  z
  S )z4Get current runtime in seconds using monotonic clock)r+   r|   r=   r~   r3   s    r4   r}   zProviderMonitoring.runtime   s     ~~$"6"6"<"<<<r6   c                 z    | j                   5  | j                  j                         cddd       S # 1 sw Y   yxY w)zGet provider statisticsN)r*   rr   copyr   s    r4   get_provider_statusz&ProviderMonitoring.get_provider_status   s0    YY 	/'',,.	/ 	/ 	/s   1:c                 ^    | j                   5  | j                  cddd       S # 1 sw Y   yxY w)zGet pipeline statisticsN)r*   r=   r   s    r4   get_pipeline_statusz&ProviderMonitoring.get_pipeline_status   s'    YY 	(''	( 	( 	(s   #,c                    | j                   5  t               }| j                         }| j                  j	                         }| j
                  }t        d |j                         D              }t        d |j                         D              }|t        |d      z  |_	        |t        |d      z  |_
        d|j                  z
  |_        t        |      }t        ||||      cddd       S # 1 sw Y   yxY w)zGet current statistics snapshotc              3   4   K   | ]  }|j                     y wNrJ   .0ps     r4   	<genexpr>z.ProviderMonitoring.snapshot.<locals>.<genexpr>   s     BAB   c              3   4   K   | ]  }|j                     y wr   rK   r   s     r4   r   z.ProviderMonitoring.snapshot.<locals>.<genexpr>   s     DaDr   r   g      ?)r}   rz   rx   summaryN)r*   r   r}   rr   r   r=   sumvaluesmaxtasks_per_secondr@   success_rater   r   )r3   r   r}   rr   r=   tasksrK   providers_readonlys           r4   snapshotzProviderMonitoring.snapshot   s    YY 	(*GllnG"22779O"22O B)?)?)ABBED?+A+A+CDDF',s7A'>G$!'#eQ-!7G#&););#;G !1/!B%(,	!	 	 	s   CC--C6c           	      P   | j                   5  t        d | j                  j                         D              }t        d | j                  j                         D              }t        d | j                  j                         D              }t        d | j                  j                         D              }t        d | j                  j                         D              }| j	                         }t        ||||||      }|j                          |cddd       S # 1 sw Y   yxY w)z0Get summarized statistics as strong-typed objectc              3   4   K   | ]  }|j                     y wr   r   r   statss     r4   r   z-ProviderMonitoring.summary.<locals>.<genexpr>   s     OOr   c              3   N   K   | ]  }|j                   |j                  z
    y wr   )rJ   rK   r   s     r4   r   z-ProviderMonitoring.summary.<locals>.<genexpr>   s     b5EKK%,,6bs   #%c              3   4   K   | ]  }|j                     y wr   r   r   s     r4   r   z-ProviderMonitoring.summary.<locals>.<genexpr>   s     Q%Qr   c              3   H   K   | ]  }|j                   j                    y wr   )r   validr   s     r4   r   z-ProviderMonitoring.summary.<locals>.<genexpr>        X,,X    "c              3   H   K   | ]  }|j                   j                    y wr   )r   linksr   s     r4   r   z-ProviderMonitoring.summary.<locals>.<genexpr>   r   r   )r   	completedfailedr}   r   keysN)r*   r   rr   r   r}   r   update_performance_metrics)r3   r   r   r   r   r   r}   r   s           r4   r   zProviderMonitoring.summary   s    YY 	O1E1E1L1L1NOOEbDDXDXD_D_DabbIQ43G3G3N3N3PQQFX$:N:N:U:U:WXXEX$:N:N:U:U:WXXEllnG'#G ..0%	 	 	s   DDD%rT   c                     t        j                  d      }t        j                  d| d|j                  j
                   d|j                          y)zDefault console alert handlerz%H:%M:%S[z	] ALERT (z): N)r+   strftimer0   warningrA   rY   rC   )r3   rT   rD   s      r4   ru   z)ProviderMonitoring._console_alert_handler
  s>    MM*-	9+Yuzz/?/?.@EMM?STr6   )rh   ri   rj   rk   r
   r5   r   r   rN   r   ry   r   r{   rl   r}   r   r   r	   r   r   r   r   r   r   ru   rm   r6   r4   ro   ro      s    IO/ O X< XD X$0S 0. 0T 0&3 34 3= =/T#~*=%> /
(Xn%= (
, 2* ,UE Ud Ur6   ro   r   r8   c                     t        |       S )z,Factory function to create monitoring system)ro   )r   s    r4   create_monitoringr     s    f%%r6   )'rk   r(   r+   collectionsr   r   typesr   typingr   r   r   r	   config.schemasr
   constant.monitoringr   r   constant.systemr   
core.enumsr   tools.loggerr   enumsr   modelsr   r   r   r   r   r   r   r   r0   r   ro   r   rm   r6   r4   <module>r      s   
   * " 1 1 + B 2 # # 	 	 	 
G	hl hlVU) UD&. &3E &r6   