<s> import ConfigParser <EOL> import argparse <EOL> import datetime <EOL> import json <EOL> import logging <EOL> import os <EOL> import smtplib <EOL> import sys <EOL> import time <EOL> import utils . mail <EOL> import deploy_utils <EOL> from optparse import make_option <EOL> from os import path <EOL> from django . conf import settings <EOL> from django . core . management . base import BaseCommand , CommandError <EOL> from django . utils import timezone <EOL> from monitor . models import Status , Service , Cluster , Job , Task <EOL> BOOL_METRIC_MAP = { <EOL> "<STR_LIT>" : "<STR_LIT:true>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> STATUS_FILE_PATH = '<STR_LIT>' <EOL> ERROR_TIMES_FOR_ALERT = <NUM_LIT:3> <EOL> logger = logging . getLogger ( __name__ ) <EOL> class CollectorConfig : <EOL> class Service : <EOL> def __init__ ( self , options , config , name ) : <EOL> self . name = name <EOL> self . jobs = config . get ( name , "<STR_LIT>" ) . split ( ) <EOL> self . clusters = { } <EOL> for cluster_name in config . get ( name , "<STR_LIT>" ) . split ( ) : <EOL> args = argparse . Namespace ( ) <EOL> args . service = self . name <EOL> args . cluster = cluster_name <EOL> self . clusters [ cluster_name ] = deploy_utils . get_service_config ( args ) <EOL> self . metric_url = config . get ( name , "<STR_LIT>" ) <EOL> def __init__ ( self , args , options ) : <EOL> config_path = os . path . join ( deploy_utils . get_config_dir ( ) , '<STR_LIT>' ) <EOL> self . args = args <EOL> self . options = options <EOL> self . config = self . parse_config_file ( config_path ) <EOL> self . services = { } <EOL> for service_name in self . config . get ( "<STR_LIT>" , "<STR_LIT>" ) . split ( ) : <EOL> self . services [ service_name ] = CollectorConfig . Service ( <EOL> options , self . config , service_name ) <EOL> self . period = self . config . getint ( "<STR_LIT>" , "<STR_LIT>" ) <EOL> def parse_config_file ( self , config_path ) : <EOL> config_parser = ConfigParser . SafeConfigParser ( ) <EOL> config_parser . optionxform = str <EOL> logger . info ( "<STR_LIT>" , config_path ) <EOL> if not config_parser . read ( config_path ) : <EOL> logger . critical ( "<STR_LIT>" , config_path ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> logger . info ( "<STR_LIT>" ) <EOL> return config_parser <EOL> class StatusChecker : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , collector_config , last_status , options , mailer ) : <EOL> self . collector_config = collector_config <EOL> self . last_status = last_status <EOL> self . options = options <EOL> self . alert_msg = '<STR_LIT>' <EOL> self . mailer = mailer <EOL> def get_latest_metric ( self , task , group_name , metric_name ) : <EOL> try : <EOL> metric = json . loads ( task . last_metrics ) <EOL> return metric [ group_name ] [ metric_name ] <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" , task , e ) <EOL> return <NUM_LIT:0> <EOL> def is_namenode_active ( self , task ) : <EOL> try : <EOL> metric = self . get_latest_metric ( <EOL> task , "<STR_LIT>" , "<STR_LIT>" ) <EOL> return bool ( metric ) <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" , task , e ) <EOL> return False <EOL> def is_master_active ( self , task ) : <EOL> try : <EOL> metric = self . get_latest_metric ( <EOL> task , "<STR_LIT>" , "<STR_LIT>" ) <EOL> return bool ( metric ) <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" , task , e ) <EOL> return False <EOL> def check_hdfs_cluster_status ( self , cluster ) : <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if ( job . running_tasks_count < <NUM_LIT:2> or <EOL> job . running_tasks_count < ( job . total_tasks_count / <NUM_LIT:2> + <NUM_LIT:1> ) ) : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> else : <EOL> active = <NUM_LIT:0> <EOL> for task in job . running_tasks . itervalues ( ) : <EOL> if self . is_namenode_active ( task ) : <EOL> cluster . entry = '<STR_LIT>' % ( task . host , task . port ) <EOL> cluster . version = self . get_latest_metric ( task , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> active += <NUM_LIT:1> <EOL> if active > <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> elif active < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> elif job . running_tasks_count < <NUM_LIT:2> : <EOL> job . last_status = Status . WARN <EOL> job . last_message = "<STR_LIT>" <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:3> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> cluster . last_status = max ( [ job . last_status for job in cluster . jobs . itervalues ( ) ] ) <EOL> def check_hbase_cluster_status ( self , cluster ) : <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> else : <EOL> active = <NUM_LIT:0> <EOL> for task in job . running_tasks . itervalues ( ) : <EOL> if self . is_master_active ( task ) : <EOL> cluster . entry = '<STR_LIT>' % ( task . host , task . port ) <EOL> version = self . get_latest_metric ( task , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:version>' ) <EOL> revision = self . get_latest_metric ( task , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> cluster . version = '<STR_LIT>' % ( version , revision ) <EOL> active += <NUM_LIT:1> <EOL> if active > <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> elif active < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> elif job . running_tasks_count < <NUM_LIT:2> : <EOL> pass <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:3> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> cluster . last_status = max ( [ job . last_status for job in cluster . jobs . itervalues ( ) ] ) <EOL> def check_yarn_cluster_status ( self , cluster ) : <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> for task in job . running_tasks . itervalues ( ) : <EOL> cluster . entry = '<STR_LIT>' % ( task . host , task . port ) <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:3> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> cluster . last_status = max ( [ job . last_status for job in cluster . jobs . itervalues ( ) ] ) <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> cluster . last_status = max ( [ job . last_status for job in cluster . jobs . itervalues ( ) ] ) <EOL> def check_impala_cluster_status ( self , cluster ) : <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> for task in job . running_tasks . itervalues ( ) : <EOL> cluster . entry = '<STR_LIT>' % ( task . host , task . port ) <EOL> if job . running_tasks_count < <NUM_LIT:1> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> job = cluster . jobs [ "<STR_LIT>" ] <EOL> if job . running_tasks_count < <NUM_LIT:3> : <EOL> job . last_status = Status . ERROR <EOL> job . last_message = "<STR_LIT>" <EOL> cluster . last_status = max ( [ job . last_status for job in cluster . jobs . itervalues ( ) ] ) <EOL> def check_cluster_status ( self , cluster ) : <EOL> cluster . jobs = { } <EOL> cluster . last_status = Status . OK <EOL> cluster . last_message = "<STR_LIT>" <EOL> for job in cluster . job_set . all ( ) : <EOL> job . running_tasks = { } <EOL> job . tasks = { } <EOL> job . last_status = Status . OK <EOL> job . last_message = "<STR_LIT>" <EOL> job . running_tasks_count = <NUM_LIT:0> <EOL> job . total_tasks_count = <NUM_LIT:0> <EOL> for task in job . task_set . filter ( active = True ) : <EOL> if task . health : <EOL> job . running_tasks [ task . id ] = task <EOL> job . running_tasks_count += <NUM_LIT:1> <EOL> job . total_tasks_count += <NUM_LIT:1> <EOL> cluster . jobs [ job . name ] = job <EOL> service_handler = { <EOL> "<STR_LIT>" : self . check_hdfs_cluster_status , <EOL> "<STR_LIT>" : self . check_hbase_cluster_status , <EOL> "<STR_LIT>" : self . check_yarn_cluster_status , <EOL> "<STR_LIT>" : self . check_impala_cluster_status , <EOL> } <EOL> service_handler [ cluster . service . name ] ( cluster ) <EOL> self . handle_status_result ( cluster ) <EOL> def handle_status_result ( self , cluster ) : <EOL> ( cluster_status , status_times ) = self . last_status . setdefault ( str ( cluster ) , ( Status . OK , <NUM_LIT:0> ) ) <EOL> need_send_alert = False <EOL> if cluster . last_status != cluster_status : <EOL> self . last_status [ str ( cluster ) ] = ( cluster . last_status , <NUM_LIT:1> ) <EOL> if cluster . last_status == Status . OK and status_times >= ERROR_TIMES_FOR_ALERT : <EOL> need_send_alert = True <EOL> else : <EOL> self . last_status [ str ( cluster ) ] = ( cluster . last_status , status_times + <NUM_LIT:1> ) <EOL> if cluster . last_status != Status . OK and status_times + <NUM_LIT:1> == ERROR_TIMES_FOR_ALERT : <EOL> need_send_alert = True <EOL> if need_send_alert : <EOL> self . alert_msg += '<STR_LIT>' % ( '<STR_LIT:OK>' if cluster . last_status == Status . OK else '<STR_LIT>' , <EOL> cluster ) <EOL> for job in cluster . jobs . itervalues ( ) : <EOL> if job . last_status != Status . OK : <EOL> self . alert_msg += '<STR_LIT>' % ( job . name , job . last_message ) <EOL> self . alert_msg += '<STR_LIT>' <EOL> def check_status ( self ) : <EOL> self . alert_msg = '<STR_LIT>' <EOL> logger . info ( "<STR_LIT>" ) <EOL> self . start_time = time . time ( ) <EOL> for cluster in Cluster . objects . filter ( active = True ) . all ( ) : <EOL> self . check_cluster_status ( cluster ) <EOL> logger . info ( "<STR_LIT>" , <EOL> time . time ( ) - self . start_time ) <EOL> if self . alert_msg : <EOL> logger . warn ( '<STR_LIT>' % self . alert_msg ) <EOL> self . mailer . send_email ( subject = '<STR_LIT>' , <EOL> content = self . alert_msg , <EOL> to_email = self . options [ '<STR_LIT>' ] ) <EOL> json . dump ( self . last_status , open ( STATUS_FILE_PATH , '<STR_LIT:w>' ) ) <EOL> class Command ( BaseCommand ) : <EOL> args = '<STR_LIT>' <EOL> help = "<STR_LIT>" <EOL> option_list = BaseCommand . option_list + ( <EOL> make_option ( <EOL> "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) , <EOL> make_option ( <EOL> "<STR_LIT>" , <EOL> default = <NUM_LIT> , <EOL> help = "<STR_LIT>" ) , <EOL> ) <EOL> def handle ( self , * args , ** options ) : <EOL> self . args = args <EOL> self . options = options <EOL> self . mailer = utils . mail . Mailer ( options ) <EOL> self . stdout . write ( "<STR_LIT>" % ( args , ) ) <EOL> self . stdout . write ( "<STR_LIT>" % options ) <EOL> self . collector_config = CollectorConfig ( self . args , self . options ) <EOL> self . last_status = { } <EOL> try : <EOL> self . last_status = json . load ( open ( STATUS_FILE_PATH , '<STR_LIT:r>' ) ) <EOL> except Exception as e : <EOL> logger . warning ( '<STR_LIT>' , e ) <EOL> status_checker = StatusChecker ( self . collector_config , <EOL> self . last_status , <EOL> self . options , <EOL> self . mailer ) <EOL> while True : <EOL> try : <EOL> status_checker . check_status ( ) <EOL> except Exception as e : <EOL> logger . warning ( '<STR_LIT>' , e ) <EOL> admin_email = '<STR_LIT>' <EOL> try : <EOL> admin_email = settings . ADMINS [ <NUM_LIT:0> ] [ <NUM_LIT:1> ] <EOL> except : <EOL> pass <EOL> self . mailer . send_email ( subject = '<STR_LIT>' , <EOL> content = repr ( e ) , <EOL> to_email = admin_email , <EOL> ) <EOL> time . sleep ( int ( self . options [ '<STR_LIT>' ] ) ) </s>
