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

概览

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

适用场景

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

插件配置参考

核心参数

字段	类型	是否必填	默认值	说明
brokers	array	是	—	Kafka Broker 列表
brokers[].host	string	是	—	Broker 主机名或 IP
brokers[].port	integer	是	—	Broker 端口（1-65535）
kafka_topic	string	是	—	目标 Kafka Topic
key	string	否	—	用于路由的分区 key
timeout	integer	否	3	连接超时时间，单位秒

SASL 身份认证

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

生产者配置

字段	类型	默认值	说明
producer_type	string	"async"	async（批量）或 sync（即时）
required_acks	integer	1	1（leader 确认）或 -1（全部副本）
producer_batch_num	integer	200	每个 Kafka 批次的消息数
producer_batch_size	integer	1048576	批次大小，单位字节（1MB）

日志格式选项

字段	类型	默认值	说明
meta_format	string	"default"	default（JSON）或 origin（原始 HTTP）
log_format	object	—	使用 $variable 语法的自定义日志格式
include_req_body	boolean	false	包含请求体
include_req_body_expr	array	—	有条件的请求体日志记录

分步指南：将日志发送到 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
未找到 Topic	Topic 不存在	在 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 页面查看所有技能。