如何将AWS MSK集群日志配置为JSON格式？

核心原理 #

MSK Broker的日志格式由Kafka的log4j配置控制，要输出JSON结构化日志，需通过MSK的Configuration资源修改log4j的输出布局，指定JSON格式的Appender。

步骤1：定义/更新MSK Configuration资源 #

在你的CloudFormation模板中，添加或更新AWS::MSK::Configuration资源，写入log4j的JSON格式配置规则。如果已有该资源，需生成新版本的配置内容：

MSKClusterConfiguration:
  Type: AWS::MSK::Configuration
  Properties:
    Name: "MSK-Json-Log-Config"
    Description: "配置MSK Broker输出JSON格式日志"
    KafkaVersionsList: [!Ref KafkaVersion]
    ServerProperties: |
      # 配置log4j输出JSON格式
      log4j.rootLogger=INFO, stdout, kafkaAppender
      # 控制台输出JSON格式
      log4j.appender.stdout=org.apache.log4j.ConsoleAppender
      log4j.appender.stdout.layout=com.fasterxml.jackson.databind.log4j.JsonLayout
      log4j.appender.stdout.layout.compact=false
      log4j.appender.stdout.layout.locationInfo=true
      # 文件输出JSON格式
      log4j.appender.kafkaAppender=org.apache.log4j.RollingFileAppender
      log4j.appender.kafkaAppender.layout=com.fasterxml.jackson.databind.log4j.JsonLayout
      log4j.appender.kafkaAppender.layout.compact=false
      log4j.appender.kafkaAppender.layout.locationInfo=true
      log4j.appender.kafkaAppender.file=/var/log/kafka/server.log
      log4j.appender.kafkaAppender.maxFileSize=100MB
      log4j.appender.kafkaAppender.maxBackupIndex=10

      # 保留你的原有Kafka核心配置（示例）
      listeners=PLAINTEXT://:9092
      advertised.listeners=PLAINTEXT://${AWS::PrivateDnsName}:9092
      # ... 其他你之前配置的Kafka参数

步骤2：更新MSK集群的配置引用 #

确保MSKCluster资源的ConfigurationInfo指向最新的配置版本，可直接通过CloudFormation属性获取最新版本号：

MSKCluster:
  Type: AWS::MSK::Cluster
  Properties:
    # ... 保留原有所有属性
    ConfigurationInfo:
      Arn: !Ref MSKClusterConfiguration
      Revision: !GetAtt MSKClusterConfiguration.LatestRevision
    # ... 保留原有LoggingInfo等其他属性

步骤3：部署更新后的CloudFormation栈 #

部署修改后的模板，MSK集群会自动滚动重启所有Broker节点，应用新的日志配置。重启完成后，CloudWatch Logs将开始接收JSON格式的日志。

预期JSON日志示例 #

更新后，日志会呈现结构化的JSON格式，方便Logs Insights进行字段提取和分析：

{
  "timestamp": "2023-03-08 13:13:38,336",
  "level": "INFO",
  "thread": "main",
  "loggerName": "org.apache.kafka.common.utils.AppInfoParser",
  "message": "Kafka version: 2.8.1",
  "locationInfo": {
    "className": "org.apache.kafka.common.utils.AppInfoParser",
    "methodName": "initialize",
    "fileName": "AppInfoParser.java",
    "lineNumber": "198"
  }
}

注意事项 #

• MSK的Kafka 2.8.1及以上版本默认包含com.fasterxml.jackson.databind.log4j.JsonLayout依赖，无需额外安装。
• 滚动重启期间，集群整体保持高可用，但单个Broker重启时会短暂不可用，建议在业务低峰期操作。
• 若需保留部分原有日志格式，可在ServerProperties中合并原有配置与JSON格式配置。

内容的提问来源于stack exchange，提问作者Brandon