<s> """<STR_LIT>""" <EOL> import logging <EOL> import requests <EOL> import time <EOL> from newrelic_plugin_agent . plugins import base <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> class RabbitMQ ( base . Plugin ) : <EOL> GUID = '<STR_LIT>' <EOL> DEFAULT_USER = '<STR_LIT>' <EOL> DEFAULT_PASSWORD = '<STR_LIT>' <EOL> DEFAULT_HOST = '<STR_LIT:localhost>' <EOL> DEFAULT_PORT = <NUM_LIT> <EOL> DEFAULT_API_PATH = '<STR_LIT>' <EOL> DUMMY_STATS = { '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : <NUM_LIT:0> } <EOL> def add_node_datapoints ( self , node_data , queue_data , channel_data ) : <EOL> """<STR_LIT>""" <EOL> channels = <NUM_LIT:0> <EOL> for node in node_data : <EOL> name = node [ '<STR_LIT:name>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] <EOL> self . add_node_channel_datapoints ( name , channel_data ) <EOL> self . add_node_message_datapoints ( name , queue_data , channel_data ) <EOL> self . add_node_queue_datapoints ( name , queue_data ) <EOL> count = <NUM_LIT:0> <EOL> for channel in channel_data : <EOL> if channel [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == name : <EOL> count += <NUM_LIT:1> <EOL> channels += count <EOL> base_name = '<STR_LIT>' % name <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , <EOL> '<STR_LIT>' , count ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> node . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> node . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> node . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> node . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' , '<STR_LIT>' , channels ) <EOL> self . add_gauge_value ( '<STR_LIT>' , '<STR_LIT>' , self . consumers ) <EOL> def add_node_channel_datapoints ( self , node , channel_data ) : <EOL> """<STR_LIT>""" <EOL> channel_flow_blocked = <NUM_LIT:0> <EOL> for channel in channel_data : <EOL> if channel [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == node : <EOL> if channel . get ( '<STR_LIT>' ) : <EOL> channel_flow_blocked += <NUM_LIT:1> <EOL> self . add_gauge_value ( '<STR_LIT>' % node , '<STR_LIT>' , <EOL> channel_flow_blocked ) <EOL> def add_node_message_datapoints ( self , node , queue_data , channel_data ) : <EOL> """<STR_LIT>""" <EOL> base_name = '<STR_LIT>' % node <EOL> keys = self . DUMMY_STATS . keys ( ) <EOL> count , total , min_val , max_val , values = self . initialize_counters ( keys ) <EOL> for channel in channel_data : <EOL> if channel [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == node : <EOL> for key in keys : <EOL> total [ key ] += channel . get ( key , <NUM_LIT:0> ) <EOL> count , total , min_val , max_val , values = self . initialize_counters ( keys ) <EOL> message_stats = list ( ) <EOL> for channel in channel_data : <EOL> if channel [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == node : <EOL> stats = channel . get ( '<STR_LIT>' ) <EOL> if stats : <EOL> message_stats . append ( stats ) <EOL> for stat_block in message_stats : <EOL> for key in keys : <EOL> total [ key ] += stat_block . get ( key , <NUM_LIT:0> ) <EOL> for key in keys : <EOL> name = key <EOL> if key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> elif key == '<STR_LIT>' : <EOL> name = '<STR_LIT>' <EOL> self . add_derive_value ( '<STR_LIT>' % ( base_name , name ) , <EOL> '<STR_LIT>' , <EOL> total [ key ] ) <EOL> keys = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> count , total , min_val , max_val , values = self . initialize_counters ( keys ) <EOL> for queue in queue_data : <EOL> if queue [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == node : <EOL> for key in keys : <EOL> total [ key ] += queue . get ( key , <NUM_LIT:0> ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> total [ '<STR_LIT>' ] ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , <EOL> '<STR_LIT>' , <EOL> total [ '<STR_LIT>' ] ) <EOL> def add_node_queue_datapoints ( self , node , queue_data ) : <EOL> """<STR_LIT>""" <EOL> keys = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> count , total , min_val , max_val , values = self . initialize_counters ( keys ) <EOL> del keys [ <NUM_LIT:2> ] <EOL> for queue in queue_data : <EOL> if queue [ '<STR_LIT>' ] . split ( '<STR_LIT:@>' ) [ - <NUM_LIT:1> ] == node : <EOL> for key in keys : <EOL> count [ key ] += <NUM_LIT:1> <EOL> value = queue . get ( key , <NUM_LIT:0> ) <EOL> total [ key ] += value <EOL> values [ key ] . append ( value ) <EOL> key = '<STR_LIT>' <EOL> count [ key ] += count [ '<STR_LIT>' ] <EOL> idle_count = total [ '<STR_LIT>' ] - total [ '<STR_LIT>' ] <EOL> total [ key ] += idle_count <EOL> values [ key ] . append ( idle_count ) <EOL> base_name = '<STR_LIT>' % node <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> total [ '<STR_LIT>' ] , <EOL> None , <EOL> None , <EOL> count [ '<STR_LIT>' ] ) <EOL> self . consumers += total [ '<STR_LIT>' ] <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> total [ '<STR_LIT>' ] , <EOL> None , <EOL> None , <EOL> count [ '<STR_LIT>' ] ) <EOL> base_name = '<STR_LIT>' % node <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> total [ '<STR_LIT>' ] , <EOL> None , <EOL> None , <EOL> count [ '<STR_LIT>' ] ) <EOL> def track_vhost_queue ( self , vhost_name , queue_name ) : <EOL> """<STR_LIT>""" <EOL> TRACK_EVERYTHING = dict ( ) <EOL> tracked_vhosts = self . config . get ( '<STR_LIT>' , TRACK_EVERYTHING ) <EOL> vhost_settings = tracked_vhosts . get ( vhost_name ) or { } <EOL> vhost_queues = vhost_settings . get ( '<STR_LIT>' , [ ] ) <EOL> if tracked_vhosts is TRACK_EVERYTHING : <EOL> return True <EOL> if vhost_name in tracked_vhosts and vhost_queues == [ ] : <EOL> return True <EOL> return queue_name in vhost_queues <EOL> def add_queue_datapoints ( self , queue_data ) : <EOL> """<STR_LIT>""" <EOL> count = <NUM_LIT:0> <EOL> available , consumers , deliver , publish , redeliver , unacked = <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> <EOL> for count , queue in enumerate ( queue_data ) : <EOL> if queue [ '<STR_LIT:name>' ] [ <NUM_LIT:0> : <NUM_LIT:7> ] == '<STR_LIT>' : <EOL> LOGGER . debug ( '<STR_LIT>' , queue [ '<STR_LIT:name>' ] ) <EOL> continue <EOL> message_stats = queue . get ( '<STR_LIT>' , dict ( ) ) <EOL> if not message_stats : <EOL> message_stats = self . DUMMY_STATS <EOL> vhost = '<STR_LIT>' if queue [ '<STR_LIT>' ] == '<STR_LIT:/>' else queue [ '<STR_LIT>' ] <EOL> base_name = '<STR_LIT>' % ( vhost , queue [ '<STR_LIT:name>' ] ) <EOL> if not self . track_vhost_queue ( vhost , queue [ '<STR_LIT:name>' ] ) : <EOL> continue <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> queue . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> base_name = '<STR_LIT>' % ( vhost , queue [ '<STR_LIT:name>' ] ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_derive_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> queue . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> self . add_gauge_value ( '<STR_LIT>' % base_name , '<STR_LIT>' , <EOL> queue . get ( '<STR_LIT>' , <NUM_LIT:0> ) ) <EOL> available += queue . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> deliver += message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> publish += message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> redeliver += message_stats . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> unacked += queue . get ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . add_derive_value ( '<STR_LIT>' , '<STR_LIT>' , <EOL> deliver , count = count ) <EOL> self . add_derive_value ( '<STR_LIT>' , '<STR_LIT>' , <EOL> publish , count = count ) <EOL> self . add_derive_value ( '<STR_LIT>' , '<STR_LIT>' , <EOL> redeliver , count = count ) <EOL> self . add_gauge_value ( '<STR_LIT>' , '<STR_LIT>' , <EOL> available , count = count ) <EOL> self . add_gauge_value ( '<STR_LIT>' , '<STR_LIT>' , <EOL> unacked , count = count ) <EOL> def http_get ( self , url , params = None ) : <EOL> """<STR_LIT>""" <EOL> kwargs = { '<STR_LIT:url>' : url , <EOL> '<STR_LIT>' : ( self . config . get ( '<STR_LIT:username>' , self . DEFAULT_USER ) , <EOL> self . config . get ( '<STR_LIT:password>' , self . DEFAULT_PASSWORD ) ) , <EOL> '<STR_LIT>' : self . config . get ( '<STR_LIT>' , True ) } <EOL> if params : <EOL> kwargs [ '<STR_LIT>' ] = params <EOL> try : <EOL> return self . requests_session . get ( ** kwargs ) <EOL> except requests . ConnectionError as error : <EOL> LOGGER . error ( '<STR_LIT>' , url , error ) <EOL> return None <EOL> def fetch_data ( self , data_type , columns = None ) : <EOL> """<STR_LIT>""" <EOL> url = '<STR_LIT>' % ( self . rabbitmq_base_url , data_type ) <EOL> params = { '<STR_LIT>' : '<STR_LIT:U+002C>' . join ( columns ) } if columns else { } <EOL> response = self . http_get ( url , params ) <EOL> if not response or response . status_code != <NUM_LIT:200> : <EOL> if response : <EOL> LOGGER . error ( '<STR_LIT>' , url , <EOL> response . status_code , response . content ) <EOL> return list ( ) <EOL> try : <EOL> return response . json ( ) <EOL> except Exception as error : <EOL> LOGGER . error ( '<STR_LIT>' , error ) <EOL> return list ( ) <EOL> def fetch_channel_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . fetch_data ( '<STR_LIT>' ) <EOL> def fetch_node_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . fetch_data ( '<STR_LIT>' ) <EOL> def fetch_queue_data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . fetch_data ( '<STR_LIT>' ) <EOL> def poll ( self ) : <EOL> """<STR_LIT>""" <EOL> LOGGER . info ( '<STR_LIT>' , self . rabbitmq_base_url ) <EOL> start_time = time . time ( ) <EOL> self . requests_session = requests . Session ( ) <EOL> self . derive = dict ( ) <EOL> self . gauge = dict ( ) <EOL> self . rate = dict ( ) <EOL> self . consumers = <NUM_LIT:0> <EOL> channel_data = self . fetch_channel_data ( ) <EOL> node_data = self . fetch_node_data ( ) <EOL> queue_data = self . fetch_queue_data ( ) <EOL> self . add_queue_datapoints ( queue_data ) <EOL> self . add_node_datapoints ( node_data , queue_data , channel_data ) <EOL> LOGGER . info ( '<STR_LIT>' , <EOL> time . time ( ) - start_time ) <EOL> @ property <EOL> def rabbitmq_base_url ( self ) : <EOL> """<STR_LIT>""" <EOL> port = self . config . get ( '<STR_LIT:port>' , self . DEFAULT_PORT ) <EOL> secure = self . config . get ( '<STR_LIT>' , False ) <EOL> host = self . config . get ( '<STR_LIT:host>' , self . DEFAULT_HOST ) <EOL> api_path = self . config . get ( '<STR_LIT>' , self . DEFAULT_API_PATH ) <EOL> scheme = '<STR_LIT>' if secure else '<STR_LIT:http>' <EOL> return '<STR_LIT>' . format ( <EOL> scheme = scheme , host = host , port = port , api_path = api_path ) </s>
