kafka-logger
The kafka-logger plugin sends request and response logs as JSON objects to Apache Kafka in batches and supports customizable log formats.
Examples
The examples show how to send gateway request logs to Kafka, customize their contents, and secure the broker connection with TLS.
The examples use the official Apache Kafka 3.9.2 image in single-node KRaft mode.
kafka-logger supports Kafka Produce API versions 0–2. Kafka 4 removes those versions, so use a Kafka 3.x broker until the plugin supports Produce API version 3 or later.
- Docker
- Kubernetes
Set GATEWAY_CONTAINER to the running APISIX or API7 Gateway container:
export GATEWAY_CONTAINER=replace-with-gateway-container-name
Create a dedicated network for the gateway and Kafka:
docker network create gateway-kafka-net
Connect the gateway to the network:
docker network connect gateway-kafka-net "$GATEWAY_CONTAINER"
Create the following Docker Compose file:
services:
kafka-server:
image: apache/kafka:3.9.2
container_name: kafka-server
environment:
KAFKA_NODE_ID: 1
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka-server:9092
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka-server:9093
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
networks:
- kafka
networks:
kafka:
name: gateway-kafka-net
external: true
Start the broker:
docker compose up -d
After the container starts, verify that the broker accepts requests:
docker exec kafka-server /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--list
If the command reports a connection error, wait a few seconds and run it again.
Create a manifest for a single-node Kafka broker in combined broker and controller mode:
apiVersion: apps/v1
kind: Deployment
metadata:
namespace: aic
name: kafka-server
spec:
replicas: 1
selector:
matchLabels:
app: kafka-server
template:
metadata:
labels:
app: kafka-server
spec:
containers:
- name: kafka-server
image: apache/kafka:3.9.2
env:
- name: KAFKA_NODE_ID
value: "1"
- name: KAFKA_PROCESS_ROLES
value: broker,controller
- name: KAFKA_LISTENERS
value: PLAINTEXT://:9092,CONTROLLER://:9093
- name: KAFKA_ADVERTISED_LISTENERS
value: PLAINTEXT://kafka-server.aic.svc:9092
- name: KAFKA_CONTROLLER_LISTENER_NAMES
value: CONTROLLER
- name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
value: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
- name: KAFKA_CONTROLLER_QUORUM_VOTERS
value: 1@localhost:9093
- name: KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR
value: "1"
- name: KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR
value: "1"
- name: KAFKA_TRANSACTION_STATE_LOG_MIN_ISR
value: "1"
- name: KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS
value: "0"
ports:
- name: kafka
containerPort: 9092
- name: controller
containerPort: 9093
readinessProbe:
exec:
command:
- /opt/kafka/bin/kafka-topics.sh
- --bootstrap-server
- localhost:9092
- --list
initialDelaySeconds: 5
periodSeconds: 10
timeoutSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
namespace: aic
name: kafka-server
spec:
selector:
app: kafka-server
ports:
- name: kafka
port: 9092
targetPort: kafka
Apply the manifest:
kubectl apply -f kafka-server.yaml
Wait for the broker to become ready:
kubectl rollout status -n aic deployment/kafka-server
Create the topic used by the examples:
- Docker
- Kubernetes
docker exec kafka-server /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--if-not-exists \
--topic apisix-logs
kubectl exec -n aic deploy/kafka-server -- \
/opt/kafka/bin/kafka-topics.sh \
--bootstrap-server localhost:9092 \
--create \
--if-not-exists \
--topic apisix-logs
To inspect records produced by the examples, run a consumer in a separate terminal:
- Docker
- Kubernetes
docker exec -it kafka-server /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic apisix-logs \
--from-beginning
kubectl exec -n aic -it deploy/kafka-server -- \
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server localhost:9092 \
--topic apisix-logs \
--from-beginning
Log in Different Meta Log Formats
The following example sends route request logs to Kafka and shows the difference between the default and origin meta formats.
Create a route with kafka-logger as follows:
- Admin API
- ADC
- Ingress Controller
curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"meta_format": "default",
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'
services:
- name: httpbin
labels:
docs-example: kafka-logging
routes:
- uris:
- /get
name: kafka-logger-route
plugins:
kafka-logger:
meta_format: "default"
brokers:
- host: "kafka-server"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
upstream:
type: roundrobin
nodes:
- host: httpbin.org
port: 80
weight: 1
Preview the changes to services with the example label:
adc diff -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
Synchronize the reviewed changes:
adc sync -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
- Gateway API
- APISIX CRD
apiVersion: v1
kind: Service
metadata:
namespace: aic
name: httpbin-external-domain
spec:
type: ExternalName
externalName: httpbin.org
ports:
- name: http
port: 80
targetPort: 80
---
apiVersion: apisix.apache.org/v1alpha1
kind: PluginConfig
metadata:
namespace: aic
name: kafka-logger-plugin-config
spec:
plugins:
- name: kafka-logger
config:
meta_format: "default"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
---
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
parentRefs:
- name: apisix
rules:
- matches:
- path:
type: Exact
value: /get
filters:
- type: ExtensionRef
extensionRef:
group: apisix.apache.org
kind: PluginConfig
name: kafka-logger-plugin-config
backendRefs:
- name: httpbin-external-domain
port: 80
apiVersion: apisix.apache.org/v2
kind: ApisixUpstream
metadata:
namespace: aic
name: httpbin-external-domain
spec:
ingressClassName: apisix
externalNodes:
- type: Domain
name: httpbin.org
---
apiVersion: apisix.apache.org/v2
kind: ApisixRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
ingressClassName: apisix
http:
- name: kafka-logger-route
match:
paths:
- /get
methods:
- GET
upstreams:
- name: httpbin-external-domain
plugins:
- name: kafka-logger
config:
meta_format: "default"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
Apply the configuration:
kubectl apply -f kafka-logger-ic.yaml
❶ meta_format: set to the default log format.
❷ batch_max_size: set to 1 to send the log entry immediately.
Send a request to the route to generate a log entry:
curl -i "http://127.0.0.1:9080/get"
You should see an HTTP/1.1 200 OK response.
You should see a log entry in the Kafka topic similar to the following. Addresses, timing values, and version details vary by environment:
{
"latency": 1030.9998989105,
"request": {
"querystring": {},
"headers": {
"host": "127.0.0.1:9080",
"user-agent": "curl/8.7.1",
"accept": "*/*",
"x-forwarded-proto": "http",
"x-forwarded-host": "127.0.0.1:9080",
"x-forwarded-port": "9080"
},
"method": "GET",
"size": 80,
"uri": "/get",
"url": "http://127.0.0.1:9080/get"
},
"response": {
"headers": {
"content-length": "311",
"access-control-allow-credentials": "true",
"content-type": "application/json",
"connection": "close",
"access-control-allow-origin": "*",
"date": "Fri, 18 Sep 2026 03:32:31 GMT",
"server": "APISIX/3.18.0"
},
"status": 200,
"size": 539
},
"route_id": "kafka-logger-route",
"client_ip": "192.168.155.1",
"server": {
"hostname": "dd2886d0b7bf",
"version": "3.18.0"
},
"apisix_latency": 408.99989891052,
"service_id": "",
"upstream_latency": 622,
"start_time": 1789702349311,
"upstream": "98.88.64.13:80"
}
Update the meta log format to origin:
- Admin API
- ADC
- Ingress Controller
curl "http://127.0.0.1:9180/apisix/admin/routes/kafka-logger-route" -X PATCH \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"plugins": {
"kafka-logger": {
"meta_format": "origin"
}
}
}'
Update adc.yaml to set meta_format to origin:
services:
- name: httpbin
labels:
docs-example: kafka-logging
routes:
- uris:
- /get
name: kafka-logger-route
plugins:
kafka-logger:
meta_format: "origin"
brokers:
- host: "kafka-server"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
upstream:
type: roundrobin
nodes:
- host: httpbin.org
port: 80
weight: 1
Preview the changes to services with the example label:
adc diff -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
Synchronize the reviewed changes:
adc sync -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
- Gateway API
- APISIX CRD
Update kafka-logger-ic.yaml to set meta_format to origin:
apiVersion: apisix.apache.org/v1alpha1
kind: PluginConfig
metadata:
namespace: aic
name: kafka-logger-plugin-config
spec:
plugins:
- name: kafka-logger
config:
meta_format: "origin"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
Update kafka-logger-ic.yaml to set meta_format to origin:
apiVersion: apisix.apache.org/v2
kind: ApisixRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
ingressClassName: apisix
http:
- name: kafka-logger-route
match:
paths:
- /get
methods:
- GET
upstreams:
- name: httpbin-external-domain
plugins:
- name: kafka-logger
config:
meta_format: "origin"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
Apply the updated configuration:
kubectl apply -f kafka-logger-ic.yaml
Send a request to the route again to generate a new log entry:
curl -i "http://127.0.0.1:9080/get"
You should see an HTTP/1.1 200 OK response.
You should see a log entry in the Kafka topic similar to the following:
GET /get HTTP/1.1
x-forwarded-proto: http
x-forwarded-host: 127.0.0.1
user-agent: curl/8.7.1
x-forwarded-port: 9080
host: 127.0.0.1:9080
accept: */*
Send Logs to a TLS-Enabled Broker
The following Docker example starts a local Kafka 3.9.2 broker with a CA-signed TLS certificate on gateway-kafka-net. It then connects with certificate verification and uses Produce API version 2 so Kafka records the message timestamp. Install OpenSSL and the Java keytool command before continuing.
Generate a sample CA, a broker certificate for the kafka-tls container hostname, and the Java key store and trust store files required by Kafka:
mkdir -p kafka-tls-certs
openssl req -x509 -newkey rsa:2048 -nodes -days 365 \
-subj "/CN=kafka-example-ca" \
-keyout kafka-tls-certs/ca.key \
-out kafka-tls-certs/ca.crt
openssl req -newkey rsa:2048 -nodes \
-subj "/CN=kafka-tls" \
-keyout kafka-tls-certs/server.key \
-out kafka-tls-certs/server.csr
printf "subjectAltName=DNS:kafka-tls\n" > kafka-tls-certs/server-ext.cnf
openssl x509 -req -days 365 \
-in kafka-tls-certs/server.csr \
-CA kafka-tls-certs/ca.crt \
-CAkey kafka-tls-certs/ca.key \
-CAcreateserial \
-extfile kafka-tls-certs/server-ext.cnf \
-out kafka-tls-certs/server.crt
openssl pkcs12 -export \
-name kafka-tls \
-in kafka-tls-certs/server.crt \
-inkey kafka-tls-certs/server.key \
-certfile kafka-tls-certs/ca.crt \
-out kafka-tls-certs/kafka.keystore.p12 \
-passout pass:changeit
keytool -importkeystore -noprompt \
-srckeystore kafka-tls-certs/kafka.keystore.p12 \
-srcstoretype PKCS12 \
-srcstorepass changeit \
-destkeystore kafka-tls-certs/kafka.keystore.jks \
-deststoretype JKS \
-deststorepass changeit \
-destkeypass changeit
keytool -importcert -noprompt \
-alias kafka-example-ca \
-file kafka-tls-certs/ca.crt \
-keystore kafka-tls-certs/kafka.truststore.jks \
-storepass changeit
printf "changeit\n" > kafka-tls-certs/kafka_keystore_creds
printf "changeit\n" > kafka-tls-certs/kafka_ssl_key_creds
Create the client configuration used later to verify the record:
security.protocol=SSL
ssl.truststore.location=/etc/kafka/secrets/kafka.truststore.jks
ssl.truststore.password=changeit
ssl.endpoint.identification.algorithm=https
Start the TLS-enabled broker on the gateway network:
docker run -d \
--name kafka-tls \
--hostname kafka-tls \
--network gateway-kafka-net \
-v "${PWD}/kafka-tls-certs:/etc/kafka/secrets:ro" \
-e KAFKA_NODE_ID=1 \
-e KAFKA_PROCESS_ROLES=broker,controller \
-e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP="SSL:SSL,CONTROLLER:PLAINTEXT" \
-e KAFKA_ADVERTISED_LISTENERS="SSL://kafka-tls:9093" \
-e KAFKA_LISTENERS="SSL://:9093,CONTROLLER://:29093" \
-e KAFKA_CONTROLLER_QUORUM_VOTERS="1@kafka-tls:29093" \
-e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
-e KAFKA_INTER_BROKER_LISTENER_NAME=SSL \
-e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
-e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
-e KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1 \
-e KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1 \
-e KAFKA_SSL_KEYSTORE_FILENAME=kafka.keystore.jks \
-e KAFKA_SSL_KEYSTORE_CREDENTIALS=kafka_keystore_creds \
-e KAFKA_SSL_KEY_CREDENTIALS=kafka_ssl_key_creds \
-e KAFKA_SSL_TRUSTSTORE_LOCATION=/etc/kafka/secrets/kafka.truststore.jks \
-e KAFKA_SSL_TRUSTSTORE_PASSWORD=changeit \
-e KAFKA_SSL_CLIENT_AUTH=none \
-e CLUSTER_ID="4L6g3nShT-eMCtK--X86sw" \
apache/kafka:3.9.2
After the container starts, verify that the TLS listener accepts requests:
docker exec kafka-tls /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server kafka-tls:9093 \
--command-config /etc/kafka/secrets/client.properties \
--list
If the command reports a connection error, wait a few seconds and run it again.
Create the topic:
docker exec kafka-tls /opt/kafka/bin/kafka-topics.sh \
--bootstrap-server kafka-tls:9093 \
--command-config /etc/kafka/secrets/client.properties \
--create \
--if-not-exists \
--topic apisix-logs \
--partitions 1 \
--replication-factor 1
Set the broker address and topic for the route configuration:
export KAFKA_TLS_HOST="kafka-tls"
export KAFKA_TLS_PORT="9093"
export KAFKA_TOPIC="apisix-logs"
Copy the generated CA certificate into the gateway container identified by GATEWAY_CONTAINER:
docker cp kafka-tls-certs/ca.crt \
"$GATEWAY_CONTAINER":/usr/local/apisix/conf/kafka-example-ca.crt
Append the certificate to the existing system trust bundle so that the gateway continues to trust the public CA certificates already installed in the container:
docker exec "$GATEWAY_CONTAINER" sh -c '
cat /etc/ssl/certs/ca-certificates.crt \
/usr/local/apisix/conf/kafka-example-ca.crt \
> /usr/local/apisix/conf/combined-ca-bundle.pem
'
Update the trust bundle path in the quickstart configuration and reload the gateway:
docker exec -u 0 "$GATEWAY_CONTAINER" sh -c '
config=/usr/local/apisix/conf/config.yaml
certificate=/usr/local/apisix/conf/combined-ca-bundle.pem
output=$(mktemp /tmp/kafka-config.XXXXXX)
if grep -q "^ ssl_trusted_certificate:" "$config"; then
sed "s#ssl_trusted_certificate:.*#ssl_trusted_certificate: $certificate#" "$config"
elif grep -q "^ ssl:$" "$config"; then
sed "/^ ssl:$/a\\
ssl_trusted_certificate: $certificate" "$config"
else
sed "/^apisix:$/a\\
ssl:\\
ssl_trusted_certificate: $certificate" "$config"
fi > "$output"
cat "$output" > "$config"
rm "$output"
apisix reload
'
For a multi-instance deployment, distribute the combined trust bundle and configuration change to every gateway instance. The kafka-tls hostname matches the DNS name in the sample broker certificate.
Create a route that sends each log entry immediately to the TLS listener:
curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d @- <<EOF
{
"id": "kafka-logger-tls-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"brokers": [
{
"host": "${KAFKA_TLS_HOST}",
"port": ${KAFKA_TLS_PORT}
}
],
"kafka_topic": "${KAFKA_TOPIC}",
"api_version": 2,
"batch_max_size": 1,
"tls": {
"verify": true
}
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}
EOF
Send a request to generate a log entry:
curl -i "http://127.0.0.1:9080/get"
Consume one record from the broker container and print its timestamp:
docker exec kafka-tls /opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server kafka-tls:9093 \
--topic apisix-logs \
--consumer.config /etc/kafka/secrets/client.properties \
--from-beginning \
--max-messages 1 \
--property print.timestamp=true
The consumed record should contain the request log and a timestamp later than the Unix epoch. If certificate verification fails, confirm that the CA bundle is readable by the gateway and that KAFKA_TLS_HOST matches a name in the broker certificate.
The record should contain at least the following fields, together with a broker timestamp. The default log entry contains additional request, response, latency, and gateway fields.
{
"route_id": "kafka-logger-tls-route",
"request": {
"method": "GET",
"uri": "/get"
},
"response": {
"status": 200
}
}
Add Request and Response Headers With Plugin Metadata
The following example uses plugin metadata and built-in variables to add selected request and response headers to every kafka-logger instance.
In APISIX, plugin metadata is used to configure the common metadata fields of all plugin instances of the same plugin. It is useful when a plugin is enabled across multiple resources and requires a universal update to their metadata fields.
First, create a route with kafka-logger as follows:
- Admin API
- ADC
- Ingress Controller
curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/get",
"plugins": {
"kafka-logger": {
"meta_format": "default",
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'
services:
- name: httpbin
labels:
docs-example: kafka-logging
routes:
- uris:
- /get
name: kafka-logger-route
plugins:
kafka-logger:
meta_format: "default"
brokers:
- host: "kafka-server"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
upstream:
type: roundrobin
nodes:
- host: httpbin.org
port: 80
weight: 1
Preview the changes to services with the example label:
adc diff -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
Synchronize the reviewed changes:
adc sync -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
- Gateway API
- APISIX CRD
apiVersion: v1
kind: Service
metadata:
namespace: aic
name: httpbin-external-domain
spec:
type: ExternalName
externalName: httpbin.org
ports:
- name: http
port: 80
targetPort: 80
---
apiVersion: apisix.apache.org/v1alpha1
kind: PluginConfig
metadata:
namespace: aic
name: kafka-logger-plugin-config
spec:
plugins:
- name: kafka-logger
config:
meta_format: "default"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
---
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
parentRefs:
- name: apisix
rules:
- matches:
- path:
type: Exact
value: /get
filters:
- type: ExtensionRef
extensionRef:
group: apisix.apache.org
kind: PluginConfig
name: kafka-logger-plugin-config
backendRefs:
- name: httpbin-external-domain
port: 80
apiVersion: apisix.apache.org/v2
kind: ApisixUpstream
metadata:
namespace: aic
name: httpbin-external-domain
spec:
ingressClassName: apisix
externalNodes:
- type: Domain
name: httpbin.org
---
apiVersion: apisix.apache.org/v2
kind: ApisixRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
ingressClassName: apisix
http:
- name: kafka-logger-route
match:
paths:
- /get
methods:
- GET
upstreams:
- name: httpbin-external-domain
plugins:
- name: kafka-logger
config:
meta_format: "default"
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
Apply the configuration:
kubectl apply -f kafka-logger-ic.yaml
❶ meta_format: keep the default format so that fields from plugin metadata are included. Fields configured with log_format_extra are ignored when this field is set to origin.
❷ batch_max_size: set to 1 to send the log entry immediately.
Next, configure the plugin metadata for kafka-logger:
- Admin API
- ADC
- Ingress Controller
curl "http://127.0.0.1:9180/apisix/admin/plugin_metadata/kafka-logger" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"log_format_extra": {
"host": "$host",
"@timestamp": "$time_iso8601",
"client_ip": "$remote_addr",
"env": "$http_env",
"resp_content_type": "$sent_http_Content_Type"
}
}'
Plugin metadata is a global collection and cannot be isolated with a label selector. Export the complete collection before changing this entry:
adc dump -o adc-metadata.yaml --with-id \
--include-resource-type plugin_metadata
Add or update the kafka-logger entry while preserving every other entry in adc-metadata.yaml:
plugin_metadata:
# Keep all other plugin metadata entries from the exported file.
kafka-logger:
log_format_extra:
host: "$host"
"@timestamp": "$time_iso8601"
client_ip: "$remote_addr"
env: "$http_env"
resp_content_type: "$sent_http_Content_Type"
Preview the complete metadata change and confirm that it contains no unintended updates or deletions:
adc diff -f adc-metadata.yaml \
--include-resource-type plugin_metadata
Synchronize the reviewed plugin metadata collection:
adc sync -f adc-metadata.yaml \
--include-resource-type plugin_metadata
Add the following entry under spec.pluginMetadata in the complete GatewayProxy manifest used by the deployment:
kafka-logger:
log_format_extra:
host: "$host"
"@timestamp": "$time_iso8601"
client_ip: "$remote_addr"
env: "$http_env"
resp_content_type: "$sent_http_Content_Type"
Apply the updated complete manifest through the deployment's normal Kubernetes or GitOps workflow.
❶ log the custom request header env.
❷ log the response header Content-Type.
Send a request to the route with the env header:
curl -i "http://127.0.0.1:9080/get" -H "env: dev"
You should see a log entry in the Kafka topic similar to the following:
{
"@timestamp": "2026-09-18T03:32:31+00:00",
"host": "127.0.0.1",
"client_ip": "192.168.155.1",
"route_id": "kafka-logger-route",
"env": "dev",
"resp_content_type": "application/json"
}
Log Request Bodies Conditionally
The following example includes request bodies only when a query parameter satisfies a configured expression.
Create a route with kafka-logger as follows:
- Admin API
- ADC
- Ingress Controller
curl "http://127.0.0.1:9180/apisix/admin/routes" -X PUT \
-H "X-API-KEY: ${ADMIN_API_KEY}" \
-d '{
"id": "kafka-logger-route",
"uri": "/post",
"plugins": {
"kafka-logger": {
"brokers": [
{
"host": "kafka-server",
"port": 9092
}
],
"kafka_topic": "apisix-logs",
"key": "key1",
"batch_max_size": 1,
"include_req_body": true,
"include_req_body_expr": [["arg_log_body", "==", "yes"]]
}
},
"upstream": {
"nodes": {
"httpbin.org:80": 1
},
"type": "roundrobin"
}
}'
services:
- name: httpbin
labels:
docs-example: kafka-logging
routes:
- uris:
- /post
name: kafka-logger-route
plugins:
kafka-logger:
brokers:
- host: "kafka-server"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
include_req_body: true
include_req_body_expr:
- - "arg_log_body"
- "=="
- "yes"
upstream:
type: roundrobin
nodes:
- host: httpbin.org
port: 80
weight: 1
Preview the changes to services with the example label:
adc diff -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
Synchronize the reviewed changes:
adc sync -f adc.yaml \
--include-resource-type service \
--label-selector docs-example=kafka-logging
- Gateway API
- APISIX CRD
apiVersion: v1
kind: Service
metadata:
namespace: aic
name: httpbin-external-domain
spec:
type: ExternalName
externalName: httpbin.org
ports:
- name: http
port: 80
targetPort: 80
---
apiVersion: apisix.apache.org/v1alpha1
kind: PluginConfig
metadata:
namespace: aic
name: kafka-logger-plugin-config
spec:
plugins:
- name: kafka-logger
config:
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
include_req_body: true
include_req_body_expr:
- - "arg_log_body"
- "=="
- "yes"
---
apiVersion: gateway.networking.k8s.io/v1
kind: HTTPRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
parentRefs:
- name: apisix
rules:
- matches:
- path:
type: Exact
value: /post
filters:
- type: ExtensionRef
extensionRef:
group: apisix.apache.org
kind: PluginConfig
name: kafka-logger-plugin-config
backendRefs:
- name: httpbin-external-domain
port: 80
apiVersion: apisix.apache.org/v2
kind: ApisixUpstream
metadata:
namespace: aic
name: httpbin-external-domain
spec:
ingressClassName: apisix
externalNodes:
- type: Domain
name: httpbin.org
---
apiVersion: apisix.apache.org/v2
kind: ApisixRoute
metadata:
namespace: aic
name: kafka-logger-route
spec:
ingressClassName: apisix
http:
- name: kafka-logger-route
match:
paths:
- /post
methods:
- POST
upstreams:
- name: httpbin-external-domain
plugins:
- name: kafka-logger
config:
brokers:
- host: "kafka-server.aic.svc"
port: 9092
kafka_topic: "apisix-logs"
key: "key1"
batch_max_size: 1
include_req_body: true
include_req_body_expr:
- - "arg_log_body"
- "=="
- "yes"
Apply the configuration:
kubectl apply -f kafka-logger-ic.yaml
❶ include_req_body: set to true to include request body.
❷ include_req_body_expr: only include request body if the URL query string log_body is yes.
Send a request to the route with a URL query string satisfying the condition:
curl -i "http://127.0.0.1:9080/post?log_body=yes" -X POST -d '{"env": "dev"}'
You should see the request body logged:
{
...,
"method": "POST",
"body": "{\"env\": \"dev\"}",
"size": 179
}
}
Send a request to the route without any URL query string:
curl -i "http://127.0.0.1:9080/post" -X POST -d '{"env": "dev"}'
You should not observe the request body in the log.
The log_format_extra field shown above preserves the default log entry, including request and response bodies collected by the plugin. If you configure log_format instead, include the corresponding variables explicitly:
{
"include_req_body": true,
"include_resp_body": true,
"log_format": {
"request_body": "$request_body",
"response_body": "$resp_body"
}
}
Body size limits still apply. Use log_format_extra to add custom fields without replacing the default log entry.