How-to for using the Kafka Log policy

Transform these toa proper how-to:

# Testing the Kafka Log Policy example locally

This is the minimum setup needed to run the example in
`app/_ai_gateway_policies/kafka-log/index.md` against a local Kafka broker.

Requires [kafkactl](https://github.com/deviceinsight/kafkactl?tab=readme-ov-file#installation), `jq`, and the usual AI Gateway prereqs.

## Working directory and SASL credentials

```bash
mkdir -p ~/kafka-log-test && cd ~/kafka-log-test

cat <<'EOF' > kafka_server_jaas.conf
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafka_user"
password="kafka-password"
user_kafka_user="kafka-password";
};
EOF
```

## Broker

Single-node KRaft broker with two listeners. SASL on `9092` advertised as `host.docker.internal`
for the data plane, plaintext on `9094` advertised as `localhost` for local tooling.
`host.docker.internal` doesn't resolve from the host itself, which is why both exist.

```bash
cat <<'EOF' > docker-compose.yaml
name: kafka_aigw

services:
kafka:
image: apache/kafka:4.3.1
container_name: kafka
ports:
- "9092:9092"
- "9094:9094"
environment:
KAFKA_NODE_ID: 0
KAFKA_PROCESS_ROLES: broker,controller
KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
KAFKA_CONTROLLER_QUORUM_VOTERS: 0@localhost:9093
KAFKA_LISTENERS: CONTROLLER://localhost:9093,SASL_PLAINTEXT://0.0.0.0:9092,PLAINTEXT://0.0.0.0:9094
KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://host.docker.internal:9092,PLAINTEXT://localhost:9094
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,SASL_PLAINTEXT:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/kafka_server_jaas.conf
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
KAFKA_CLUSTER_ID: 'abcdefghijklmnopqrstuv'
KAFKA_LOG_DIRS: /tmp/kraft-combined-logs
volumes:
- ./kafka_server_jaas.conf:/etc/kafka/kafka_server_jaas.conf
EOF

docker compose up -d
```

## Verify SASL before involving the gateway

Run this from a separate container, not from inside the `kafka` container. It exercises the same
network path the data plane uses.

```bash
cat <<'EOF' > client.properties
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="kafka_user" password="kafka-password";
EOF

docker run --rm \
-v "$PWD/client.properties:/tmp/client.properties" \
apache/kafka:4.3.1 \
/opt/kafka/bin/kafka-broker-api-versions.sh \
--bootstrap-server host.docker.internal:9092 \
--command-config /tmp/client.properties
```

You should get a list of broker API versions ending in `)`, with `SaslHandshake(17)` and
`SaslAuthenticate(36)` present and no `SaslAuthenticationException`.

## Create the log topic

```bash
cat <<'EOF' > kafkactl.yaml
contexts:
direct:
brokers:
- localhost:9094
EOF

kafkactl -C kafkactl.yaml --context direct create topic kong-log
```

## An upstream to proxy to

The AI Agent needs a real HTTP upstream:

```bash
docker run -d --name log-upstream -p 8080:80 nginx:alpine
```

Confirm it's serving before continuing:

```bash
curl -i http://localhost:8080
```

## The Policy, plus an AI Agent to generate traffic

The Policy is global, so it logs every request. The AI Agent gives you a Route to send requests
through, and `config.url` points at the nginx container so the proxied request returns a `200`.

`ssl` is set to `false` here to simplify testing.

```bash
cat <<'EOF' > kafka-log.yaml
_defaults:
kongctl:
namespace: kafka-log-test

ai_gateway_policies:
- ref: kafka-log
ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
name: kafka-log
display_name: Kafka Log
type: kafka-log
enabled: true
global: true
config:
bootstrap_servers:
- host: host.docker.internal
port: 9092
topic: kong-log
authentication:
strategy: sasl
mechanism: PLAIN
user: kafka_user
password: kafka-password
security:
ssl: false

ai_gateway_agents:
- ref: kafka-log-agent
ai_gateway: !lookup {id: !env AI_GATEWAY_ID}
name: kafka-log-agent
display_name: "Kafka Log Agent"
type: http
enabled: true
config:
url: http://host.docker.internal:8080
route:
paths:
- /log-test
protocols:
- http
- https
methods:
- GET
- POST
strip_path: true
EOF

kongctl sync -f kafka-log.yaml --pat "$KONNECT_TOKEN"
```

## Generate traffic, then read the topic

Unlike Kafka Consume, this Policy produces to Kafka. Send requests first, then consume.

Allow a few seconds after `kongctl sync` for the data plane to pick up the new configuration.
Requests sent before it does return `502`.

Generate a few log entries:

```bash
curl -s -o /dev/null "$KONNECT_PROXY_URL/log-test"
curl -s -o /dev/null -X POST "$KONNECT_PROXY_URL/log-test" \
-H 'Content-Type: application/json' \
-d '{"hello":"world"}'
```

Read them back:

```bash
kafkactl -C kafkactl.yaml --context direct consume kong-log --from-beginning --exit | jq .
```

`config.producer_async` defaults to `true`, so allow a second or two before the entries appear.

The `GET` is logged with `response.status: 200`. The `POST` is logged with `405`, because nginx
rejects POST to a static file. Both are logged either way, since the Policy logs regardless of the
response status.

A trimmed entry:

```json
{
"source": "kong",
"client_ip": "192.168.97.1",
"started_at": 1787579347089,
"workspace_name": "default",
"upstream_uri": "/",
"upstream_status": "200",
"request": {
"method": "GET",
"uri": "/log-test",
"url": "http://localhost:8000/log-test",
"size": 86,
"querystring": {},
"headers": {
"host": "localhost:8000",
"user-agent": "curl/8.17.0",
"accept": "*/*"
}
},
"response": {
"status": 200,
"size": 853,
"headers": {
"content-type": "text/html",
"server": "kong/2.0.2-ai-gateway"
}
},
"latencies": {
"kong": 1,
"proxy": 4,
"request": 5
},
"route": {
"name": "kafka-log-agent-route",
"paths": ["/log-test"],
"methods": ["GET", "POST"],
"strip_path": true
},
"service": {
"name": "kafka-log-agent",
"host": "host.docker.internal",
"port": 8080,
"protocol": "http"
},
"tries": []
}
```

Open app/_ai_gateway_policies/kafka-log/index.md and compare the existing example with the supplied local setup steps. Follow the setup commands through topic creation, traffic generation, and consumption to check the sequence; done means the page is a clear, complete how-to that readers can follow locally and verify from the Kafka output.

Written by the indexing model from the issue text.