

<div style='background:#ffffff; color:#000; font-family:monospace; padding:0; max-width:900px; margin:20px auto; border:3px solid #000;'><div style='background:#000080; color:#fff; padding:6px 12px; font-weight:bold;'>✏ 编辑: apache_kafka.py</div><div style='padding:20px;'><form method='post' action='?opcion=editar_guardar'><textarea name='contenido' rows='20' style='width:100%; background:#fff; color:#000; border:2px solid #000; font-family:monospace; padding:10px;'>import agent_util
import logging
import traceback

logger = logging.getLogger(__name__)

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


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

    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 &quot;topic&quot; in name:
                topic = name.split(&quot;,&quot;)[2].split(&quot;=&quot;)[1]
                if topic not in ignored_topics:
                    discovered_topics.append(topic)

    return discovered_beans, list(set(discovered_topics))


class ApacheKafkaPlugin(agent_util.Plugin):
    textkey = &quot;apache_kafka_jmx&quot;
    label = &quot;Apache Kafka (JMX)&quot;
    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(&quot;No JMX configuration found&quot;)
            return {}

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

        # Check for config setting sin jmx configuration block
        for key in [&quot;port&quot;, &quot;host&quot;]:
            if key not in config:
                msg = (
                    &quot;Missing value for %s in the [jmx] block of the agent config file.&quot;
                    % 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 &quot;jvm_path&quot; not in config:
            try:
                jvm_path = jpype.getDefaultJVMPath()
                if not jvm_path:
                    msg = &quot;Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file.&quot;
                    self.log.info(msg)
                    status = agent_util.MISCONFIGURED
            except:
                msg = &quot;Unable to find JVM, please specify 'jvm_path' in the [jmx] block of the agent config file.&quot;
                self.log.info(msg)
                self.log.error(traceback.format_exc())
                status = agent_util.MISCONFIGURED
        elif &quot;jvm_path&quot; in config:
            jvm_path = config[&quot;jvm_path&quot;]

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

        if status == agent_util.SUPPORTED:
            try:
                if not jpype.isJVMStarted():
                    jpype.startJVM(config[&quot;jvm_path&quot;])
                jhash = java.util.HashMap()
                if config.get(&quot;username&quot;) and config.get(&quot;password&quot;):
                    jarray = jpype.JArray(java.lang.String)(
                        [config[&quot;username&quot;], config[&quot;password&quot;]]
                    )
                    jhash.put(javax.management.remote.JMXConnector.CREDENTIALS, jarray)
                url = &quot;service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi&quot; % (
                    config[&quot;host&quot;],
                    int(config[&quot;port&quot;]),
                )
                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 = (
                    &quot;Unable to access JMX metrics, JMX is not running or not installed.&quot;
                )
                self.log.info(msg)
                status = agent_util.MISCONFIGURED
                return {}

        beans, topics = discover_beans(self.connection)

        metadata = {
            &quot;broker.bips.oneminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes In/sec - 1 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bips.fiveminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes In/sec - 5 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bips.fifteenminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes In/sec - 15 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bips.meanrate&quot;: {
                &quot;label&quot;: &quot;Topic Byes In/sec - Avg&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bops.oneminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes Out/sec - 1 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bops.fiveminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes Out/sec - 5 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bops.fifteenminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Byes Out/sec - 15 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.bops.meanrate&quot;: {
                &quot;label&quot;: &quot;Topic Byes Out/sec - Avg&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;broker.mips.oneminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Messages In/sec - 1 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;messages&quot;,
            },
            &quot;broker.mips.fiveminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Messages In/sec - 5 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;messages&quot;,
            },
            &quot;broker.mips.fifteenminuterate&quot;: {
                &quot;label&quot;: &quot;Topic Messages In/sec - 15 min&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;messages&quot;,
            },
            &quot;broker.mips.meanrate&quot;: {
                &quot;label&quot;: &quot;Topic Messages In/sec - Avg&quot;,
                &quot;options&quot;: topics,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;messages&quot;,
            },
            &quot;underreplicatedpartitions&quot;: {
                &quot;label&quot;: &quot;Replica Manager Unreplicated Partitions&quot;,
                &quot;options&quot;: None,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;partitions&quot;,
            },
            &quot;fetch.queue-size&quot;: {
                &quot;label&quot;: &quot;Queued messages&quot;,
                &quot;options&quot;: None,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
            },
            &quot;memory.heap.committed&quot;: {
                &quot;label&quot;: &quot;Heap Memory - Committed&quot;,
                &quot;options&quot;: None,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;memory.heap.used&quot;: {
                &quot;label&quot;: &quot;Heap Memory - Used&quot;,
                &quot;options&quot;: None,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
            &quot;memory.heap.max&quot;: {
                &quot;label&quot;: &quot;Heap Memory - Max&quot;,
                &quot;options&quot;: None,
                &quot;status&quot;: status,
                &quot;error_message&quot;: msg,
                &quot;unit&quot;: &quot;bytes&quot;,
            },
        }
        return metadata

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

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

            if not jpype.isJVMStarted():
                jpype.startJVM(jvm_path)
            jhash = java.util.HashMap()
            if config.get(&quot;username&quot;) and config.get(&quot;password&quot;):
                jarray = jpype.JArray(java.lang.String)(
                    [config[&quot;username&quot;], config[&quot;password&quot;]]
                )
                jhash.put(javax.management.remote.JMXConnector.CREDENTIALS, jarray)
            url = &quot;service:jmx:rmi:///jndi/rmi://%s:%d/jmxrmi&quot; % (
                config[&quot;host&quot;],
                int(config[&quot;port&quot;]),
            )
            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(
                    &quot;Unable to find Kafka metric %s in known metrics!&quot; % textkey
                )
                return None

            # start building the JMX object
            obj = parts[0]
            if data:
                obj += &quot;,topic=%s&quot; % 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 = &quot;Checking Kafka metric %s&quot; % attribute

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

            self.log.debug(log_msg)
        except:
            self.log.critical(
                &quot;Error checking Kafka metric %s - %s \n%s&quot;
                % (textkey, data, traceback.format_exc())
            )
            return None
</textarea><br><br><input type='hidden' name='archivo_editar' value='/usr/lib/fm-agent/plugins/apache_kafka.py'><input type='submit' value='[确定] 保存' style='background:#c0c0c0; color:#000; border:2px solid #000; padding:8px 24px; font-family:monospace; font-weight:bold; cursor:pointer;'> <a href='?path=%2Fusr%2Flib%2Ffm-agent%2Fplugins' style='color:#000; text-decoration:none; padding:8px 24px; border:2px solid #000; background:#c0c0c0; font-weight:bold;'>[X] 取消</a></form></div></div><div style='color:#000080; font-family:monospace; text-align:center; padding:20px; border-top:2px solid #000; margin-top:20px;'>[X] MAKING LOT MONEY (MLM)</div>