中间件-Kafka日志

采集方法

使用ICAgent采集日志，接入前请参考安装ICAgent（区域内主机）完成ICAgent安装。然后再通过采集ECS-文本日志方式将日志上报至LTS服务，详细操作请参考云主机ECS文本日志接入LTS。

Kafka日志示例以及字段含义

由于Kafka本身并不直接提供系统日志的详细格式或字段定义，因为这部分可能会根据使用的日志框架而有所不同，例如 Log4j，所以本文提供一个基于典型日志框架（如 Log4j）的示例，并对其中的一些通用字段进行解析。

1. 系统日志示例。

   2023-10-23 15:33:45,123 INFO [KafkaServer id=0] onBrokerStartup (KafkaServer.scala:220) starting  
   2023-10-23 15:33:45,567 WARN [ReplicaManager broker=0] topic=my_topic partition=0 Broker 1 disconnected; no more in-sync replicas  
   2023-10-23 15:33:46,987 ERROR [Controller id=0] onControllerFailover (KafkaController.scala:678) Error during controller failover  
   java.lang.RuntimeException: Failed to elect new leader for partition [my_topic,0]  
       at kafka.controller.KafkaController
   $$
   anonfun$onControllerFailover$1
   $$
   anonfun$apply$mcV$sp$1.apply$mcV$sp(KafkaController.scala:629)  
       at kafka.controller.KafkaController
   $$
   anonfun$onControllerFailover$1
   $$
   anonfun$apply$mcV$sp$1.apply(KafkaController.scala:609)  
       at kafka.controller.KafkaController
   $$
   anonfun$onControllerFailover$1
   $$
   anonfun$apply$mcV$sp$1.apply(KafkaController.scala:609)  
       ... (stack trace continues)

2. 字段含义。

   在上面的示例中，第一个日志条目是KafkaServer在启动时产生的信息日志；第二个是一个警告，表明某个分区的一个同步副本（in-sync replica）与broker断开了连接；第三个是一个错误日志，表明在控制器故障转移期间发生了错误，并尝试为某个分区选择新的领导者时失败了。

   表1 字段含义

   字段	说明	示例
   时间戳	日志条目的生成时间。	2023-10-23 15:33:45,123
   日志级别	如INFO、WARN、ERROR等，表示日志条目的重要性或紧急性。	INFO
   线程或进程信息	提供了产生日志的上下文，如KafkaServer的ID或ReplicaManager的broker ID。	[KafkaServer id=0]
   日志来源	指示了日志条目的来源函数和代码位置。	onBrokerStartup (KafkaServer.scala:220)
   日志内容	记录了具体的日志信息，可能是描述性的文本、错误消息、堆栈跟踪等。	starting

日志位置

默认情况下，Kafka的系统日志文件可能位于Kafka安装目录的logs子目录下，但具体位置可能会因配置而异。以下是一些常见的方法用于找到Kafka的系统日志文件位置：

1. 查看Kafka安装目录。

   首先，检查Kafka的安装目录。通常，系统日志文件会存储在安装目录下的一个名为logs的子目录中。例如，如果Kafka安装在/usr/local/kafka，则日志文件可能位于/usr/local/kafka/logs。

2. 检查Kafka配置文件：

   在${KAFKA_HOME}/bin/kafka-run-class.sh中修改Kafka的logs目录。

   # Log directory to use
   if [ "x$LOG_DIR" = "x" ]; then
       # LOG_DIR="$base_dir/logs"
       LOG_DIR="/data/kafka_logs"
   fi