EVOLUTION-NINJA

Edit File: 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