<s> import Queue <EOL> import datetime <EOL> import json <EOL> import logging <EOL> import os <EOL> import socket <EOL> import time <EOL> import traceback <EOL> from collect_utils import METRIC_TASK_TYPE , STATUS_TASK_TYPE , AGGREGATE_TASK_TYPE <EOL> from collect_utils import QueueTask <EOL> from django . db import connection <EOL> from monitor import dbutil <EOL> from monitor import metric_helper <EOL> from monitor . models import Region , RegionServer , Table , HBaseCluster <EOL> REGION_SERVER_DYNAMIC_STATISTICS_BEAN_NAME = "<STR_LIT>" "<STR_LIT>" <EOL> REGION_SERVER_BEAN_NAME = "<STR_LIT>" <EOL> REGION_SERVER_REPLICATION_BEAN_NAME_PREFIX = "<STR_LIT>" "<STR_LIT>" <EOL> BOOL_METRIC_MAP = { <EOL> "<STR_LIT>" : "<STR_LIT:true>" , <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> } <EOL> HBASE_AGGREGATED_METRICS_KEY = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> logger = logging . getLogger ( __name__ ) <EOL> def reset_aggregated_metrics ( record ) : <EOL> for key in HBASE_AGGREGATED_METRICS_KEY : <EOL> setattr ( record , key , <NUM_LIT:0> ) <EOL> def aggregate_metrics ( from_record , to_record ) : <EOL> for key in HBASE_AGGREGATED_METRICS_KEY : <EOL> old_value = getattr ( to_record , key ) <EOL> setattr ( to_record , key , old_value + getattr ( from_record , key ) ) <EOL> def analyze_hbase_region_server_metrics ( metric_task , metrics ) : <EOL> region_server_name = None <EOL> region_operation_metrics_dict = { } <EOL> replication_metrics_dict = { } <EOL> for bean in metrics [ '<STR_LIT>' ] : <EOL> try : <EOL> if bean [ '<STR_LIT:name>' ] == REGION_SERVER_BEAN_NAME : <EOL> region_server_name = bean [ '<STR_LIT>' ] <EOL> elif bean [ '<STR_LIT:name>' ] == REGION_SERVER_DYNAMIC_STATISTICS_BEAN_NAME : <EOL> for metricName in bean . keys ( ) : <EOL> if Region . is_region_operation_metric_name ( metricName ) : <EOL> encodeName = Region . get_encode_name_from_region_operation_metric_name ( metricName ) <EOL> region_operation_metrics = region_operation_metrics_dict . setdefault ( encodeName , { } ) <EOL> region_operation_metrics [ metricName ] = bean [ metricName ] <EOL> elif bean [ '<STR_LIT:name>' ] . startswith ( REGION_SERVER_REPLICATION_BEAN_NAME_PREFIX ) : <EOL> peerId = metric_helper . parse_replication_source ( bean [ '<STR_LIT:name>' ] ) <EOL> replication_metrics = replication_metrics_dict . setdefault ( peerId , { } ) <EOL> for metricName in bean . keys ( ) : <EOL> replication_metrics [ metricName ] = bean [ metricName ] <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" , metric_task , e ) <EOL> continue <EOL> region_server = None <EOL> if region_server_name is None : <EOL> return <EOL> else : <EOL> try : <EOL> region_server = RegionServer . objects . get ( name = region_server_name ) <EOL> except RegionServer . DoesNotExist : <EOL> logger . warning ( "<STR_LIT>" , <EOL> metric_task , region_server_name ) <EOL> return <EOL> region_server . replication_last_attempt_time = metric_task . last_attempt_time <EOL> region_server . replicationMetrics = json . dumps ( replication_metrics_dict ) <EOL> region_server . save ( ) <EOL> region_record_need_save = [ ] <EOL> for encodeName , operationMetrics in region_operation_metrics_dict . iteritems ( ) : <EOL> region_record = dbutil . get_region_by_regionserver_and_encodename ( <EOL> region_server , encodeName ) <EOL> if region_record is None : <EOL> continue <EOL> region_record . analyze_from_region_server_operation_metrics ( operationMetrics , <EOL> metric_task . last_attempt_time ) <EOL> region_record_need_save . append ( region_record ) <EOL> begin = datetime . datetime . now ( ) <EOL> dbutil . update_regions_for_region_server_metrics ( region_record_need_save ) <EOL> logger . info ( "<STR_LIT>" "<STR_LIT>" , <EOL> metric_task , len ( region_record_need_save ) , <EOL> str ( ( datetime . datetime . now ( ) - begin ) . total_seconds ( ) ) ) <EOL> def get_host_and_port_from_region_server_name ( rs_name ) : <EOL> tokens = rs_name . split ( '<STR_LIT:U+002C>' ) <EOL> host = tokens [ <NUM_LIT:0> ] <EOL> host_name = None <EOL> try : <EOL> host_name = socket . gethostbyaddr ( host ) [ <NUM_LIT:0> ] <EOL> except : <EOL> logger . warning ( "<STR_LIT>" , host ) <EOL> host_name = host <EOL> port = int ( tokens [ <NUM_LIT:1> ] ) + <NUM_LIT:1> <EOL> return [ host_name , port ] <EOL> def analyze_hbase_master_metrics ( metric_task , metrics ) : <EOL> cluster = metric_task . job . cluster <EOL> hbase_cluster_record , created = HBaseCluster . objects . get_or_create ( cluster = cluster ) <EOL> reset_aggregated_metrics ( hbase_cluster_record ) <EOL> tables = { } <EOL> region_record_need_save = [ ] <EOL> for bean in metrics [ '<STR_LIT>' ] : <EOL> try : <EOL> if '<STR_LIT>' not in bean : <EOL> continue <EOL> for rs_metrics in bean [ '<STR_LIT>' ] : <EOL> rs_name = rs_metrics [ '<STR_LIT:key>' ] <EOL> [ rs_hostname , rs_port ] = get_host_and_port_from_region_server_name ( rs_name ) <EOL> rs_task = dbutil . get_task_by_host_and_port ( rs_hostname , rs_port ) <EOL> rs_record , created = RegionServer . objects . get_or_create ( cluster = cluster , <EOL> task = rs_task ) <EOL> rs_record . name = rs_name <EOL> rs_value = rs_metrics [ '<STR_LIT:value>' ] <EOL> rs_record . last_attempt_time = metric_task . last_attempt_time <EOL> rs_record . load = int ( rs_value [ '<STR_LIT>' ] ) <EOL> rs_record . numberOfRegions = int ( rs_value [ '<STR_LIT>' ] ) <EOL> reset_aggregated_metrics ( rs_record ) <EOL> all_regions_in_rs = Region . objects . filter ( region_server = rs_record ) <EOL> all_regions_in_rs = dbutil . get_alive_regions_by_rs ( rs_record ) <EOL> all_regions_map = { } <EOL> logger . info ( "<STR_LIT>" , metric_task , len ( all_regions_in_rs ) ) <EOL> for region in all_regions_in_rs : <EOL> all_regions_map [ region . name ] = region <EOL> regionsLoad = rs_value [ '<STR_LIT>' ] <EOL> for region_metrics in regionsLoad : <EOL> region_value = region_metrics [ '<STR_LIT:value>' ] <EOL> region_name = region_value [ '<STR_LIT>' ] <EOL> try : <EOL> table_name = region_name . split ( '<STR_LIT:U+002C>' ) [ <NUM_LIT:0> ] <EOL> except Exception as e : <EOL> logger . warning ( "<STR_LIT>" , <EOL> metric_task , e , region_name ) <EOL> continue <EOL> region_metrics = { } <EOL> if table_name not in tables : <EOL> table_record , created = Table . objects . get_or_create ( cluster = cluster , <EOL> name = table_name ) <EOL> reset_aggregated_metrics ( table_record ) <EOL> tables [ table_name ] = table_record <EOL> table_record = tables [ table_name ] <EOL> region_record = None <EOL> if region_name in all_regions_map : <EOL> region_record = all_regions_map [ region_name ] <EOL> else : <EOL> begin = datetime . datetime . now ( ) <EOL> region_record , created = Region . objects . get_or_create ( table = table_record , <EOL> name = region_name , encodeName = Region . get_encode_name ( region_name ) , <EOL> defaults = { "<STR_LIT>" : rs_record } ) <EOL> logger . info ( "<STR_LIT>" "<STR_LIT>" , <EOL> metric_task , str ( ( datetime . datetime . now ( ) - begin ) . total_seconds ( ) ) , <EOL> region_name , rs_record . name , region_record . region_server . name ) <EOL> logger . info ( "<STR_LIT>" , metric_task ) <EOL> region_record . region_server = rs_record <EOL> region_record . analyze_region_record ( region_value , <EOL> metric_task . last_attempt_time ) <EOL> region_record_need_save . append ( region_record ) <EOL> aggregate_metrics ( region_record , rs_record ) <EOL> aggregate_metrics ( region_record , table_record ) <EOL> aggregate_metrics ( region_record , hbase_cluster_record ) <EOL> rs_record . save ( ) <EOL> for table_record in tables . itervalues ( ) : <EOL> table_record . last_attempt_time = metric_task . last_attempt_time <EOL> table_record . availability = dbutil . getTableAvailability ( <EOL> table_record . cluster . name , table_record . name ) <EOL> table_record . save ( ) <EOL> hbase_cluster_record . save ( ) <EOL> begin = datetime . datetime . now ( ) <EOL> dbutil . update_regions_for_master_metrics ( region_record_need_save ) <EOL> logger . info ( "<STR_LIT>" "<STR_LIT>" , <EOL> metric_task , len ( region_record_need_save ) , <EOL> str ( ( datetime . datetime . now ( ) - begin ) . total_seconds ( ) ) ) <EOL> except Exception as e : <EOL> traceback . print_exc ( ) <EOL> logger . warning ( "<STR_LIT>" , metric_task , e ) <EOL> continue <EOL> def analyze_metrics ( metric_task , metrics ) : <EOL> if '<STR_LIT>' not in metrics : <EOL> return <EOL> if metric_task . job . cluster . service . name == '<STR_LIT>' : <EOL> start_time = time . time ( ) <EOL> if metric_task . job . name == '<STR_LIT>' : <EOL> analyze_hbase_master_metrics ( metric_task , metrics ) <EOL> elif metric_task . job . name == '<STR_LIT>' : <EOL> analyze_hbase_region_server_metrics ( metric_task , metrics ) <EOL> logger . info ( "<STR_LIT>" , <EOL> metric_task , time . time ( ) - start_time ) <EOL> def update_metrics_in_process ( output_queue , metric_task ) : <EOL> try : <EOL> logger . info ( "<STR_LIT>" , os . getpid ( ) ) <EOL> metricsRawData = metric_task . last_metrics_raw <EOL> start_time = time . time ( ) <EOL> if metric_task . need_analyze : <EOL> if metricsRawData : <EOL> metrics = json . loads ( metricsRawData ) <EOL> metrics_saved = { } <EOL> for bean_output in metrics [ "<STR_LIT>" ] : <EOL> bean_name = bean_output [ "<STR_LIT:name>" ] <EOL> for metric_name , metric_value in bean_output . iteritems ( ) : <EOL> if metric_name in [ "<STR_LIT:name>" , "<STR_LIT>" ] : <EOL> continue <EOL> metric_type = type ( metric_value ) <EOL> if metric_name in BOOL_METRIC_MAP : <EOL> metric_value = int ( metric_value == BOOL_METRIC_MAP [ metric_name ] ) <EOL> elif metric_type is list or metric_type is dict : <EOL> metric_value = len ( metric_value ) <EOL> elif metric_type is bool : <EOL> metric_value = int ( metric_value ) <EOL> elif metric_value is None : <EOL> metric_value = <NUM_LIT:0> <EOL> elif not ( metric_type is int or metric_type is float <EOL> or metric_type is unicode or metric_type is str ) : <EOL> logger . warning ( "<STR_LIT>" , <EOL> bean_name , metric_name , metric_type , metric_value ) <EOL> continue <EOL> group = metrics_saved . setdefault ( bean_name , { } ) <EOL> group [ metric_name ] = metric_value <EOL> metric_task . last_metrics = json . dumps ( metrics_saved ) <EOL> analyze_metrics ( metric_task , metrics ) <EOL> metric_task . save ( ) <EOL> logger . info ( "<STR_LIT>" , <EOL> metric_task , time . time ( ) - start_time ) <EOL> output_queue . put ( QueueTask ( METRIC_TASK_TYPE , metric_task . metric_source_id ) ) <EOL> except Exception , e : <EOL> logger . warning ( "<STR_LIT>" , metric_task , e ) <EOL> traceback . print_exc ( ) </s>
