Kafka Topics Metrics

With Kafka Topics Metrics Dashboard we can visualize the Metrics related to the topic. We can select the particular topic from the dropdown and see the data related to this topic.

Telegraf Input Plugins

Following are the basic telegraf Configuration can be use.

[global_tags]
Configuration for telegraf agent
[agent]
interval = "10s"
debug = false
hostname = "theone.us-central1-a.c.theone-277307.internal"
round_interval = true
flush_interval = "10s"
flush_jitter = "0s"
collection_jitter = "0s"
metric_batch_size = 1000
metric_buffer_limit = 10000
quiet = false
logfile = "/var/log/telegraf/telegraf.log"
omit_hostname = false
###############################################################################
OUTPUTS                                   
###############################################################################
[[outputs.prometheus_client]]
listen = ":9273"
###############################################################################
INPUTS                                    
###############################################################################
[[inputs.cpu]]
percpu = true
[[inputs.disk]]
[[inputs.io]]
[[inputs.mem]]
[[inputs.net]]
[[inputs.system]]
[[inputs.swap]]
[[inputs.netstat]]
[[inputs.processes]]
[[inputs.kernel]]
[[inputs.diskio]]
[[inputs.procstat]]
user = "root,telegraf,ubuntu"

Jolokia Agent Config

## Read JMX metrics through Jolokia
 [[inputs.jolokia2_agent]]
   ## An array of Kafka servers URI to gather stats.
    urls = ["http://localhost:8778/jolokia"]
    name_prefix = "kafka."
[[inputs.jolokia2_agent.metric]]
    name         = &#34;controller&#34;
    mbean        = &#34;kafka.controller:name=*,type=*&#34;
    field_prefix = &#34;$1.&#34;

[[inputs.jolokia2_agent.metric]]
    name         = &#34;controller&#34;
    mbean        = &#34;kafka.consumer:type=consumer-fetch-manager-metrics,client-id=([-.w]+)&#34;
    field_prefix = &#34;$1.&#34;
    tag_keys     = [&#34;topic&#34;, &#34;consumer&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;replica_manager&#34;
    mbean        = &#34;kafka.server:name=*,type=ReplicaManager&#34;
    field_prefix = &#34;$1.&#34;

[[inputs.jolokia2_agent.metric]]
    name         = &#34;server_replicafetchmanager.maxlag&#34;
    mbean        = &#34;kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica&#34;
    field_prefix = &#34;$1.&#34;
    tag_keys     = [&#34;topic&#34;, &#34;ReplicaFetcherManager&#34;]

[[inputs.jolokia2_agent.metric]]
    name = &#34;server_requesthandler_avgidlepct&#34;
    mbean  = &#34;kafka.server:name=RequestHandlerAvgIdlePercent,type=KafkaRequestHandlerPool&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;topic&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;purgatory&#34;
    mbean        = &#34;kafka.server:delayedOperation=*,name=*,type=DelayedOperationPurgatory&#34;
    field_prefix = &#34;$1.&#34;
    field_name   = &#34;$2&#34;

[[inputs.jolokia2_agent.metric]]
    name         = &#34;client&#34;
    mbean        = &#34;kafka.server:client-id=*,type=*&#34;
    tag_keys     = [&#34;client-id&#34;, &#34;type&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;server_replicafetchmanager.maxlag&#34;
    mbean        = &#34;kafka.server:type=ReplicaFetcherManager,name=MaxLag,clientId=Replica&#34;
    tag_keys     = [&#34;request&#34;, &#34;network&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;request&#34;
    mbean        = &#34;kafka.network:name=*,request=*,type=RequestMetrics&#34;
    field_prefix = &#34;$1.&#34;
    tag_keys     = [&#34;request&#34;, &#34;network&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;network_processor_avgidlepct&#34;
    mbean        = &#34;kafka.network:name=NetworkProcessorAvgIdlePercent,type=SocketServer&#34;
    field_prefix = &#34;$1.&#34;
    tag_keys     = [&#34;socketserver&#34;, &#34;network&#34;]

[[inputs.jolokia2_agent.metric]]
    name         = &#34;topics&#34;
    mbean        = &#34;kafka.server:name=*,type=BrokerTopicMetrics&#34;
    field_prefix = &#34;$1.&#34;

[[inputs.jolokia2_agent.metric]]
    name         = &#34;topic&#34;
    mbean        = &#34;kafka.server:name=*,topic=*,type=BrokerTopicMetrics&#34;
    field_prefix = &#34;$1.&#34;
    tag_keys     = [&#34;topic&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;partition&#34;
    mbean      = &#34;kafka.log:name=*,partition=*,topic=*,type=Log&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;topic&#34;, &#34;partition&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;partition&#34;
    mbean      = &#34;kafka.cluster:name=UnderReplicated,partition=*,topic=*,type=Partition&#34;
    field_name = &#34;UnderReplicatedPartitions&#34;
    tag_keys   = [&#34;topic&#34;, &#34;partition&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;controller_activecontrollers&#34;
    mbean      = &#34;kafka.controller:type=KafkaController,name=ActiveControllerCount&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;controller&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;controller_offlinepartitions&#34;
    mbean      = &#34;kafka.controller:type=KafkaController,name=OfflinePartitionsCount&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;controller&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;controller_stats_leaderelectionrateandtime&#34;
    mbean      = &#34;kafka.controller:type=ControllerStats,name=LeaderElectionRateAndTimeMs&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;controller&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;controller_stats_uncleanleaderelections&#34;
    mbean      = &#34;kafka.controller:type=ControllerStats,name=UncleanLeaderElectionsPerSec&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;controller&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;server_requesthandler_avgidlepct&#34;
    mbean      = &#34;kafka.server:name=RequestHandlerAvgIdlePercent,type=KafkaRequestHandlerPool&#34;
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;topic&#34;, &#34;requesthandler&#34;]

# Zookeeper Metrics
[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_disconnects&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperDisconnectsPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_sync_connects&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperSyncConnectsPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_auth_failures&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperAuthFailuresPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_readonly_connects&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperReadOnlyConnectsPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_authentications&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperSaslAuthenticationsPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;zookeeper_expires&#34;
    mbean      = &#34;kafka.server:type=SessionExpireListener,name=ZooKeeperExpiresPerSec&#34;
    paths      = [&#34;Count&#34;,&#34;OneMinuteRate&#34;,&#34;FiveMinuteRate&#34;,&#34;FifteenMinuteRate&#34;,&#34;MeanRate&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;heap_memory_usage&#34;
    mbean      = &#34;java.lang:type=Memory&#34;
    paths      = [&#34;HeapMemoryUsage&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;thread_count&#34;
    mbean      = &#34;java.lang:type=Threading&#34;
    paths      = [&#34;TotalStartedThreadCount&#34;,&#34;ThreadCount&#34;,&#34;DaemonThreadCount&#34;,&#34;PeakThreadCount&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;zookeeper&#34;]

[[inputs.jolokia2_agent.metric]]
    name       = &#34;garbage_collector&#34;
    mbean      = &#34;java.lang:type=GarbageCollector,name=*&#34;
    paths      = [&#34;CollectionCount&#34;,&#34;CollectionTime&#34;]
    field_name = &#34;$1&#34;
    tag_keys   = [&#34;garbage_collector&#34;]