Kafka

·

Collect Kafka indicators and logs and report them to Guance to help you monitor and analyze various abnormal situations of Kafka.

Configuration¶

Requirements¶

Install or download Jolokia. The downloaded Jolokia jar package is already available in the data directory under the DataKit installation directory.

Jolokia is a Java agent of Kafka, which provides an external interface using JSON as data format based on HTTP protocol for DataKit to use. When Kafka starts, first configure the KAFKA_OPTS environment variable: (port can be modified to be available according to the actual situation)

export KAFKA_OPTS="$KAFKA_OPTS -javaagent:/usr/local/datakit/data/jolokia-jvm-agent.jar=host=*,port=8080"

Alternatively, you can start Jolokia separately and point it to the Kafka process PID:

java -jar </path/to/jolokia-jvm-agent.jar> --host 127.0.0.1 --port=8080 start <Kafka-PID>

Collector Configuration¶

[[inputs.kafka]]
  # default_tag_prefix      = ""
  # default_field_prefix    = ""
  # default_field_separator = "."

  # username = ""
  # password = ""
  # response_timeout = "5s"

  ## Optional TLS config
  # tls_ca   = "/var/private/ca.pem"
  # tls_cert = "/var/private/client.pem"
  # tls_key  = "/var/private/client-key.pem"
  # insecure_skip_verify = false

  ## Monitor Interval
  # interval   = "60s"

  election = true

  # Add agents URLs to query
  urls = ["http://localhost:8080/jolokia"]

  ## Auto collect mode: when enable_auto_collect is true,
  ## automatically discover and collect all Kafka MBeans using search interface (kafka.*:*).
  ## Default is true.
  enable_auto_collect = true

  ## MBean blacklist for auto collect mode: exclude MBeans matching these patterns (supports wildcard * and ?).
  ## Note: Auto collect mode only collects Kafka-related MBeans (kafka.*:*)
  # mbean_blacklist = [
  #   "kafka.log:*",
  #   "kafka.server:name=*,topic=*,type=BrokerTopicMetrics",
  # ]

  ## Add metrics to read (only used when enable_auto_collect = false)
  [[inputs.kafka.metric]]
    name         = "kafka_controller"
    mbean        = "kafka.controller:name=*,type=*"
    field_prefix = "#1."

  [[inputs.kafka.metric]]
    name         = "kafka_replica_manager"
    mbean        = "kafka.server:name=*,type=ReplicaManager"
    field_prefix = "#1."

  [[inputs.kafka.metric]]
    name         = "kafka_zookeeper"
    mbean        = "kafka.server:type=ZooKeeperClientMetrics,name=*"
    field_prefix = "#1."

  [[inputs.kafka.metric]]
    name         = "kafka_purgatory"
    mbean        = "kafka.server:delayedOperation=*,name=*,type=DelayedOperationPurgatory"
    field_name   = "#1.#2"

  [[inputs.kafka.metric]]
    name     = "kafka_client"
    mbean    = "kafka.server:client-id=*,type=*"
    tag_keys = ["client-id", "type"]

  [[inputs.kafka.metric]]
    name         = "kafka_request"
    mbean        = "kafka.network:name=*,request=*,type=RequestMetrics"
    field_prefix = "#1."
    tag_keys     = ["request"]

  [[inputs.kafka.metric]]
    name         = "kafka_request_handler"
    mbean        = "kafka.server:type=KafkaRequestHandlerPool,name=*"
    field_prefix = "#1."

  [[inputs.kafka.metric]]
    name         = "kafka_network"
    mbean        = "kafka.network:type=*,name=*"
    field_name   = "#2"
    tag_keys     = ["type"]

  [[inputs.kafka.metric]]
    name         = "kafka_topics"
    mbean        = "kafka.server:name=*,type=BrokerTopicMetrics"
    field_prefix = "#1."

  [[inputs.kafka.metric]]
    name         = "kafka_topic"
    mbean        = "kafka.server:name=*,topic=*,type=BrokerTopicMetrics"
    field_prefix = "#1."
    tag_keys     = ["topic"]

  [[inputs.kafka.metric]]
    name       = "kafka_partition"
    mbean      = "kafka.log:name=*,partition=*,topic=*,type=Log"
    field_name = "#1"
    tag_keys   = ["topic", "partition"]

  [[inputs.kafka.metric]]
    name       = "kafka_log"
    mbean      = "kafka.log:type=*,name=*"
    field_name = "#2"
    tag_keys   = ["type"]

  [[inputs.kafka.metric]]
    name       = "kafka_partition"
    mbean      = "kafka.cluster:name=UnderReplicated,partition=*,topic=*,type=Partition"
    field_name = "UnderReplicatedPartitions"
    tag_keys   = ["topic", "partition"]

  # # The following metrics are available on consumer instances.
  # [[inputs.kafka.metric]]
  #   name       = "kafka_consumer"
  #   mbean      = "kafka.consumer:type=*,client-id=*"
  #   tag_keys   = ["client-id", "type"]

  # # The following metrics are available on producer instances.  
  # [[inputs.kafka.metric]]
  #   name       = "kafka_producer"
  #   mbean      = "kafka.producer:type=*,client-id=*"
  #   tag_keys   = ["client-id", "type"]

  # # The following metrics are available on connector instances.
  # [[inputs.kafka.metric]]
  #   name       = "kafka_connect"
  #   mbean      = "kafka.connect:type=*"
  #   tag_keys   = ["type"]

  # [[inputs.kafka.metric]]
  #   name       = "kafka_connect"
  #   mbean      = "kafka.connect:type=*,connector=*"
  #   tag_keys   = ["type", "connector"]

  # [[inputs.kafka.metric]]
  #   name       = "kafka_connect"
  #   mbean      = "kafka.connect:type=*,connector=*,task=*"
  #   tag_keys   = ["type", "connector", "task"]

  # [inputs.kafka.log]
  # files = []
  # #grok pipeline script path
  # pipeline = "kafka.p"

  [inputs.kafka.tags]
  # some_tag = "some_value"
  # more_tag = "some_other_value"

Collection Mode¶

The Kafka collector supports two collection modes:

1. Auto Collect Mode (default): Automatically discovers and collects all Kafka-related MBean metrics (kafka.*:*) without manual configuration.
2. Manual Mode: Requires manual configuration of MBean metrics to collect via [[inputs.kafka.metric]] configuration items.

Auto Collect Mode (Default)¶

By default, the Kafka collector uses auto collect mode, which automatically discovers and collects all Kafka MBean metrics. No additional configuration is required:

[[inputs.kafka]]
  urls = ["http://localhost:8080/jolokia"]
  enable_auto_collect = true  # Enabled by default

Important: When auto collect mode is enabled, all metrics will be unified into the kafka measurement, and field names will follow the domain.type.name.attr format. If you need to use the previous measurements (such as kafka_controller, kafka_replica_manager, kafka_topic, etc.) and field names, please set enable_auto_collect = false and use manual mode for configuration.

MBean Blacklist¶

In auto collect mode, you can exclude unwanted MBeans using the mbean_blacklist configuration item (supports wildcards * and ?):

[[inputs.kafka]]
  mbean_blacklist = [
    "kafka.log:*",  # Exclude all MBeans under kafka.log domain
    "kafka.server:name=*,topic=*,type=BrokerTopicMetrics",  # Exclude all topic-level BrokerTopicMetrics MBeans
  ]

Field Naming Rules¶

In auto collect mode, field names follow this format: domain.type.name.attr

• domain: MBean domain name (e.g., kafka.server, kafka.controller)
• type: MBean type property value (e.g., BrokerTopicMetrics, ReplicaManager)
• name: MBean name property value (e.g., MessagesInPerSec, BytesInPerSec)
• attr: MBean attribute name (e.g., Count, Value, MeanRate, OneMinuteRate, FiveMinuteRate, FifteenMinuteRate)

Common MBean attributes include:

• Count: Cumulative value since startup
• Value: Current value (for Gauge type metrics)
• MeanRate: Average rate since startup
• OneMinuteRate: Average rate over the last minute
• FiveMinuteRate: Average rate over the last five minutes
• FifteenMinuteRate: Average rate over the last fifteen minutes

Example field names:

• kafka.controller.KafkaController.GlobalTopicCount.Value: Global topic count
• kafka.server.ReplicaManager.PartitionCount.Value: Number of partitions

Tag Extraction Rules¶

Auto collect mode extracts tags from MBean properties. Common tags include:

• request: Request type
• topic: Kafka topic name
• delayedOperation: Delayed operation type
• error: Error type
• version: Version information
• partition: Kafka partition ID

Note: domain, type, and name will not be used as tags by default. These are included as part of the field name (format: domain.type.name.attr).

For detailed information about Kafka MBeans and their meanings, please refer to the Kafka official documentation monitoring metrics section.

Using Manual Mode¶

If you need to use manual mode, set enable_auto_collect = false and configure [[inputs.kafka.metric]] items:

[[inputs.kafka]]
  urls = ["http://localhost:8080/jolokia"]
  enable_auto_collect = false

  [[inputs.kafka.metric]]
    name         = "kafka_controller"
    mbean        = "kafka.controller:name=*,type=*"
    field_prefix = "#1."

Metric¶

For all of the following data collections, the global election tags will added automatically, we can add extra tags in [inputs.kafka.tags] if needed:

[inputs.kafka.tags]
  # some_tag = "some_value"
  # more_tag = "some_other_value"
  # ...

kafka¶

Metric set for auto collect mode only. Contains all Kafka MBean metrics collected automatically when enable_auto_collect=true. Fields are formatted as domain.type.name.attr.

collector¶

Log Collection¶

To collect kafka's log, open files in kafka.conf and write to the absolute path of the kafka log file. For example:

[[inputs.kafka]]
  ...
  [inputs.kafka.log]
    files = ["/usr/local/var/log/kafka/error.log","/usr/local/var/log/kafka/kafka.log"]

When log collection is turned on, a log with a log source of kafka is generated by default.

Note: DataKit must be installed on Kafka's host to collect Kafka logs.

Example of cutting logs:

[2020-07-07 15:04:29,333] DEBUG Progress event: HTTP_REQUEST_COMPLETED_EVENT, bytes: 0 (io.confluent.connect.s3.storage.S3OutputStream:286)

The list of cut fields is as follows:

FAQ¶

Why can't see kafka_producer/kafka_consumer/kafka_connect measurements?¶

After Kafka service is started, if you need to collect Producer/Consumer/Connector indicators, you need to configure Jolokia for them respectively.

Referring to Kafka Quick Start, configure the KAFKA_OPTS environment variable for the example of Producer, as follows:

export KAFKA_OPTS="-javaagent:/usr/local/datakit/data/jolokia-jvm-agent.jar=host=127.0.0.1,port=8090"

Go into the Kafka directory and start a Producer:

bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092

Copy a Kafka.conf to open multiple Kafka collectors and configure the url:

  urls = ["http://localhost:8090/jolokia"]

And remove comments from the fields in the collect producer metrics section:

  # The following metrics are available on producer instances.  
  [[inputs.kafka.metric]]
    name       = "kafka_producer"
    mbean      = "kafka.producer:type=*,client-id=*"
    tag_keys   = ["client-id", "type"]

Restart DataKit, which then collects metrics for the Producer instance.