<s> from base_monitor import BaseMonitor <EOL> from kafka import KafkaClient , SimpleProducer <EOL> from kafka . common import KafkaUnavailableError <EOL> from scutils . method_timer import MethodTimer <EOL> import json <EOL> import sys <EOL> class KafkaBaseMonitor ( BaseMonitor ) : <EOL> '''<STR_LIT>''' <EOL> def setup ( self , settings ) : <EOL> '''<STR_LIT>''' <EOL> @ MethodTimer . timeout ( settings [ '<STR_LIT>' ] , False ) <EOL> def _hidden_setup ( ) : <EOL> try : <EOL> self . kafka_conn = KafkaClient ( settings [ '<STR_LIT>' ] ) <EOL> self . producer = SimpleProducer ( self . kafka_conn ) <EOL> self . topic_prefix = settings [ '<STR_LIT>' ] <EOL> except KafkaUnavailableError as ex : <EOL> message = "<STR_LIT>" "<STR_LIT>" . format ( type ( ex ) . __name__ , ex . args ) <EOL> self . logger . error ( message ) <EOL> return False <EOL> return True <EOL> ret_val = _hidden_setup ( ) <EOL> self . use_appid_topics = settings [ '<STR_LIT>' ] <EOL> if ret_val : <EOL> self . logger . debug ( "<STR_LIT>" <EOL> . format ( name = self . __class__ . __name__ ) ) <EOL> else : <EOL> self . logger . error ( "<STR_LIT>" <EOL> "<STR_LIT>" . format ( name = self . __class__ . __name__ ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> def _send_to_kafka ( self , master ) : <EOL> '''<STR_LIT>''' <EOL> appid_topic = "<STR_LIT>" . format ( <EOL> prefix = self . topic_prefix , <EOL> appid = master [ '<STR_LIT>' ] ) <EOL> firehose_topic = "<STR_LIT>" . format ( <EOL> prefix = self . topic_prefix ) <EOL> try : <EOL> self . kafka_conn . ensure_topic_exists ( firehose_topic ) <EOL> dump = json . dumps ( master ) <EOL> if self . use_appid_topics : <EOL> self . kafka_conn . ensure_topic_exists ( appid_topic ) <EOL> self . producer . send_messages ( appid_topic , dump ) <EOL> self . producer . send_messages ( firehose_topic , dump ) <EOL> return True <EOL> except Exception as ex : <EOL> message = "<STR_LIT>" "<STR_LIT>" . format ( type ( ex ) . __name__ , ex . args ) <EOL> self . logger . error ( message ) <EOL> return False </s>
