<s> import os <EOL> import logging <EOL> import tempfile <EOL> import gzip <EOL> import shutil <EOL> import time <EOL> import csv <EOL> import copy <EOL> from mtgraphite import MTGraphiteClient <EOL> import json <EOL> try : <EOL> import kafka as kafka_python <EOL> import pykafka <EOL> except ImportError : <EOL> kafka_python = None <EOL> pykafka = None <EOL> import requests <EOL> from features import ( OSFeature , FileFeature , ConfigFeature , DiskFeature , <EOL> ProcessFeature , MetricFeature , ConnectionFeature , <EOL> PackageFeature , MemoryFeature , CpuFeature , <EOL> InterfaceFeature , LoadFeature , DockerPSFeature , <EOL> DockerHistoryFeature ) <EOL> from misc import NullHandler <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class Emitter : <EOL> """<STR_LIT>""" <EOL> mtgclient = None <EOL> def __init__ ( <EOL> self , <EOL> urls , <EOL> emitter_args = { } , <EOL> format = '<STR_LIT>' , <EOL> max_emit_retries = <NUM_LIT:9> , <EOL> ) : <EOL> self . urls = urls <EOL> self . emitter_args = emitter_args <EOL> self . compress = emitter_args . get ( '<STR_LIT>' , False ) <EOL> self . format = format <EOL> self . max_emit_retries = max_emit_retries <EOL> self . mtgclient = None <EOL> def __enter__ ( self ) : <EOL> ( self . temp_fd , self . temp_fpath ) = tempfile . mkstemp ( prefix = '<STR_LIT>' ) <EOL> os . close ( self . temp_fd ) <EOL> if self . compress : <EOL> self . emitfile = gzip . open ( self . temp_fpath , '<STR_LIT:wb>' ) <EOL> else : <EOL> self . emitfile = open ( self . temp_fpath , '<STR_LIT:wb>' ) <EOL> self . csv_writer = csv . writer ( self . emitfile , delimiter = '<STR_LIT:\t>' , <EOL> quotechar = "<STR_LIT:'>" ) <EOL> self . begin_time = time . time ( ) <EOL> self . num_features = <NUM_LIT:0> <EOL> return self <EOL> def _get_feature_type ( self , feature ) : <EOL> if isinstance ( feature , OSFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , FileFeature ) : <EOL> return '<STR_LIT:file>' <EOL> if isinstance ( feature , ConfigFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , DiskFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , ProcessFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , ConnectionFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , MetricFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , PackageFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , MemoryFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , CpuFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , InterfaceFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , LoadFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , DockerPSFeature ) : <EOL> return '<STR_LIT>' <EOL> if isinstance ( feature , DockerHistoryFeature ) : <EOL> return '<STR_LIT>' <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def emit_dict_as_graphite ( <EOL> self , <EOL> sysname , <EOL> group , <EOL> suffix , <EOL> data , <EOL> timestamp = None , <EOL> ) : <EOL> timestamp = int ( timestamp or time . time ( ) ) <EOL> try : <EOL> items = data . items ( ) <EOL> except : <EOL> return <EOL> sysname = sysname . replace ( '<STR_LIT:/>' , '<STR_LIT:.>' ) <EOL> for ( metric , value ) in items : <EOL> try : <EOL> value = float ( value ) <EOL> except Exception : <EOL> continue <EOL> metric = metric . replace ( '<STR_LIT:(>' , '<STR_LIT:_>' ) . replace ( '<STR_LIT:)>' , '<STR_LIT>' ) <EOL> metric = metric . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:_>' ) . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) <EOL> metric = metric . replace ( '<STR_LIT:/>' , '<STR_LIT:_>' ) . replace ( '<STR_LIT:\\>' , '<STR_LIT:_>' ) <EOL> suffix = suffix . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) <EOL> if '<STR_LIT>' in suffix or '<STR_LIT>' in suffix : <EOL> metric = metric . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) <EOL> if '<STR_LIT>' in metric : <EOL> metric = metric . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> metric = metric . replace ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> if suffix == '<STR_LIT>' : <EOL> suffix = '<STR_LIT>' <EOL> suffix = suffix . replace ( '<STR_LIT:/>' , '<STR_LIT:$>' ) <EOL> tmp_message = '<STR_LIT>' % ( sysname , suffix , <EOL> metric , value , timestamp ) <EOL> self . emitfile . write ( tmp_message ) <EOL> return <EOL> def emit ( <EOL> self , <EOL> feature_key , <EOL> feature_val , <EOL> feature_type = None , <EOL> ) : <EOL> if self . num_features == <NUM_LIT:0> : <EOL> try : <EOL> metadata = copy . deepcopy ( self . emitter_args ) <EOL> metadata [ '<STR_LIT>' ] = time . strftime ( '<STR_LIT>' ) <EOL> if '<STR_LIT>' in metadata : <EOL> del metadata [ '<STR_LIT>' ] <EOL> if self . emitter_args [ '<STR_LIT>' ] : <EOL> metadata . update ( json . loads ( self . emitter_args [ '<STR_LIT>' <EOL> ] ) ) <EOL> if '<STR_LIT>' in metadata : <EOL> del metadata [ '<STR_LIT>' ] <EOL> if self . format == '<STR_LIT>' : <EOL> self . csv_writer . writerow ( <EOL> [ '<STR_LIT>' , <EOL> json . dumps ( '<STR_LIT>' ) , <EOL> json . dumps ( metadata , <EOL> separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) ) ] ) <EOL> self . num_features += <NUM_LIT:1> <EOL> except Exception as e : <EOL> logger . exception ( e ) <EOL> raise <EOL> feature_type = feature_type or self . _get_feature_type ( feature_val ) <EOL> if isinstance ( feature_val , dict ) : <EOL> feature_val_as_dict = feature_val <EOL> else : <EOL> feature_val_as_dict = feature_val . _asdict ( ) <EOL> if '<STR_LIT>' in self . emitter_args and self . emitter_args [ '<STR_LIT>' ] and '<STR_LIT>' in self . emitter_args and self . emitter_args [ '<STR_LIT>' ] == True : <EOL> feature_val_as_dict . update ( json . loads ( self . emitter_args [ '<STR_LIT>' <EOL> ] ) ) <EOL> try : <EOL> if self . format == '<STR_LIT>' : <EOL> self . csv_writer . writerow ( <EOL> [ feature_type , <EOL> json . dumps ( feature_key ) , <EOL> json . dumps ( feature_val_as_dict , <EOL> separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) ) ] ) <EOL> elif self . format == '<STR_LIT>' : <EOL> if '<STR_LIT>' in self . emitter_args : <EOL> namespace = self . emitter_args [ '<STR_LIT>' ] <EOL> else : <EOL> namespace = '<STR_LIT>' <EOL> self . emit_dict_as_graphite ( <EOL> namespace , <EOL> feature_type , <EOL> feature_key , <EOL> feature_val_as_dict ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> self . num_features += <NUM_LIT:1> <EOL> except Exception as e : <EOL> logger . exception ( e ) <EOL> raise <EOL> def _close_file ( self ) : <EOL> self . emitfile . close ( ) <EOL> def _publish_to_stdout ( self ) : <EOL> with open ( self . temp_fpath , '<STR_LIT:r>' ) as fd : <EOL> if self . compress : <EOL> print fd . read ( ) <EOL> else : <EOL> for line in fd . readlines ( ) : <EOL> print line . strip ( ) <EOL> def _publish_to_broker ( self , url , max_emit_retries = <NUM_LIT:5> ) : <EOL> broker_alive = False <EOL> retries = <NUM_LIT:0> <EOL> while not broker_alive and retries <= max_emit_retries : <EOL> try : <EOL> retries += <NUM_LIT:1> <EOL> requests . get ( url ) <EOL> broker_alive = True <EOL> except Exception : <EOL> if retries <= max_emit_retries : <EOL> wait_time = <NUM_LIT> ** retries * <NUM_LIT:0.1> <EOL> logger . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( url , wait_time ) ) <EOL> time . sleep ( wait_time ) <EOL> else : <EOL> raise <EOL> with open ( self . temp_fpath , '<STR_LIT:rb>' ) as framefp : <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> if self . compress : <EOL> headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> try : <EOL> requests . post ( url , headers = headers , <EOL> params = self . emitter_args , data = framefp ) <EOL> except Exception : <EOL> logger . error ( '<STR_LIT>' <EOL> % url ) <EOL> raise <EOL> def _publish_to_kafka_no_retries ( self , url ) : <EOL> if kafka_python is None or pykafka is None : <EOL> raise ImportError ( '<STR_LIT>' ) <EOL> try : <EOL> list = url [ len ( '<STR_LIT>' ) : ] . split ( '<STR_LIT:/>' ) <EOL> if len ( list ) == <NUM_LIT:2> : <EOL> kurl , topic = list <EOL> else : <EOL> raise Exception ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % url ) <EOL> h = NullHandler ( ) <EOL> logging . getLogger ( '<STR_LIT>' ) . addHandler ( h ) <EOL> kafka_python_client = kafka_python . KafkaClient ( kurl ) <EOL> kafka_python_client . ensure_topic_exists ( topic ) <EOL> kafka = pykafka . KafkaClient ( hosts = kurl ) <EOL> publish_topic_object = kafka . topics [ topic ] <EOL> producer = publish_topic_object . get_producer ( ) <EOL> if self . format == '<STR_LIT>' : <EOL> with open ( self . temp_fpath , '<STR_LIT:r>' ) as fp : <EOL> text = fp . read ( ) <EOL> logger . debug ( producer . produce ( [ text ] ) ) <EOL> elif self . format == '<STR_LIT>' : <EOL> with open ( self . temp_fpath , '<STR_LIT:r>' ) as fp : <EOL> for line in fp . readlines ( ) : <EOL> producer . produce ( [ line ] ) <EOL> else : <EOL> logger . debug ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' . format ( self . format ) ) <EOL> raise <EOL> kafka_python_client . close ( ) <EOL> except Exception as e : <EOL> logger . debug ( '<STR_LIT>' . format ( url , <EOL> e ) ) <EOL> raise <EOL> def _publish_to_kafka ( self , url , max_emit_retries = <NUM_LIT:8> ) : <EOL> broker_alive = False <EOL> retries = <NUM_LIT:0> <EOL> while not broker_alive and retries <= max_emit_retries : <EOL> try : <EOL> retries += <NUM_LIT:1> <EOL> self . _publish_to_kafka_no_retries ( url ) <EOL> broker_alive = True <EOL> except Exception : <EOL> if retries <= max_emit_retries : <EOL> wait_time = <NUM_LIT> ** retries * <NUM_LIT:0.1> <EOL> logger . error ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % ( url , wait_time ) ) <EOL> time . sleep ( wait_time ) <EOL> else : <EOL> print '<STR_LIT>' % ( url , max_emit_retries ) <EOL> exit ( <NUM_LIT:1> ) <EOL> def _publish_to_mtgraphite ( self , url ) : <EOL> if not Emitter . mtgclient : <EOL> Emitter . mtgclient = MTGraphiteClient ( url ) <EOL> with open ( self . temp_fpath , '<STR_LIT:r>' ) as fp : <EOL> num_pushed_to_queue = Emitter . mtgclient . send_messages ( fp . readlines ( ) ) <EOL> logger . debug ( '<STR_LIT>' <EOL> % num_pushed_to_queue ) <EOL> def _write_to_file ( self , url ) : <EOL> output_path = url [ len ( '<STR_LIT>' ) : ] <EOL> if self . compress : <EOL> output_path += '<STR_LIT>' <EOL> shutil . move ( self . temp_fpath , output_path ) <EOL> def __exit__ ( <EOL> self , <EOL> typ , <EOL> exc , <EOL> trc , <EOL> ) : <EOL> if exc : <EOL> self . _close_file ( ) <EOL> if os . path . exists ( self . temp_fpath ) : <EOL> os . remove ( self . temp_fpath ) <EOL> return False <EOL> try : <EOL> self . _close_file ( ) <EOL> for url in self . urls : <EOL> logger . debug ( '<STR_LIT>' . format ( url ) ) <EOL> if url . startswith ( '<STR_LIT>' ) : <EOL> self . _publish_to_stdout ( ) <EOL> elif url . startswith ( '<STR_LIT>' ) : <EOL> self . _publish_to_broker ( url , self . max_emit_retries ) <EOL> elif url . startswith ( '<STR_LIT>' ) : <EOL> self . _write_to_file ( url ) <EOL> elif url . startswith ( '<STR_LIT>' ) : <EOL> self . _publish_to_kafka ( url , self . max_emit_retries ) <EOL> elif url . startswith ( '<STR_LIT>' ) : <EOL> self . _publish_to_mtgraphite ( url ) <EOL> else : <EOL> if os . path . exists ( self . temp_fpath ) : <EOL> os . remove ( self . temp_fpath ) <EOL> raise ValueError ( <EOL> '<STR_LIT>' . format ( url ) ) <EOL> except Exception as e : <EOL> logger . exception ( e ) <EOL> raise <EOL> finally : <EOL> if os . path . exists ( self . temp_fpath ) : <EOL> os . remove ( self . temp_fpath ) <EOL> self . end_time = time . time ( ) <EOL> elapsed_time = self . end_time - self . begin_time <EOL> logger . info ( <EOL> '<STR_LIT>' . format ( <EOL> self . num_features , <EOL> elapsed_time ) ) <EOL> return False </s>
