kafka-logger plugin for API7 Gateway

Skill for configuring the API7 Enterprise Edition (API7 EE) kafka-logger plugin via the a7 CLI. Covers pushing access logs to Apache Kafka topics, broke…

Overview

The kafka-logger plugin pushes request/response logs to Apache Kafka topics. It supports multiple brokers, SASL authentication, async/sync producing, custom log formats, and batch processing for efficient delivery.

When to Use

Stream access logs to Kafka for downstream processing
Feed real-time API analytics pipelines
Integrate with Kafka-based logging infrastructure
Need SASL-authenticated Kafka clusters

Plugin Configuration Reference

Core Parameters

Field	Type	Required	Default	Description
`brokers`	array	Yes	—	Kafka broker list
`brokers[].host`	string	Yes	—	Broker hostname or IP
`brokers[].port`	integer	Yes	—	Broker port (1-65535)
`kafka_topic`	string	Yes	—	Target Kafka topic
`key`	string	No	—	Partition key for routing
`timeout`	integer	No	`3`	Connection timeout in seconds

SASL Authentication

Field	Type	Required	Default	Description
`brokers[].sasl_config`	object	No	—	SASL config per broker
`brokers[].sasl_config.mechanism`	string	No	`"PLAIN"`	`PLAIN`, `SCRAM-SHA-256`, or `SCRAM-SHA-512`
`brokers[].sasl_config.user`	string	Yes*	—	SASL username (*if sasl_config set)
`brokers[].sasl_config.password`	string	Yes*	—	SASL password (*if sasl_config set)

Producer Configuration

Field	Type	Default	Description
`producer_type`	string	`"async"`	`async` (batched) or `sync` (immediate)
`required_acks`	integer	`1`	`1` (leader ack) or `-1` (all replicas)
`producer_batch_num`	integer	`200`	Messages per Kafka batch
`producer_batch_size`	integer	`1048576`	Batch size in bytes (1MB)

Log Format Options

Field	Type	Default	Description
`meta_format`	string	`"default"`	`default` (JSON) or `origin` (raw HTTP)
`log_format`	object	—	Custom log format with `$variable` syntax
`include_req_body`	boolean	`false`	Include request body
`include_req_body_expr`	array	—	Conditional request body logging

Step-by-Step: Ship Logs to Kafka

1. Create a service and route with kafka-logger

Replace <gateway-group-id> with the ID returned by a7 gateway-group list -o json, then enable logging:

a7 service create --gateway-group <gateway-group-id> -f - <<'EOF'
{
  "id": "kafka-logged-api-service",
  "name": "Kafka logged API service",
  "upstream": {
    "type": "roundrobin",
    "nodes": [{"host": "backend", "port": 8080, "weight": 1}]
  }
}
EOF

a7 route create --gateway-group <gateway-group-id> -f - <<'EOF'
{
  "id": "kafka-logged-api",
  "name": "Kafka logged API route",
  "paths": ["/api/*"],
  "service_id": "kafka-logged-api-service",
  "plugins": {
    "kafka-logger": {
      "brokers": [
        {"host": "kafka-1", "port": 9092},
        {"host": "kafka-2", "port": 9092}
      ],
      "kafka_topic": "api7-logs",
      "batch_max_size": 100
    }
  }
}
EOF

2. Global logging for a gateway group

Apply a Global Rule for all traffic in the target gateway group:

Do not set an id in the create payload. The CLI derives the Global Rule ID from the plugin name.

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

Common Patterns

SASL-authenticated Kafka cluster

{
  "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
    }
  }
}

Custom log format

{
  "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"
      }
    }
  }
}

Partition by route ID

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

Troubleshooting

Symptom	Cause	Fix
No messages in Kafka	Broker unreachable	Verify broker host/port; check firewall from gateway nodes
SASL auth failure	Wrong credentials or mechanism	Verify user/password; ensure mechanism matches Kafka config
Messages delayed	Large batch/timeout settings	Reduce `inactive_timeout` and `producer_time_linger`
Messages dropped	Buffer overflow	Increase `producer_max_buffering`; add more brokers
Topic not found	Topic doesn't exist	Create topic manually in Kafka or enable auto-creation
Config not applied	Wrong gateway group specified	Ensure `--gateway-group` matches the desired cluster

Config Sync Example

Save the following as kafka-logger.yaml:

version: "1"
services:
  - id: kafka-logged-api-service
    name: Kafka logged API service
    upstream:
      type: roundrobin
      nodes:
        - host: backend
          port: 8080
          weight: 1
routes:
  - id: kafka-logged-api
    name: Kafka logged API route
    paths:
      - /api/*
    service_id: kafka-logged-api-service
    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

Validate and apply this partial configuration to the target gateway group:

a7 config validate -f kafka-logger.yaml
a7 config sync -g <gateway-group-id> -f kafka-logger.yaml --delete=false

Disabling deletion preserves resources that are not included in this partial configuration.

This page is generated from a7-plugin-kafka-logger/SKILL.md in the api7/a7 repository. Browse all skills on the AI Agent Skills page.