<s> """<STR_LIT>""" <EOL> import urllib2 <EOL> import re <EOL> from diamond . collector import str_to_bool <EOL> try : <EOL> import json <EOL> except ImportError : <EOL> import simplejson as json <EOL> import diamond . collector <EOL> RE_LOGSTASH_INDEX = re . compile ( '<STR_LIT>' ) <EOL> class ElasticSearchCollector ( diamond . collector . Collector ) : <EOL> def process_config ( self ) : <EOL> super ( ElasticSearchCollector , self ) . process_config ( ) <EOL> instance_list = self . config [ '<STR_LIT>' ] <EOL> if isinstance ( instance_list , basestring ) : <EOL> instance_list = [ instance_list ] <EOL> if len ( instance_list ) == <NUM_LIT:0> : <EOL> host = self . config [ '<STR_LIT:host>' ] <EOL> port = self . config [ '<STR_LIT:port>' ] <EOL> instance_list . append ( '<STR_LIT>' % ( host , port ) ) <EOL> self . instances = { } <EOL> for instance in instance_list : <EOL> if '<STR_LIT:@>' in instance : <EOL> ( alias , hostport ) = instance . split ( '<STR_LIT:@>' , <NUM_LIT:1> ) <EOL> else : <EOL> alias = '<STR_LIT:default>' <EOL> hostport = instance <EOL> if '<STR_LIT::>' in hostport : <EOL> host , port = hostport . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> else : <EOL> host = hostport <EOL> port = <NUM_LIT> <EOL> self . instances [ alias ] = ( host , int ( port ) ) <EOL> def get_default_config_help ( self ) : <EOL> config_help = super ( ElasticSearchCollector , <EOL> self ) . get_default_config_help ( ) <EOL> config_help . update ( { <EOL> '<STR_LIT:host>' : "<STR_LIT>" , <EOL> '<STR_LIT:port>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> + "<STR_LIT>" <EOL> + "<STR_LIT>" <EOL> + "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> + "<STR_LIT>" <EOL> + "<STR_LIT>" <EOL> + "<STR_LIT>" , <EOL> } ) <EOL> return config_help <EOL> def get_default_config ( self ) : <EOL> """<STR_LIT>""" <EOL> config = super ( ElasticSearchCollector , self ) . get_default_config ( ) <EOL> config . update ( { <EOL> '<STR_LIT:host>' : '<STR_LIT:127.0.0.1>' , <EOL> '<STR_LIT:port>' : <NUM_LIT> , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT:path>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : False , <EOL> } ) <EOL> return config <EOL> def _get ( self , host , port , path , assert_key = None ) : <EOL> """<STR_LIT>""" <EOL> url = '<STR_LIT>' % ( host , port , path ) <EOL> try : <EOL> response = urllib2 . urlopen ( url ) <EOL> except Exception , err : <EOL> self . log . error ( "<STR_LIT>" , url , err ) <EOL> return False <EOL> try : <EOL> doc = json . load ( response ) <EOL> except ( TypeError , ValueError ) : <EOL> self . log . error ( "<STR_LIT>" <EOL> + "<STR_LIT>" ) <EOL> return False <EOL> if assert_key and not assert_key in doc : <EOL> self . log . error ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( assert_key , url ) ) <EOL> return False <EOL> return doc <EOL> def _copy_one_level ( self , metrics , prefix , data , filter = lambda key : True ) : <EOL> for key , value in data . iteritems ( ) : <EOL> if filter ( key ) : <EOL> metric_path = '<STR_LIT>' % ( prefix , key ) <EOL> self . _set_or_sum_metric ( metrics , metric_path , value ) <EOL> def _copy_two_level ( self , metrics , prefix , data , filter = lambda key : True ) : <EOL> for key1 , d1 in data . iteritems ( ) : <EOL> self . _copy_one_level ( metrics , '<STR_LIT>' % ( prefix , key1 ) , d1 , filter ) <EOL> def _index_metrics ( self , metrics , prefix , index ) : <EOL> if self . config [ '<STR_LIT>' ] : <EOL> """<STR_LIT>""" <EOL> m = RE_LOGSTASH_INDEX . match ( prefix ) <EOL> if m : <EOL> prefix = m . group ( <NUM_LIT:1> ) <EOL> self . _set_or_sum_metric ( metrics , <EOL> '<STR_LIT>' % prefix , <NUM_LIT:1> ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' % prefix , index , <EOL> [ '<STR_LIT>' , '<STR_LIT:count>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' % prefix , index , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' % prefix , index , <EOL> [ '<STR_LIT:store>' , '<STR_LIT>' ] ) <EOL> self . _copy_two_level ( <EOL> metrics , prefix , index , <EOL> lambda key : key . endswith ( '<STR_LIT>' ) or key . endswith ( '<STR_LIT>' ) ) <EOL> def _add_metric ( self , metrics , metric_path , data , data_path ) : <EOL> """<STR_LIT>""" <EOL> current_item = data <EOL> for path_element in data_path : <EOL> current_item = current_item . get ( path_element ) <EOL> if current_item is None : <EOL> return <EOL> self . _set_or_sum_metric ( metrics , metric_path , current_item ) <EOL> def _set_or_sum_metric ( self , metrics , metric_path , value ) : <EOL> """<STR_LIT>""" <EOL> if metric_path in metrics : <EOL> metrics [ metric_path ] += value <EOL> else : <EOL> metrics [ metric_path ] = value <EOL> def collect_instance_cluster_stats ( self , host , port , metrics ) : <EOL> result = self . _get ( host , port , '<STR_LIT>' ) <EOL> if not result : <EOL> return <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , <EOL> result , [ '<STR_LIT>' ] ) <EOL> def collect_instance_index_stats ( self , host , port , metrics ) : <EOL> result = self . _get ( host , port , <EOL> '<STR_LIT>' <EOL> + '<STR_LIT>' , '<STR_LIT>' ) <EOL> if not result : <EOL> return <EOL> _all = result [ '<STR_LIT>' ] <EOL> self . _index_metrics ( metrics , '<STR_LIT>' , _all [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in _all : <EOL> indices = _all [ '<STR_LIT>' ] <EOL> elif '<STR_LIT>' in result : <EOL> indices = result [ '<STR_LIT>' ] <EOL> else : <EOL> return <EOL> for name , index in indices . iteritems ( ) : <EOL> self . _index_metrics ( metrics , '<STR_LIT>' % name , <EOL> index [ '<STR_LIT>' ] ) <EOL> def collect_instance ( self , alias , host , port ) : <EOL> result = self . _get ( host , port , '<STR_LIT>' , '<STR_LIT>' ) <EOL> if not result : <EOL> return <EOL> metrics = { } <EOL> node = result [ '<STR_LIT>' ] . keys ( ) [ <NUM_LIT:0> ] <EOL> data = result [ '<STR_LIT>' ] [ node ] <EOL> metrics [ '<STR_LIT>' ] = data [ '<STR_LIT:http>' ] [ '<STR_LIT>' ] <EOL> indices = data [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = indices [ '<STR_LIT>' ] [ '<STR_LIT:count>' ] <EOL> metrics [ '<STR_LIT>' ] = indices [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = indices [ '<STR_LIT:store>' ] [ '<STR_LIT>' ] <EOL> transport = data [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = transport [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = transport [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = transport [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = transport [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in indices : <EOL> cache = indices [ '<STR_LIT>' ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , <EOL> [ '<STR_LIT>' ] ) <EOL> metrics [ '<STR_LIT>' ] = cache [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = cache [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = cache [ '<STR_LIT>' ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , <EOL> [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in indices : <EOL> cache = indices [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = cache [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = cache [ '<STR_LIT>' ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , [ '<STR_LIT:count>' ] ) <EOL> if '<STR_LIT>' in indices : <EOL> cache = indices [ '<STR_LIT>' ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , cache , <EOL> [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in indices : <EOL> fielddata = indices [ '<STR_LIT>' ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fielddata , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fielddata , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , data , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , data , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , data , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , data , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in data and '<STR_LIT:data>' in data [ '<STR_LIT>' ] and data [ '<STR_LIT>' ] [ '<STR_LIT:data>' ] : <EOL> fs_data = data [ '<STR_LIT>' ] [ '<STR_LIT:data>' ] [ <NUM_LIT:0> ] <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fs_data , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fs_data , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fs_data , <EOL> [ '<STR_LIT>' ] ) <EOL> self . _add_metric ( metrics , '<STR_LIT>' , fs_data , <EOL> [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in self . config [ '<STR_LIT>' ] : <EOL> jvm = data [ '<STR_LIT>' ] <EOL> mem = jvm [ '<STR_LIT>' ] <EOL> for k in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> metrics [ '<STR_LIT>' % k ] = mem [ '<STR_LIT>' % k ] <EOL> if '<STR_LIT>' in mem : <EOL> metrics [ '<STR_LIT>' ] = mem [ '<STR_LIT>' ] <EOL> for pool , d in mem [ '<STR_LIT>' ] . iteritems ( ) : <EOL> pool = pool . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:_>' ) <EOL> metrics [ '<STR_LIT>' % pool ] = d [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' % pool ] = d [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' ] = jvm [ '<STR_LIT>' ] [ '<STR_LIT:count>' ] <EOL> gc = jvm [ '<STR_LIT>' ] <EOL> collection_count = <NUM_LIT:0> <EOL> collection_time_in_millis = <NUM_LIT:0> <EOL> for collector , d in gc [ '<STR_LIT>' ] . iteritems ( ) : <EOL> metrics [ '<STR_LIT>' % collector ] = d [ <EOL> '<STR_LIT>' ] <EOL> collection_count += d [ '<STR_LIT>' ] <EOL> metrics [ '<STR_LIT>' % collector ] = d [ <EOL> '<STR_LIT>' ] <EOL> collection_time_in_millis += d [ '<STR_LIT>' ] <EOL> if '<STR_LIT>' in gc : <EOL> metrics [ '<STR_LIT>' ] = gc [ '<STR_LIT>' ] <EOL> else : <EOL> metrics [ '<STR_LIT>' ] = collection_count <EOL> k = '<STR_LIT>' <EOL> if k in gc : <EOL> metrics [ '<STR_LIT>' ] = gc [ k ] <EOL> else : <EOL> metrics [ '<STR_LIT>' ] = collection_time_in_millis <EOL> if '<STR_LIT>' in self . config [ '<STR_LIT>' ] : <EOL> self . _copy_two_level ( metrics , '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> self . _copy_two_level ( metrics , '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> if str_to_bool ( self . config [ '<STR_LIT>' ] ) : <EOL> self . collect_instance_cluster_stats ( host , port , metrics ) <EOL> if '<STR_LIT>' in self . config [ '<STR_LIT>' ] : <EOL> self . collect_instance_index_stats ( host , port , metrics ) <EOL> for key in metrics : <EOL> full_key = key <EOL> if alias != '<STR_LIT>' : <EOL> full_key = '<STR_LIT>' % ( alias , full_key ) <EOL> self . publish ( full_key , metrics [ key ] ) <EOL> def collect ( self ) : <EOL> if json is None : <EOL> self . log . error ( '<STR_LIT>' ) <EOL> return { } <EOL> for alias in sorted ( self . instances ) : <EOL> ( host , port ) = self . instances [ alias ] <EOL> self . collect_instance ( alias , host , port ) </s>
