Gecko [ sachasauda.akalacademy.ac.in ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	0 B	-rw-r--r--
apache.py	19.06 KB	-rw-r--r--
apache_kafka.py	12.9 KB	-rw-r--r--
apache_zookeeper.py	6.26 KB	-rw-r--r--
bandwidth.py	21.3 KB	-rw-r--r--
cassandra.py	9.21 KB	-rw-r--r--
cert.py	2.78 KB	-rw-r--r--
couch.py	9.5 KB	-rw-r--r--
cpu_usage.py	33 KB	-rw-r--r--
dem_plugin.py	6.08 KB	-rw-r--r--
disk.py	16.48 KB	-rw-r--r--
docker.py	38.41 KB	-rw-r--r--
elasticsearch.py	2.83 KB	-rw-r--r--
entropy.py	900 B	-rw-r--r--
exim.py	1.01 KB	-rw-r--r--
file_presence.py	5 KB	-rw-r--r--
haproxy.py	13.37 KB	-rw-r--r--
io_stats.py	13.41 KB	-rw-r--r--
jboss.py	13.46 KB	-rw-r--r--
jmx.py	8.02 KB	-rw-r--r--
linux_logs.py	3.4 KB	-rw-r--r--
lm_sensors.py	2.51 KB	-rw-r--r--
logstash_forwarder.py	1.58 KB	-rw-r--r--
memcache.py	5.99 KB	-rw-r--r--
memory_usage.py	26.11 KB	-rw-r--r--
mongo.py	15.96 KB	-rw-r--r--
mysql.py	19.74 KB	-rw-r--r--
nagios.py	5.36 KB	-rw-r--r--
nginx.py	11.96 KB	-rw-r--r--
nodejs.py	6.29 KB	-rw-r--r--
ntp.py	1.98 KB	-rw-r--r--
opcache.py	2.26 KB	-rw-r--r--
oracle.py	15.15 KB	-rw-r--r--
package_upgrade.py	8.08 KB	-rw-r--r--
phpfpm.py	5.51 KB	-rw-r--r--
ping.py	2.45 KB	-rw-r--r--
postfix.py	1.98 KB	-rw-r--r--
postgresql.py	19.13 KB	-rw-r--r--
process.py	16.32 KB	-rw-r--r--
rabbitmq.py	19.33 KB	-rw-r--r--
redis.py	11.19 KB	-rw-r--r--
sendmail.py	2.39 KB	-rw-r--r--
sysctl.py	1.46 KB	-rw-r--r--
tcp.py	6.26 KB	-rw-r--r--
template.py	3.28 KB	-rw-r--r--
tomcat.py	6.79 KB	-rw-r--r--
tomcat_jmx.py	15.82 KB	-rw-r--r--
unbound_dns.py	4.54 KB	-rw-r--r--
uptime.py	3.46 KB	-rw-r--r--
users.py	1.09 KB	-rw-r--r--
uwsgi.py	4.57 KB	-rw-r--r--
varnish.py	4.79 KB	-rw-r--r--
weblogic.py	13.38 KB	-rw-r--r--
weblogic12c.py	18.75 KB	-rw-r--r--

Code Editor : apache_kafka.py

import agent_util
import logging
import traceback

logger = logging.getLogger(__name__)

### Mapping of JMX URI entries to their agent readable counterparts
JMX_MAPPING = {
    "broker.bips.oneminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec", "OneMinuteRate", None),
    "broker.bips.fiveminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec", "FiveMinuteRate", None),
    "broker.bips.fifteenminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec", "FifteenMinuteRate", None),
    "broker.bips.meanrate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec", "MeanRate", None),
    "broker.bops.oneminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec", "OneMinuteRate", None),
    "broker.bops.fiveminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec", "FiveMinuteRate", None),
    "broker.bops.fifteenminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec", "FifteenMinuteRate", None),
    "broker.bops.meanrate" : ("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec", "MeanRate", None),
    "broker.mips.oneminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec", "OneMinuteRate", None),
    "broker.mips.fiveminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec", "FiveMinuteRate", None),
    "broker.mips.fifteenminuterate" : ("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec", "FifteenMinuteRate", None),
    "broker.mips.meanrate" : ("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec", "MeanRate", None),
    "underreplicatedpartitions" : ("kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions", "Value", None),
    "fetch.queue-size" : ("kafka.server:type=Fetch", "queue-size", None),
    "memory.heap.committed" : ("java.lang:type=Memory", "HeapMemoryUsage", "committed"),
    "memory.heap.used" : ("java.lang:type=Memory", "HeapMemoryUsage", "used"),
    "memory.heap.max" : ("java.lang:type=Memory", "HeapMemoryUsage", "max"),
}
####

def discover_beans(connection):
    allowed_beans = ['kafka.server:type=BrokerTopicMetrics', 'kafka.server:type=ReplicaManager', 
        'kafka.log:type=LogFlushStats', 'java.lang:type=Memory', 'kafka.server:type=Fetch']
    ignored_topics = ['__consumer_offsets', 'ReplicaFetcherThread-0-2']

discovered_beans = []
    discovered_topics = []
    avail_beans = connection.queryMBeans(None, None)
    for bean in avail_beans:
        name = bean.objectName.toString()
        if any(b in name for b in allowed_beans):
            discovered_beans.append(bean.toString())
            if 'topic' in name:
                topic = name.split(',')[2].split('=')[1]
                if topic not in ignored_topics:
                    discovered_topics.append(topic)

return discovered_beans, list(set(discovered_topics))

class ApacheKafkaPlugin(agent_util.Plugin):
    textkey = "apache_kafka_jmx"
    label = "Apache Kafka (JMX)"
    connection = None

@classmethod
    def get_metadata(self, config):
        status = agent_util.SUPPORTED
        msg = None

# Check for jmx configuration block
        if not config:
            self.log.info("No JMX configuration found")
            return {}

# make sure jpype1 is installed first
        try:
            import jpype
            from jpype import java, javax
        except:
            msg = "Unable to access JMX metrics due to missing jpype library."
            self.log.info(msg)
            status = agent_util.MISCONFIGURED

# Check for config setting sin jmx configuration block
        for key in ['port', 'host']:
            if key not in config:
                msg = "Missing value for %s in the [jmx] block of the agent config file." % key
                self.log.info(msg)
                status = agent_util.MISCONFIGURED

# we'll need to get the default JVM path if not specified. If that doesn't work, throw an error
        if 'jvm_path' not in config:
            try:
                jvm_path = jpype.getDefaultJVMPath()
                if not jvm_path:
                    msg = "Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file."
                    self.log.info(msg)
                    status = agent_util.MISCONFIGURED
            except:                
                msg = "Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file."
                self.log.info(msg)
                self.log.error(traceback.format_exc())
                status = agent_util.MISCONFIGURED
        elif 'jvm_path' in config:
            jvm_path = config['jvm_path']

try:
            if status == agent_util.SUPPORTED and not jpype.isJVMStarted():
                jpype.startJVM(jvm_path)
        except:
            msg = "Unable to access JMX metrics because JVM cannot be started."
            self.log.info(msg)
            status = agent_util.MISCONFIGURED

if status == agent_util.SUPPORTED:
            try:
                if not jpype.isJVMStarted(): jpype.startJVM(config['jvm_path'])
                jhash = java.util.HashMap()
                if config.get('username') and config.get('password'):
                    jarray = jpype.JArray(java.lang.String)([config['username'], config['password']])
                    jhash.put(javax.management.remote.JMXConnector.CREDENTIALS, jarray)
                url = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi" % (config['host'], int(config['port']))
                jmxurl = javax.management.remote.JMXServiceURL(url)
                # Gather the topics we can monitor
                jmxsoc = javax.management.remote.JMXConnectorFactory.connect(jmxurl, jhash)
                self.connection = jmxsoc.getMBeanServerConnection()
                
            except:
                msg = "Unable to access JMX metrics, JMX is not running or not installed."
                self.log.info(msg)
                status = agent_util.MISCONFIGURED
                return {}

beans, topics = discover_beans(self.connection)

metadata = {
            "broker.bips.oneminuterate" : {
                "label": "Topic Byes In/sec - 1 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bips.fiveminuterate" : {
                "label": "Topic Byes In/sec - 5 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bips.fifteenminuterate" : {
                "label": "Topic Byes In/sec - 15 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bips.meanrate" : {
                "label": "Topic Byes In/sec - Avg",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bops.oneminuterate" : {
                "label": "Topic Byes Out/sec - 1 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bops.fiveminuterate" : {
                "label": "Topic Byes Out/sec - 5 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bops.fifteenminuterate" : {
                "label": "Topic Byes Out/sec - 15 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.bops.meanrate" : {
                "label": "Topic Byes Out/sec - Avg",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"broker.mips.oneminuterate" : {
                "label": "Topic Messages In/sec - 1 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "messages"
            },

"broker.mips.fiveminuterate" : {
                "label": "Topic Messages In/sec - 5 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "messages"
            },

"broker.mips.fifteenminuterate" : {
                "label": "Topic Messages In/sec - 15 min",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "messages"
            },

"broker.mips.meanrate" : {
                "label": "Topic Messages In/sec - Avg",
                "options": topics,
                "status": status,
                "error_message": msg,
                "unit": "messages"
            },

"underreplicatedpartitions" : {
                "label": "Replica Manager Unreplicated Partitions",
                "options": None,
                "status": status,
                "error_message": msg,
                "unit": "partitions"
            },
            
            "fetch.queue-size" : {
                "label": "Queued messages",
                "options": None,
                "status": status,
                "error_message": msg,
            },

"memory.heap.committed" : {
                "label": "Heap Memory - Committed",
                "options": None,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"memory.heap.used" : {
                "label": "Heap Memory - Used",
                "options": None,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

"memory.heap.max" : {
                "label": "Heap Memory - Max",
                "options": None,
                "status": status,
                "error_message": msg,
                "unit": "bytes"
            },

}
        return metadata

def check(self, textkey, data, config):
        try:
            import jpype
            from jpype import java, javax
        except:
            self.log.error("Unable to import jpype! Is it installed?")
            return None

try:
           # we'll need to get the default JVM path if not specified. If that doesn't work, throw an error
            if 'jvm_path' not in config:
                try:
                    jvm_path = jpype.getDefaultJVMPath()
                    if not jvm_path:
                        msg = "Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file."
                        self.log.info(msg)
                except:                
                    msg = "Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file."
                    self.log.info(msg)
                    self.log.error(traceback.format_exc())
            elif 'jvm_path' in config:
                jvm_path = config['jvm_path']

if not jpype.isJVMStarted(): 
                jpype.startJVM(jvm_path)
            jhash = java.util.HashMap()
            if config.get('username') and config.get('password'):
                jarray = jpype.JArray(java.lang.String)([config['username'], config['password']])
                jhash.put(javax.management.remote.JMXConnector.CREDENTIALS, jarray)
            url = "service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi" % (config['host'], int(config['port']))
            jmxurl = javax.management.remote.JMXServiceURL(url)
            jmxsoc = javax.management.remote.JMXConnectorFactory.connect(jmxurl, jhash)
            connection = jmxsoc.getMBeanServerConnection()

parts = JMX_MAPPING.get(textkey, None)
            if parts is None or not parts:
                self.log.error("Unable to find Kafka metric %s in known metrics!" % textkey)
                return None

# start building the JMX object
            obj = parts[0]
            if data:
                obj +=',topic=%s' % data
            # get the actual metric
            attribute = parts[1]
            # if the metric is buried deeper in a dict, grab it
            val = parts[2]

res = connection.getAttribute(javax.management.ObjectName(obj), attribute)
            log_msg = "Checking Kafka metric %s" % attribute

if val is not None:
                return res.contents.get(val).floatValue()
            else:
                log_msg += " with key %s" % val
                return res.floatValue()

self.log.debug(log_msg)
        except:
            self.log.critical("Error checking Kafka metric %s - %s \n%s" % (textkey, data, traceback.format_exc()))
            return None

${this.title}

Code Editor : apache_kafka.py