<s> from collections import defaultdict <EOL> from Queue import Empty , Queue <EOL> import threading <EOL> import time <EOL> from checks import AgentCheck <EOL> from checks . libs . thread_pool import Pool <EOL> from config import _is_affirmative <EOL> TIMEOUT = <NUM_LIT> <EOL> DEFAULT_SIZE_POOL = <NUM_LIT:6> <EOL> MAX_LOOP_ITERATIONS = <NUM_LIT:1000> <EOL> FAILURE = "<STR_LIT>" <EOL> class Status : <EOL> DOWN = "<STR_LIT>" <EOL> WARNING = "<STR_LIT>" <EOL> UP = "<STR_LIT>" <EOL> class EventType : <EOL> DOWN = "<STR_LIT>" <EOL> UP = "<STR_LIT>" <EOL> class NetworkCheck ( AgentCheck ) : <EOL> SOURCE_TYPE_NAME = '<STR_LIT>' <EOL> SERVICE_CHECK_PREFIX = '<STR_LIT>' <EOL> STATUS_TO_SERVICE_CHECK = { <EOL> Status . UP : AgentCheck . OK , <EOL> Status . WARNING : AgentCheck . WARNING , <EOL> Status . DOWN : AgentCheck . CRITICAL <EOL> } <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , name , init_config , agentConfig , instances ) : <EOL> AgentCheck . __init__ ( self , name , init_config , agentConfig , instances ) <EOL> self . statuses = { } <EOL> self . notified = { } <EOL> self . nb_failures = <NUM_LIT:0> <EOL> self . pool_started = False <EOL> names = [ ] <EOL> for inst in instances : <EOL> if '<STR_LIT:name>' not in inst : <EOL> raise Exception ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( inst ) ) <EOL> if inst [ '<STR_LIT:name>' ] in names : <EOL> raise Exception ( "<STR_LIT>" <EOL> . format ( inst [ '<STR_LIT:name>' ] ) ) <EOL> def stop ( self ) : <EOL> self . stop_pool ( ) <EOL> self . pool_started = False <EOL> def start_pool ( self ) : <EOL> self . log . info ( "<STR_LIT>" ) <EOL> default_size = min ( self . instance_count ( ) , DEFAULT_SIZE_POOL ) <EOL> self . pool_size = int ( self . init_config . get ( '<STR_LIT>' , default_size ) ) <EOL> self . pool = Pool ( self . pool_size ) <EOL> self . resultsq = Queue ( ) <EOL> self . jobs_status = { } <EOL> self . pool_started = True <EOL> def stop_pool ( self ) : <EOL> self . log . info ( "<STR_LIT>" ) <EOL> if self . pool_started : <EOL> self . pool . terminate ( ) <EOL> self . pool . join ( ) <EOL> self . jobs_status . clear ( ) <EOL> assert self . pool . get_nworkers ( ) == <NUM_LIT:0> <EOL> def restart_pool ( self ) : <EOL> self . stop_pool ( ) <EOL> self . start_pool ( ) <EOL> def check ( self , instance ) : <EOL> if not self . pool_started : <EOL> self . start_pool ( ) <EOL> if threading . activeCount ( ) > <NUM_LIT:5> * self . pool_size + <NUM_LIT:5> : <EOL> raise Exception ( "<STR_LIT>" % threading . activeCount ( ) ) <EOL> self . _process_results ( ) <EOL> self . _clean ( ) <EOL> name = instance . get ( '<STR_LIT:name>' , None ) <EOL> if name is None : <EOL> self . log . error ( '<STR_LIT>' ) <EOL> return <EOL> if name not in self . jobs_status : <EOL> self . jobs_status [ name ] = time . time ( ) <EOL> self . pool . apply_async ( self . _process , args = ( instance , ) ) <EOL> else : <EOL> self . log . error ( "<STR_LIT>" % name ) <EOL> def _process ( self , instance ) : <EOL> try : <EOL> statuses = self . _check ( instance ) <EOL> if isinstance ( statuses , tuple ) : <EOL> status , msg = statuses <EOL> self . resultsq . put ( ( status , msg , None , instance ) ) <EOL> elif isinstance ( statuses , list ) : <EOL> for status in statuses : <EOL> sc_name , status , msg = status <EOL> self . resultsq . put ( ( status , msg , sc_name , instance ) ) <EOL> except Exception : <EOL> result = ( FAILURE , FAILURE , FAILURE , FAILURE ) <EOL> self . resultsq . put ( result ) <EOL> def _process_results ( self ) : <EOL> for i in range ( MAX_LOOP_ITERATIONS ) : <EOL> try : <EOL> status , msg , sc_name , instance = self . resultsq . get_nowait ( ) <EOL> except Empty : <EOL> break <EOL> if status == FAILURE : <EOL> self . nb_failures += <NUM_LIT:1> <EOL> if self . nb_failures >= self . pool_size - <NUM_LIT:1> : <EOL> self . nb_failures = <NUM_LIT:0> <EOL> self . restart_pool ( ) <EOL> continue <EOL> self . report_as_service_check ( sc_name , status , instance , msg ) <EOL> skip_event = _is_affirmative ( instance . get ( '<STR_LIT>' , False ) ) <EOL> instance_name = instance [ '<STR_LIT:name>' ] <EOL> if not skip_event : <EOL> self . warning ( "<STR_LIT>" ) <EOL> event = None <EOL> if instance_name not in self . statuses : <EOL> self . statuses [ instance_name ] = defaultdict ( list ) <EOL> self . statuses [ instance_name ] [ sc_name ] . append ( status ) <EOL> window = int ( instance . get ( '<STR_LIT>' , <NUM_LIT:1> ) ) <EOL> if window > <NUM_LIT> : <EOL> self . log . warning ( "<STR_LIT>" ) <EOL> window = <NUM_LIT> <EOL> threshold = instance . get ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> if len ( self . statuses [ instance_name ] [ sc_name ] ) > window : <EOL> self . statuses [ instance_name ] [ sc_name ] . pop ( <NUM_LIT:0> ) <EOL> nb_failures = self . statuses [ instance_name ] [ sc_name ] . count ( Status . DOWN ) <EOL> if nb_failures >= threshold : <EOL> if self . notified . get ( ( instance_name , sc_name ) , Status . UP ) != Status . DOWN : <EOL> event = self . _create_status_event ( sc_name , status , msg , instance ) <EOL> self . notified [ ( instance_name , sc_name ) ] = Status . DOWN <EOL> else : <EOL> if self . notified . get ( ( instance_name , sc_name ) , Status . UP ) != Status . UP : <EOL> event = self . _create_status_event ( sc_name , status , msg , instance ) <EOL> self . notified [ ( instance_name , sc_name ) ] = Status . UP <EOL> if event is not None : <EOL> self . events . append ( event ) <EOL> if instance_name in self . jobs_status : <EOL> del self . jobs_status [ instance_name ] <EOL> def _check ( self , instance ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError <EOL> def _clean ( self ) : <EOL> now = time . time ( ) <EOL> for name in self . jobs_status . keys ( ) : <EOL> start_time = self . jobs_status [ name ] <EOL> if now - start_time > TIMEOUT : <EOL> self . log . critical ( "<STR_LIT>" % name ) <EOL> self . restart_pool ( ) <EOL> break </s>
