
    `+nj                     P   d dl Z d dl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 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 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& ddl'm(Z( ddl)m*Z*m+Z+  e jX                  e-      Z.de/defdZ0 G d d      Z1y)    N)ThreadPoolExecutor)TimeoutError)datetime)Path)Anycast   )	CollectorCollectorContext)CronCollector)DockerCollector)GenericWorkerCollector)ProcessCollector)SystemdCollector)	AppConfig)connect)apply_migrations)WatcherRepository)AgentRegistrationCollectorErrorCollectorResultRemoteEventWatcherCycleResult)observation_from_remote_event)isoutc_nowconfig_typereturnc                     t         t        t        t        t        t        d}| |vrt        d|        t        t         ||                 S )N)crondockersystemdprocessgeneric_workerzgeneric-workerzunsupported collector type: )r   r   r   r   r   
ValueErrorr   r
   )r   
collectorss     K/opt/struktur/hermes-workspace/worker-watcher/src/worker_watcher/watcher.pybuild_collectorr(       sP    *7?_o.>Rh5K"MJ *$7}EFF	2:k2455    c                       e Zd Zddededz  ddfdZedefd       Zde	fdZ
deddfd	Zd
edefdZdedeeef   fdZdeeef   fdZddZy)WatcherNconfigbase_dirr   c                 ,   || _         |xs t        j                         | _        t	        |j
                  j                        | _        t        | j                        | _	        t               | _        d | _        d | _        d | _        d | _        i | _        y N)r,   r   cwdr-   r   watcherdatabase_path
connectionr   schema_versionr   
started_atlast_cycle_started_atlast_cycle_finished_atlast_successful_cycle_atlast_cycle_duration_mslast_results)selfr,   r-   s      r'   __init__zWatcher.__init__*   ss     .DHHJ!&..">">?.t?!)6:"7;#9=%26#8:r)   c                 ,    t        | j                        S r/   )r   r3   r;   s    r'   
repositoryzWatcher.repository6   s     11r)   c                     t                _         j                  j                  D cg c]  }|j                  s| }}g }t        j                         dt        dt        f fd}|D ]w  }	 t        dd|j                         5 }|j                  ||      }	 |j                  |j                        }d d d        |j)                          j+                  |       y t               }	|	 _        t/        |	z
  j1                         dz         _        |D 
ci c]  }
|
j4                  |
 c}
 _        |rt9        d |D              r|	 _         j<                  j?                          tA        |	tC        |      tE        d |D              tE        d |D              tE        d |D                    S c c}w # t        $ rA t        |j                  dt               d	d
t        dd|j                   d      fi       }Y Mt         $ rV}t        |j                  dt               d	d
t        t#        |      j$                  t'        |            fi       }Y d }~d }~ww xY w# 1 sw Y   xY w# t         $ rV}t        |j                  dt               d	d
t        t#        |      j$                  t'        |            fi       }Y d }~d }~ww xY wc c}
w )Nitemr   c                     t        | j                        }t        | j                        }|j	                  |      S r/   )r(   typer   r-   collect)rA   	collectorcontextcontext_hostr;   starteds      r'   executez!Watcher.run_once.<locals>.executeA   s6    '		2I&wdDMMRG$$W--r)   r	   z
collector-)max_workersthread_name_prefix)timeoutunknownF rL   zcollector timed out after si  c              3   4   K   | ]  }|j                     y wr/   success.0results     r'   	<genexpr>z#Watcher.run_once.<locals>.<genexpr>[   s     @f6>>@   c              3   4   K   | ]  }|j                     y wr/   rQ   rS   s     r'   rV   z#Watcher.run_once.<locals>.<genexpr>^   s     FlZ`v~~FlrW   c              3   6   K   | ]  }|j                      y wr/   rQ   rS   s     r'   rV   z#Watcher.run_once.<locals>.<genexpr>_   s     %OV&..&8%Os   c              3   F   K   | ]  }t        |j                          y wr/   )lenobservationsrS   s     r'   rV   z#Watcher.run_once.<locals>.<genexpr>_   s&       UErxUXY_YlYlUm  UEs   !)#r   r6   r,   r&   enabledsocketgethostnamer   r   r   namesubmitrU   timeout_secondsFutureTimeoutr   	ExceptionrC   __name__strappend_persist_resultr7   inttotal_secondsr9   collector_namer:   allr8   r3   commitr   r[   sum)r;   rA   r]   resultsrI   executorfuturecurrentexcfinishedrU   rG   rH   s   `          @@r'   run_oncezWatcher.run_once:   s   )%,"$(KK$:$:KDdll4KK)+))+	.# 	./ 	.
  	*D_'AJW[W`W`VaJbc 	ggo%__Wd;Fg"(--8L8L-"M	g NN7#  )!	*" 9&.#&)8g+=*L*L*NQU*U&V#IPQvV22F:Qs@@@,4D) !'8S\3FldkFlCl"%%Ow%O"OQT  UE  }D  UE  REF 	FC L ) J"1$))YQXQZ\ace3A)Oijnj~j~i  @A  NB  4C  3E  GI#J$ g"1$))YQXQZ\ace3A$s)BTBTVYZ]V^3_2ace#gg	g 	g  _)$))YTY[]+9$s):L:LcRUh+W*Y[]__ Rss   F>F>5I:I-#G?I:$KAI*	I-I*AI%I-%I**I--I7	2I::	KAKKrU   c                    | j                   }d}| j                  5  |j                  D ]$  }|j                  |d |j                         |dz  }& |j                  ||       |j                  sQ|j                  D ]B  }|j                  dd|j                  |j                  |j                  |j                  d       D d d d        y # 1 sw Y   y xY w)Nr   r	   collector_failedwarning)
error_typeinstance_key)r?   r3   r\   persist_observationfinished_atsave_collector_runrR   errorssave_watcher_eventmessagerk   ry   rz   )r;   rU   r?   observations_createdobservationerrors         r'   rh   zWatcher._persist_resulta   s    __
 __ 	x%22 *..{D&BTBTU$)$* ))&2FG>>#]] xE112DiQVQ^Q^`f`u`uAFAQAQchcucu2vxx	x 	x 	xs   B#CCregistrationc                     | j                   5  | j                  j                  |t                     cd d d        S # 1 sw Y   y xY wr/   )r3   r?   ensure_agentr   )r;   r   s     r'   register_remote_agentzWatcher.register_remote_agentn   s8    __ 	I??//giH	I 	I 	Is	   $;Aeventc                 n   t               }| j                  }| j                  5  |j                  |j                  |      }|j                  |||      \  }}|sdd|dcd d d        S t        |      }|j                  |d |      \  }}	|j                  |d       dd|||	dcd d d        S # 1 sw Y   y xY w)NT)accepted	duplicateremote_event_idF)r   r   r   observation_idstatus_changed)	r   r?   r3   r   agentrecord_remote_eventr   r{   mark_remote_event)
r;   r   received_atr?   agent_idr   createdr   r   changeds
             r'   ingest_remote_eventzWatcher.ingest_remote_eventr   s    i__
__ 		Q!..u{{KHH'1'E'EeXWb'c$OW$(tP_`			Q 		Q
 8>K&0&D&D[RVXc&d#NG(($? $5_&4Q		Q 		Q 		Qs   :B+':B++B4c                    | j                   j                  t               | j                  j                  j
                         | j                   j                         t        d | j                  j                         D              }t        d | j                  j                         D              }d}	 | j                  j                  d      j                          t        fddD              }|dk(  r	|dk(  r|sd	nd
}t        d | j                  j                  D              }|d| j                   t#        t                     t#        | j$                        t#        | j&                        t#        | j(                        t#        | j*                        | j,                  |t/        | j                  j                        |t/        | j                  j                        |z
  ||dS # t        $ r d}Y w xY w)Nc              3   :   K   | ]  }|j                   sd   ywr	   NrQ   rS   s     r'   rV   z!Watcher.health.<locals>.<genexpr>   s     Vvv~~V   c              3   :   K   | ]  }|j                   rd   ywr   rQ   rS   s     r'   rV   z!Watcher.health.<locals>.<genexpr>   s     V6v~~QVr   okzSELECT 1r   c              3   .   K   | ]  }|   d kD    yw)r   NrN   )rT   keysummarys     r'   rV   z!Watcher.health.<locals>.<genexpr>   s     sSq(ss   )jobs_degradedjobs_failed
jobs_stalejobs_unknownr   healthydegradedc              3   :   K   | ]  }|j                   sd   ywr   )r]   )rT   rA   s     r'   rV   z!Watcher.health.<locals>.<genexpr>   s      Vt Vr   z0.1.0)watcher_statusapplication_versionr4   current_time_utcr5   r6   r7   r8   r9   database_statuscollectors_totalcollectors_enabledcollectors_disabledcollectors_successfulcollectors_failed)r?   refresh_remote_stalenessr   r,   r1   remote_stale_after_secondsr   rn   r:   valuesr3   rI   fetchonerd   anyr&   r4   r   r5   r6   r7   r8   r9   r[   )r;   
successfulfailedr   	job_issuer   enabled_collectorsr   s          @r'   healthzWatcher.health   s   00DKK<O<O<j<jk//))+V):):)A)A)CVV
VT%6%6%=%=%?VV	&OO##J/88: s4rss	&5&=&A+V_eo  Vt{{/E/E VV"0dhdwdw$'	N#dooBV),T-G-G)Hdghl  iD  iD  eE,/0M0M,Njn  kF  kF#2DKKLbLbHc&8QTUYU`U`UkUkQl  pB  RB)3&] U\] 	]  	&%O	&s   5)G* *G98G9c                 8    | j                   j                          y r/   )r3   closer>   s    r'   r   zWatcher.close   s    r)   r/   )r   N)re   
__module____qualname__r   r   r<   propertyr   r?   r   ru   r   rh   r   ri   r   r   dictrf   r   r   r   r   rN   r)   r'   r+   r+   )   s    
;y 
;D4K 
;4 
; 2- 2 2%F, %FNxo x$ xI2C I IQ Qc3h Q]S#X ]* r)   r+   )2loggingr^   concurrent.futuresr   r   rc   r   pathlibr   typingr   r   collectors.baser
   r   collectors.cronr   collectors.dockerr   collectors.generic_workerr   collectors.processr   collectors.systemdr   r,   r   db.connectionr   db.migrationsr   db.repositoriesr   modelsr   r   r   r   r   remoter   timeutilr   r   	getLoggerre   LOGGERrf   r(   r+   rN   r)   r'   <module>r      s{      1 <    8 * . = 0 0  " + .  2 "			8	$6 6 6m  m r)   