跳到主要内容
版本：3.10.x

API7 网关 AI Agent Skill：kafka-logger 插件

概览​

kafka-logger 插件会将请求/响应日志推送到 Apache Kafka topic。它支持多个 broker、SASL 认证、异步/同步 生产、自定义日志格式和批处理，以实现高效投递。

适用场景​

  • 将访问日志流式写入 Kafka，供下游处理
  • 为实时 API 分析流水线提供数据
  • 与基于 Kafka 的日志基础设施集成
  • 适用于需要 SASL 认证的 Kafka 集群

插件配置参考​

核心参数​

字段类型是否必填默认值说明
brokersarray是—Kafka broker 列表
brokers[].hoststring是—Broker 主机名或 IP
brokers[].portinteger是—Broker 端口（1-65535）
kafka_topicstring是—目标 Kafka topic
keystring否—用于路由的分区 key
timeoutinteger否3连接超时时间，单位秒

SASL 认证​

字段类型是否必填默认值说明
brokers[].sasl_configobject否—每个 broker 的 SASL 配置
brokers[].sasl_config.mechanismstring否"PLAIN"PLAIN、SCRAM-SHA-256 或 SCRAM-SHA-512
brokers[].sasl_config.userstring是*—SASL 用户名（设置 sasl_config 时必填）
brokers[].sasl_config.passwordstring是*—SASL 密码（设置 sasl_config 时必填）

生产者配置​

字段类型默认值说明
producer_typestring"async"async（批量）或 sync（即时）
required_acksinteger11（leader 确认）或 -1（全部副本）
producer_batch_numinteger200每个 Kafka 批次的消息数
producer_batch_sizeinteger1048576批次大小，单位字节（1MB）

日志格式选项​

字段类型默认值说明
meta_formatstring"default"default（JSON）或 origin（原始 HTTP）
log_formatobject—使用 $variable 语法的自定义日志格式
include_req_bodybooleanfalse包含请求体
include_req_body_exprarray—有条件的请求体日志记录

分步指南：将日志发送到 Kafka​

1. 创建启用 kafka-logger 的路由​

为网关组 default 启用日志：

a7 route create --gateway-group default -f - <<'EOF'
{
  "id": "kafka-logged-api",
  "uri": "/api/*",
  "plugins": {
    "kafka-logger": {
      "brokers": [
        {"host": "kafka-1", "port": 9092},
        {"host": "kafka-2", "port": 9092}
      ],
      "kafka_topic": "api7-logs",
      "batch_max_size": 100
    }
  },
  "upstream": {
    "type": "roundrobin",
    "nodes": [{"host": "backend", "port": 8080, "weight": 1}]
  }
}
EOF

2. 为网关组配置全局日志​

为 prod 网关组中的所有流量应用全局规则：

a7 global_rule create --gateway-group prod -f - <<'EOF'
{
  "id": "kafka-logger-global",
  "plugins": {
    "kafka-logger": {
      "brokers": [{"host": "kafka-broker", "port": 9092}],
      "kafka_topic": "prod-logs",
      "batch_max_size": 500
    }
  }
}
EOF

常见模式​

使用 SASL 认证的 Kafka 集群​

{
  "plugins": {
    "kafka-logger": {
      "brokers": [
        {
          "host": "kafka.example.com",
          "port": 9092,
          "sasl_config": {
            "mechanism": "SCRAM-SHA-256",
            "user": "api7-user",
            "password": "secret-password"
          }
        }
      ],
      "kafka_topic": "secure-logs",
      "required_acks": -1
    }
  }
}

自定义日志格式​

{
  "plugins": {
    "kafka-logger": {
      "brokers": [{"host": "kafka", "port": 9092}],
      "kafka_topic": "api-logs",
      "log_format": {
        "@timestamp": "$time_iso8601",
        "client_ip": "$remote_addr",
        "method": "$request_method",
        "uri": "$request_uri",
        "status": "$status",
        "latency": "$request_time"
      }
    }
  }
}

按路由 ID 分区​

{
  "plugins": {
    "kafka-logger": {
      "brokers": [{"host": "kafka", "port": 9092}],
      "kafka_topic": "api-logs",
      "key": "$route_id"
    }
  }
}

故障排查​

现象原因修复方式
Kafka 中没有消息Broker 不可达验证 Broker 主机和端口，并检查网关节点的防火墙规则
SASL 身份验证失败凭证或机制错误验证用户名和密码，并确保机制与 Kafka 配置匹配
消息延迟批次或超时设置过大降低 inactive_timeout 和 producer_time_linger
消息丢失缓冲区溢出增大 producer_max_buffering，并增加 broker
未找到 topictopic 不存在在 Kafka 中手动创建 topic，或启用自动创建
配置未生效指定了错误的网关组确保 --gateway-group 与目标集群匹配

配置同步示例​

version: "1"
gateway_group: default
routes:
  - id: kafka-logged-api
    uri: /api/*
    plugins:
      kafka-logger:
        brokers:
          - host: kafka-1
            port: 9092
          - host: kafka-2
            port: 9092
        kafka_topic: api7-logs
        producer_type: async
        required_acks: 1
        batch_max_size: 200
        inactive_timeout: 5
    upstream:
      type: roundrobin
      nodes:
        - host: backend
          port: 8080
          weight: 1

本页面由 api7/a7 仓库中的 a7-plugin-kafka-logger/SKILL.md 生成。你可以在 AI Agent Skills 页面查看所有技能。