Skip to main content
API7.ai Logo

kafka-logger

The kafka-logger plugin pushes request and response logs as JSON objects to Apache Kafka clusters in batches and supports the customization of log formats.

Examples​

The examples below demonstrate how you can configure kafka-logger plugin for different scenarios.

To follow along the examples, start a sample Kafka cluster. The Docker example attaches the broker to the apisix-quickstart-net network created by the APISIX Docker quickstart, so the gateway can resolve notkafka:29092.

docker-compose.yml
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.8.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    networks:
      - apisix-quickstart-net

  notkafka:
    image: confluentinc/cp-kafka:7.8.0
    container_name: notkafka
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://notkafka:29092,PLAINTEXT_HOST://127.0.0.1:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: "true"
    ports:
      - "9092:9092"
    networks:
      - apisix-quickstart-net

networks:
  apisix-quickstart-net:
    external: true

Start containers:

docker compose up -d

Wait for messages in the configured Kafka topic:

docker exec -it notkafka kafka-console-consumer --bootstrap-server localhost:9092 --topic test2 --from-beginning

Open a new terminal session for the following steps working with APISIX.

Log in Different Meta Log Formats​

The following example demonstrates how you can enable the kafka-logger plugin on a route, which logs client requests to the route and pushes logs to Kafka. You will also understand the differences between the default and origin meta log formats.

Create a route with kafka-logger as follows:

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": "notkafka",
            "port": 29092
          }
        ],
        "kafka_topic": "test2",
        "key": "key1",
        "batch_max_size": 1
      }
    },
    "upstream": {
      "nodes": {
        "httpbin.org:80": 1
      },
      "type": "roundrobin"
    }
  }'

❶ 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:

{
  "latency": 411.00001335144,
  "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",
      "x-forwarded-port": "9080"
    },
    "method": "GET",
    "size": 83,
    "uri": "/get",
    "url": "http://127.0.0.1:9080/get"
  },
  "response": {
    "headers": {
      "content-length": "233",
      "access-control-allow-credentials": "true",
      "content-type": "application/json",
      "connection": "close",
      "access-control-allow-origin": "*",
      "date": "Fri, 10 Nov 2023 06:02:44 GMT",
      "server": "APISIX/3.16.0"
    },
    "status": 200,
    "size": 475
  },
  "route_id": "kafka-logger-route",
  "client_ip": "127.0.0.1",
  "server": {
    "hostname": "apisix",
    "version": "3.16.0"
  },
  "apisix_latency": 18.00001335144,
  "service_id": "",
  "upstream_latency": 393,
  "start_time": 1699596164550,
  "upstream": "54.90.18.68:80"
}

Update the meta log format to origin:

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

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 broker with a CA-signed TLS certificate on the apisix-quickstart-net network. 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:

kafka-tls-certs/client.properties
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 APISIX quickstart network:

docker run -d \
  --name kafka-tls \
  --hostname kafka-tls \
  --network apisix-quickstart-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:4.1.0

After the broker log contains Transition from STARTING to STARTED, 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 quickstart gateway. Append it to the existing system trust bundle so that the gateway continues to trust the public CA certificates already installed in the container:

docker cp kafka-tls-certs/ca.crt \
  apisix-quickstart:/usr/local/apisix/conf/kafka-example-ca.crt

docker exec apisix-quickstart 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 apisix-quickstart sh -c '
  config=/usr/local/apisix/conf/config.yaml
  certificate=/usr/local/apisix/conf/combined-ca-bundle.pem

  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 > /tmp/config.yaml

  cat /tmp/config.yaml > /usr/local/apisix/conf/config.yaml
  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}" \
  --data-binary @- <<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
  }
}

Log Request and Response Headers With Plugin Metadata​

The following example demonstrates how you can customize log format using plugin metadata and built-in variables to log specific headers from request and response.

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:

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": "notkafka",
            "port": 29092
          }
        ],
        "kafka_topic": "test2",
        "key": "key1",
        "batch_max_size": 1
      }
    },
    "upstream": {
      "nodes": {
        "httpbin.org:80": 1
      },
      "type": "roundrobin"
    }
  }'

❶ meta_format: set to the default log format. It is important to note that this is mandatory if you would like to customize log format with plugin metadata. If meta_format is set to origin, the log entries will remain in origin format.

❷ batch_max_size: set to 1 to send the log entry immediately.

Next, configure the plugin metadata for kafka-logger:

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": {
      "host": "$host",
      "@timestamp": "$time_iso8601",
      "client_ip": "$remote_addr",
      "env": "$http_env",
      "resp_content_type": "$sent_http_Content_Type"
    }
  }'

❶ 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": "2023-11-10T23:09:04+00:00",
  "host": "127.0.0.1",
  "client_ip": "127.0.0.1",
  "route_id": "kafka-logger-route",
  "env": "dev",
  "resp_content_type":"application/json"
}

Log Request Bodies Conditionally​

The following example demonstrates how you can conditionally log request body.

Create a route with kafka-logger as follows:

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": "notkafka",
            "port": 29092
          }
        ],
        "kafka_topic": "test2",
        "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"
    }
  }'

❶ 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.

info

If you have customized the log_format in addition to setting include_req_body or include_resp_body to true, the plugin would not include the bodies in the logs.

As a workaround, you may be able to use the NGINX variable $request_body in the log format, such as:

{
  "kafka-logger": {
    ...,
    "log_format": {"body": "$request_body"}
  }
}