Kafka 4

Kafka 4의 지표 데이터를 수집

구성¶

사전 조건¶

JMX Exporter 다운로드¶

다운로드 주소：https://github.com/prometheus/jmx_exporter/releases/tag/1.3.0

JMX 스크립트와 시작 매개변수 구성¶

주의: Producer, Consumer, Streams, Connect 지표를 수집하려면 각각 독립된 프로세스를 열어야 하며, 각 프로세스를 시작할 때는 해당 yaml 파일과 시작 스크립트를 바꿔야 합니다. 아래를 참고하세요.

KRaft Metrics¶

KRaft Metrics 설정 파일 kafka.yml 생성

# ------------------------------------------------------------
# Kafka 4 Prometheus JMX Exporter Configuration
# ------------------------------------------------------------
lowercaseOutputName: false
lowercaseOutputLabelNames: true
cacheRules: true
rules:

# 1. Broker / Topic / Partition Metrics
  - pattern: kafka.server<type=BrokerTopicMetrics, name=(BytesInPerSec|BytesOutPerSec|MessagesInPerSec|TotalFetchRequestsPerSec|ProduceRequestsPerSec|FailedProduceRequestsPerSec|TotalProduceRequestsPerSec|ReassignmentBytesInPerSec|ReassignmentBytesOutPerSec|ProduceMessageConversionsPerSec|FetchMessageConversionsPerSec)(?:, topic=([-\.\w]*))?><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate)
    name: kafka_server_broker_topic_metrics_$1
    type: GAUGE
    labels:
      topic: "$2"

# 2. Request / Network Metrics
  - pattern: kafka.network<type=RequestMetrics, name=(.+)><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate)
    name: kafka_network_request_metrics_$1
    type: GAUGE

# 3. Socket Server Metrics
  - pattern: kafka.network<type=SocketServer, name=(.+)><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate|Value)
    name: kafka_network_socket_server_metrics_$1
    type: GAUGE

# 4. Log / Segment / Cleaner Metrics
  - pattern: kafka.log<type=LogFlushStats, name=(.+)><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate)
    name: kafka_log_$1_$2
    type: GAUGE

# 5. Controller (KRaft) Metrics
  - pattern: kafka.controller<type=KafkaController, name=(.+)><>(Count|Value)
    name: kafka_controller_$1
    type: GAUGE

# 6. Group / Coordinator Metrics
  - pattern: kafka.coordinator.group<type=GroupMetadataManager, name=(.+)><>(Count|Value)
    name: kafka_coordinator_group_metadata_manager_$1
    type: GAUGE

# 7. KRaft Specific Metrics
  - pattern: kafka.controller<type=KafkaController, name=(LeaderElectionSuccessRate|LeaderElectionLatencyMs)><>(Count|Value)
    name: kafka_controller_$1
    type: GAUGE

# 8. New Generation Consumer Rebalance Protocol Metrics
  - pattern: kafka.coordinator.group<type=GroupMetadataManager, name=(RebalanceTimeMs|RebalanceFrequency)><>(Count|Value)
    name: kafka_coordinator_group_metadata_manager_$1
    type: GAUGE

# 9. Queue Metrics
  - pattern: kafka.server<type=Queue, name=(QueueSize|QueueConsumerRate)><>(Count|Value)
    name: kafka_server_queue_$1
    type: GAUGE

# 10. Client Metrics
  - pattern: kafka.network<type=RequestMetrics, name=(ClientConnections|ClientRequestRate|ClientResponseTime)><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate)
    name: kafka_network_request_metrics_$1
    type: GAUGE

# 11. Log Flush Rate and Time
  - pattern: kafka.log<type=LogFlushStats, name=LogFlushRateAndTimeMs><>(Count|OneMinuteRate|FiveMinuteRate|FifteenMinuteRate|MeanRate)
    name: kafka_log_log_flush_rate_and_time_ms
    type: GAUGE

시작 매개변수

export KAFKA_HEAP_OPTS="-Xms1g -Xmx1g"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote.port=9999 \
  -Dcom.sun.management.jmxremote.rmi.port=9999 \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Djava.rmi.server.hostname=127.0.0.1"
export Environment="KAFKA_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7071:/opt/jmx_exporter/kafka.yml"

/opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties

Producer Metrics¶

Producer Metrics 설정 파일 producer.yml 생성

---
lowercaseOutputName: true
rules:
  # 신규: producer-node-metrics
  - pattern: kafka\.producer<type=producer-node-metrics, client-id=([^,]+), node-id=([^>]+)><>([^:]+)
    name: kafka_producer_node_$3
    labels:
      client_id: "$1"
      node_id: "$2"
    type: GAUGE

  - pattern: 'kafka\.producer<type=producer-metrics, client-id=([^>]+)><>([^:,\s]+).*'
    name: 'kafka_producer_metrics_$2'
    labels:
      client_id: "$1"
    type: GAUGE

  # Selector의 모든 지표 수집(Kafka 4.0 신규)
  - pattern: 'kafka\.(?:(producer|consumer|connect))<type=(producer|consumer|connect)-metrics, client-id=([^>]+)><>(connection-.+|io-.+|network-.+|select-.+|send-.+|receive-.+|reauthentication-.+)'
    name: 'kafka_${1}_${4}'
    labels:
      client_id: '$3'
    type: GAUGE

시작 매개변수

export Environment="KAFKA_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7072:/opt/jmx_exporter/producer.yml"

/opt/kafka/kafka/bin/kafka-console-producer.sh \
  --broker-list localhost:9092 \
  --topic xxxx \
  --producer-property bootstrap.servers=localhost:9092

Consumer Metrics¶

Consumer Metrics 설정 파일consumer.yml 생성

lowercaseOutputName: true
rules:
  # consumer-coordinator-metrics
  - pattern: 'kafka\.consumer<type=consumer-coordinator-metrics, client-id=([^>]+)><>([^:,\s]+).*'
    name: 'kafka_consumer_coordinator_metrics_$2'
    labels:
      client_id: "$1"
    type: GAUGE

  - pattern: 'kafka\.consumer<type=consumer-metrics, client-id=([^>]+)><>([^:,\s]+).*'
    name: 'kafka_consumer_metrics_$2'
    labels:
      client_id: "$1"

시작 매개변수

export Environment="KAFKA_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7073:/opt/jmx_exporter/consumer.yml"

/opt/kafka/kafka/bin/kafka-console-consumer.sh \
  --broker-list localhost:9092 \
  --topic xxxx \
  --producer-property bootstrap.servers=localhost:9092

Streams Metrics¶

Streams Metrics 설정 파일stream.yml 생성

lowercaseOutputName: true
lowercaseOutputLabelNames: true

rules:
  # Kafka Streams 애플리케이션 지표 - 특수 문자 제거
  - pattern: 'kafka.streams<type=stream-metrics, client-id=(.+)><>([a-zA-Z0-9\-]+)$'
    name: kafka_streams_$2
    labels:
      client_id: "$1"

  # 특수 문자가 포함된 속성명 처리
  - pattern: 'kafka.streams<type=stream-metrics, client-id=(.+)><>([a-zA-Z0-9\-]+):(.+)$'
    name: kafka_streams_$2_$3
    labels:
      client_id: "$1"

  # Processor Node 지표
  - pattern: 'kafka.streams<type=stream-processor-node-metrics, client-id=(.+), task-id=(.+), processor-node-id=(.+)><>(.+)'
    name: kafka_streams_processor_$4
    labels:
      client_id: "$1"
      task_id: "$2"
      processor_node_id: "$3"

  # Task 지표
  - pattern: 'kafka.streams<type=stream-task-metrics, client-id=(.+), task-id=(.+)><>(.+)'
    name: kafka_streams_task_$3
    labels:
      client_id: "$1"
      task_id: "$2"

  # 스레드 지표
  - pattern: 'kafka.streams<type=stream-thread-metrics, client-id=(.+), thread-id=(.+)><>(.+)'
    name: kafka_streams_thread_$3
    labels:
      client_id: "$1"
      thread_id: "$2"

  # JVM 지표
  - pattern: 'java.lang<type=Memory><>(.+)'
    name: jvm_memory_$1

  - pattern: 'java.lang<type=GarbageCollector, name=(.+)><>(\w+)'
    name: jvm_gc_$2
    labels:
      gc: "$1"

  # 스레드 풀 지표
  - pattern: 'java.lang<type=Threading><>(.+)'
    name: jvm_threads_$1

  # 기본 규칙
  - pattern: '(.*)'

시작 매개변수

export KAFKA_HEAP_OPTS="-Xms512m -Xmx512m"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote.port=9996 \
  -Dcom.sun.management.jmxremote.rmi.port=9996 \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Djava.rmi.server.hostname=127.0.0.1"
export Environment="KAFKA_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7075:/opt/jmx_exporter/stream.yml"

java $KAFKA_HEAP_OPTS $KAFKA_JMX_OPTS $EXTRA_ARGS -cp "libs/*:my-streams.jar" WordCountDemo

Connect Metrics¶

Connect Metrics 설정 파일connect.yml 생성

lowercaseOutputName: true
lowercaseOutputLabelNames: true
rules:
  # 1) connect-worker-metrics (전역)
  - pattern: 'kafka\.connect<type=connect-worker-metrics><>([^:]+)'
    name: 'kafka_connect_worker_$1'
    type: GAUGE

  # 2) connect-worker-metrics,connector=xxx
  - pattern: 'kafka\.connect<type=connect-worker-metrics, connector=([^>]+)><>([^:]+)'
    name: 'kafka_connect_worker_$2'
    labels:
      connector: "$1"
    type: GAUGE

  # 3) connect-worker-rebalance-metrics
  - pattern: 'kafka\.connect<type=connect-worker-rebalance-metrics><>([^:]+)'
    name: 'kafka_connect_worker_rebalance_$1'
    type: GAUGE

  # 4) connector-task-metrics
  - pattern: 'kafka\.connect<type=connector-task-metrics, connector=([^>]+), task=([^>]+)><>([^:]+)'
    name: 'kafka_connect_task_$3'
    labels:
      connector: "$1"
      task_id: "$2"
    type: GAUGE

  # 5) sink-task-metrics
  - pattern: 'kafka\.connect<type=sink-task-metrics, connector=([^>]+), task=([^>]+)><>([^:]+)'
    name: 'kafka_connect_sink_task_$3'
    labels:
      connector: "$1"
      task_id: "$2"

시작 매개변수

export KAFKA_HEAP_OPTS="-Xms512m -Xmx512m"
export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
  -Dcom.sun.management.jmxremote.authenticate=false \
  -Dcom.sun.management.jmxremote.ssl=false \
  -Dcom.sun.management.jmxremote.port=9995 \
  -Dcom.sun.management.jmxremote.rmi.port=9995 \
  -Djava.rmi.server.hostname=127.0.0.1"
export Environment="KAFKA_OPTS=-javaagent:/opt/jmx_exporter/jmx_prometheus_javaagent.jar=7074:/opt/jmx_exporter/connect.yml"

# Kafka Connect 시작
/opt/kafka/kafka/bin/connect-distributed.sh /opt/kafka/kafka/config/connect-distributed.properties

시작이 완료되면 curl http://IP:포트번호/metrics로 수집된 모니터링 데이터를 확인할 수 있습니다.

DataKit 구성¶

datakit 설치 디렉터리의 conf.d/samples 디렉터리로 들어가 prom.conf.sample을 복사해 kafka.conf로 이름을 바꿉니다

cp prom.conf.sample kafka.conf

kafka.conf 조정

[[inputs.prom]]
  ## Exporter URLs.
  urls = ["http://127.0.0.1:7071/metrics","http://127.0.0.1:7072/metrics","http://127.0.0.1:7073/metrics","http://127.0.0.1:7074/metrics","http://127.0.0.1:7075/metrics"]

  ## Collector alias.
  source = "kafka"

  ## Prioritier over 'measurement_name' configuration.
  [[inputs.prom.measurements]]
    prefix = "kafka_controller_"
    name = "kafka_controller"

  [[inputs.prom.measurements]]
    prefix = "kafka_network_"
    name = "kafka_network"

  [[inputs.prom.measurements]]
    prefix = "kafka_log_"
    name = "kafka_log"

  [[inputs.prom.measurements]]
    prefix = "kafka_server_"
    name = "kafka_server"

  [[inputs.prom.measurements]]
    prefix = "kafka_connect_"
    name = "kafka_connect"

  [[inputs.prom.measurements]]
    prefix = "kafka_stream_"
    name = "kafka_stream"

DataKit 재시작

다음 명령을 실행합니다

datakit service -R

지표¶

아래는 kafka4의 일부 지표 설명이며, 더 많은 지표는 Kafka 지표 상세를 참고하세요

`kafka_server` 지표 집합¶

지표명	설명	단위
`Fetch_queue_size`	Fetch 대기열 크기	count
`Produce_queue_size`	Producer 대기열 크기	count
`Request_queue_size`	Request 대기열 크기	count
`broker_topic_metrics_BytesInPerSec`	클라이언트 바이트 입력 속도	bytes/s
`broker_topic_metrics_BytesOutPerSec`	클라이언트 바이트 출력 속도	bytes/s
`broker_topic_metrics_FailedProduceRequestsPerSec`	생산 요청 실패율	count/s
`broker_topic_metrics_FetchMessageConversionsPerSec`	Fetch 메시지 변환 속도	count/s
`broker_topic_metrics_MessagesInPerSec`	들어오는 메시지 속도	count/s
`broker_topic_metrics_ProduceMessageConversionsPerSec`	Producer 메시지 변환 속도	count/s
`broker_topic_metrics_TotalFetchRequestsPerSec`	가져오기 요청(클라이언트 또는 팔로워에서) 속도	count/s
`broker_topic_metrics_TotalProduceRequestsPerSec`	프로듀서 요청 속도	count/s
`socket_server_metrics_connection_count`	SocketServer 연결 수	count
`socket_server_metrics_connection_close_total`	SocketServer 닫힌 연결 수	count
`socket_server_metrics_incoming_byte_rate`	SocketServer 입력 바이트 속도	bytes/s

`kafka_network` 지표 집합¶

지표명	설명	단위
`request_metrics_RequestBytes_request_AddOffsetsToTxn`	AddOffsetsToTxn 요청 크기	bytes
`request_metrics_RequestBytes_request_Fetch`	Fetch 요청 크기	count
`request_metrics_RequestBytes_request_FetchConsumer`	FetchConsumer 요청 크기	bytes
`request_metrics_RequestBytes_request_FetchFollower`	FetchFollower 요청 크기	bytes
`request_metrics_TotalTimeMs_request_CreateTopics`	CreateTopics 요청 총 시간	ms
`request_metrics_TotalTimeMs_request_CreatePartitions`	CreatePartitions 요청 총 시간	ms
`request_metrics_RequestQueueTimeMs_request_CreateTopics`	CreateTopics 요청 대기열 대기 시간	ms
`request_metrics_RequestQueueTimeMs_request_CreatePartitions`	CreatePartitions 요청 대기열 대기 시간	ms
`request_metrics_RequestQueueTimeMs_request_Produce`	Produce 요청 대기열 대기 시간	ms
`request_metrics_ResponseSendTimeMs_request_CreateTopics`	CreateTopics 요청 응답 수신 시간	ms
`request_metrics_ResponseSendTimeMs_request_CreatePartitions`	CreatePartitions 요청 응답 수신 시간	ms

`kafka_controller` 지표 집합¶

지표명	설명	단위
`ActiveBrokerCount`	활성 Broker 수	count
`ActiveControllerCount`	활성 컨트롤러 수	count
`GlobalPartitionCount`	파티션 수	count
`GlobalTopicCount`	토픽 수	count
`OfflinePartitionsCount`	오프라인 파티션 수	count
`PreferredReplicaImbalanceCount`	Preferred Leader 선출 조건의 파티션 수	count
`OfflinePartitionsCount`	오프라인 파티션 수	count
`TimedOutBrokerHeartbeatCount`	Broker 심박수 초과 횟수	count
`LastAppliedRecordLagMs`	마지막으로 적용된 레코드 지연 시간	ms
`LastAppliedRecordOffset`	마지막으로 적용된 레코드 오프셋	-
`MetadataErrorCount`	메타데이터 오류 수	count
`NewActiveControllersCount`	새 컨트롤러 선출 횟수	count

`kafka_producer` 지표 집합¶

지표명	설명	단위
`producer_metrics_batch_split_rate`	배치 분할율	count/s
`producer_metrics_buffer_available_bytes`	미사용 버퍼 메모리 총량	bytes
`producer_metrics_buffer_exhausted_rate`	버퍼가 소진되어 폐기된 초당 평균 레코드 전송 수	count/s
`producer_metrics_buffer_total_bytes`	버퍼 총 바이트 크기	bytes
`producer_metrics_bufferpool_wait_ratio`	버퍼 풀 대기 비율	%
`producer_metrics_bufferpool_wait_time_ns_total`	버퍼 풀 대기 시간	ms
`producer_metrics_connection_close_rate`	연결 종료율	count/s
`producer_metrics_connection_count`	연결 종료 수	count
`producer_metrics_flush_time_ns_total`	플러시 총 시간	ns
`producer_metrics_incoming_byte_rate`	입력 바이트율	bytes/s
`producer_metrics_outgoing_byte_rate`	출력 바이트율	bytes/s
`producer_metrics_request_rate`	요청율	count/s
`producer_metrics_request_size_avg`	요청 크기	bytes

`kafka_consumer` 지표 집합¶

지표명	설명	단위
`consumer_coordinator_metrics_failed_rebalance_total`	재조정 실패 수	count
`consumer_coordinator_metrics_heartbeat_rate`	초당 평균 심박수	count/s
`consumer_coordinator_metrics_heartbeat_response_time_max`	심박 응답 최대 시간	count
`consumer_coordinator_metrics_join_rate`	Group 초당 가입율	count/s
`consumer_coordinator_metrics_join_total`	Group 가입 총수	count
`consumer_coordinator_metrics_last_rebalance_seconds_ago`	마지막 재조정 이벤트 이후의 초 수	ms
`consumer_coordinator_metrics_rebalance_latency_total`	재조정 지연 총계	ms
`consumer_fetch_manager_metrics_bytes_consumed_rate`	초당 소비한 바이트 수	bytes/s
`consumer_fetch_manager_metrics_fetch_latency_avg`	Fetch 요청 지연	ms
`consumer_metrics_connection_count`	연결 수	count
`consumer_metrics_connection_count`	닫힌 연결 수	count/s
`consumer_metrics_incoming_byte_rate`	입력 바이트율	bytes/s
`consumer_metrics_outgoing_byte_rate`	출력 바이트율	bytes/s
`consumer_metrics_select_rate`	Select 속도	count/s
`consumer_metrics_last_poll_seconds_ago`	IO 대기 시간	ms
`consumer_metrics_last_poll_seconds_ago`	IO 대기 시간	ms

`kafka_connect` 지표 집합¶

지표명	설명	단위
`worker_connector_count`	Connector 수	count
`worker_task_startup_attempts_total`	작업 시작 재시도 횟수	count
`worker_connector_startup_attempts_total`	커넥터 시작 시도 횟수	count
`worker_task_startup_failure_total`	작업 시작 실패 수	count
`worker_connector_startup_failure_percentage`	연결 실패율	%
`worker_rebalance_completed_rebalances_total`	재조정 완료 총수	count
`worker_task_startup_failure_percentage`	작업 시작 실패 비율	%
`worker_rebalance_time_since_last_rebalance_ms`	마지막 재조정 이후 경과 시간	ms
`worker_task_startup_attempts_total`	작업 시작 시도 횟수	count

`kafka_stream` 지표 집합¶

지표명	설명	단위
`stream_thread_metrics_thread_start_time`	스레드 시작 시간	타임스탬프 ms
`stream_thread_metrics_task_created_total`	작업 생성 총수	count
`stream_state_metrics_block_cache_capacity`	블록 캐시 크기	bytes
`stream_state_metrics_all_rate`	전체 작업 속도	count/s
`stream_state_metrics_block_cache_usage`	블록 캐시 사용률	%
`stream_state_metrics_bytes_read_compaction_rate`	바이트 읽기 압축률	bytes/s
`stream_state_metrics_bytes_written_compaction_rate`	바이트 쓰기 압축률	bytes/s
`stream_state_metrics_block_cache_index_hit_ratio`	블록 캐시 인덱스 적중률	%
`stream_state_metrics_block_cache_data_hit_ratio`	블록 캐시 데이터 적중률	%
`stream_state_metrics_block_cache_filter_hit_ratio`	블록 캐시 필터 적중률	%
`stream_state_metrics_bytes_written_rate`	바이트 쓰기 속도	bytes/s
`stream_state_metrics_bytes_read_rate`	바이트 읽기 속도	bytes/s
`stream_state_metrics_block_cache_filter_hit_ratio`	캐시 크기 바이트 수	bytes
`stream_task_metrics_process_rate`	초당 처리된 레코드 수	bytes/s
`stream_task_metrics_enforced_processing_rate`	초당 강제 처리 수	bytes/s
`stream_task_metrics_active_process_ratio`	활성 프로세스 비율	%
`stream_thread_metrics_commit_rate`	커밋율	count/s
`stream_thread_metrics_poll_latency_avg`	폴링 지연 시간	ms
`stream_thread_metrics_poll_rate`	폴링 속도	count/s
`stream_thread_metrics_blocked_time_ns_total`	차단 시간	ns
`stream_topic_metrics_bytes_consumed_total`	소비 바이트 수	bytes
`stream_topic_metrics_bytes_produced_total`	생성 바이트 수	bytes
`stream_topic_metrics_records_consumed_total`	소스 프로세서 노드가 소비한 레코드 총수	count
`stream_topic_metrics_records_produced_total`	싱크 프로세서 노드가 생성한 레코드 총수	count

Kafka 4

구성¶

사전 조건¶

JMX Exporter 다운로드¶

JMX 스크립트와 시작 매개변수 구성¶

KRaft Metrics¶

Producer Metrics¶

Consumer Metrics¶

Streams Metrics¶

Connect Metrics¶

DataKit 구성¶

지표¶

kafka_server 지표 집합¶

kafka_network 지표 집합¶

kafka_controller 지표 집합¶

kafka_producer 지표 집합¶

kafka_consumer 지표 집합¶

kafka_connect 지표 집합¶

kafka_stream 지표 집합¶

이 페이지가 도움이 되었나요? ×