This is an automated email from the ASF dual-hosted git repository.
AlinsRan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/apisix.git
The following commit(s) were added to refs/heads/master by this push:
new b86a6c8ba feat(kafka-logger): add TLS support for Kafka brokers
(#13607)
b86a6c8ba is described below
commit b86a6c8bad0554260f58324442c9ce2f8f1af607
Author: JinHwanKim <[email protected]>
AuthorDate: Thu Jul 16 19:12:27 2026 +0900
feat(kafka-logger): add TLS support for Kafka brokers (#13607)
---
apisix/plugins/error-log-logger.lua | 21 +++-
apisix/plugins/kafka-logger.lua | 16 +++
ci/init-plugin-test-service.sh | 3 +
ci/pod/docker-compose.plugin.yml | 4 +
ci/pod/kafka/kafka-server/env/common.env | 7 +-
docs/en/latest/plugins/error-log-logger.md | 31 +++++
docs/en/latest/plugins/kafka-logger.md | 49 ++++++++
t/plugin/error-log-logger-kafka.t | 101 ++++++++++++++++
t/plugin/kafka-logger2.t | 177 +++++++++++++++++++++++++++++
t/plugin/security-warning.t | 111 ++++++++++++++++++
10 files changed, 518 insertions(+), 2 deletions(-)
diff --git a/apisix/plugins/error-log-logger.lua
b/apisix/plugins/error-log-logger.lua
index ee530e14a..0551c7001 100644
--- a/apisix/plugins/error-log-logger.lua
+++ b/apisix/plugins/error-log-logger.lua
@@ -108,6 +108,16 @@ local metadata_schema = {
},
uniqueItems = true,
},
+ tls = {
+ type = "object",
+ description = "tls config for connecting to kafka brokers",
+ properties = {
+ verify = {
+ type = "boolean",
+ default = false,
+ },
+ },
+ },
kafka_topic = {type = "string"},
producer_type = {
type = "string",
@@ -185,7 +195,12 @@ local _M = {
function _M.check_schema(conf, schema_type)
if schema_type == core.schema.TYPE_METADATA then
- return core.schema.check(metadata_schema, conf)
+ local ok, err = core.schema.check(metadata_schema, conf)
+ if not ok then
+ return nil, err
+ end
+ core.utils.check_tls_bool({"kafka.tls.verify"}, conf, plugin_name)
+ return true
end
local check = {"skywalking.endpoint_addr", "clickhouse.endpoint_addr"}
@@ -382,6 +397,10 @@ local function send_to_kafka(log_message)
broker_config["producer_type"] = config.kafka.producer_type
broker_config["required_acks"] = config.kafka.required_acks
broker_config["refresh_interval"] = config.kafka.meta_refresh_interval *
1000
+ if config.kafka.tls then
+ broker_config["ssl"] = true
+ broker_config["ssl_verify"] = config.kafka.tls.verify
+ end
-- reuse producer via kafka_prod_lrucache to avoid unbalanced partitions
of messages in kafka
local prod, err = kafka_prod_lrucache(plugin_name, metadata.modifiedIndex,
diff --git a/apisix/plugins/kafka-logger.lua b/apisix/plugins/kafka-logger.lua
index 845fa684b..b9b254e2f 100644
--- a/apisix/plugins/kafka-logger.lua
+++ b/apisix/plugins/kafka-logger.lua
@@ -89,6 +89,16 @@ local schema = {
},
uniqueItems = true,
},
+ tls = {
+ type = "object",
+ description = "tls config for connecting to kafka brokers",
+ properties = {
+ verify = {
+ type = "boolean",
+ default = false,
+ },
+ },
+ },
kafka_topic = {type = "string"},
producer_type = {
type = "string",
@@ -179,6 +189,8 @@ function _M.check_schema(conf, schema_type)
if not ok then
return nil, err
end
+
+ core.utils.check_tls_bool({"tls.verify"}, conf, plugin_name)
return log_util.check_log_schema(conf)
end
@@ -279,6 +291,10 @@ function _M.log(conf, ctx)
broker_config["flush_time"] = conf.producer_time_linger * 1000
broker_config["refresh_interval"] = conf.meta_refresh_interval * 1000
broker_config["api_version"] = conf.api_version
+ if conf.tls then
+ broker_config["ssl"] = true
+ broker_config["ssl_verify"] = conf.tls.verify
+ end
local prod, err = core.lrucache.plugin_ctx(lrucache, ctx, nil,
create_producer,
broker_list, broker_config,
conf.cluster_name)
diff --git a/ci/init-plugin-test-service.sh b/ci/init-plugin-test-service.sh
index 6ef72f1df..7c65a8af3 100755
--- a/ci/init-plugin-test-service.sh
+++ b/ci/init-plugin-test-service.sh
@@ -94,6 +94,9 @@ after() {
before() {
# download keycloak cas provider
sudo wget -q
https://github.com/jacekkow/keycloak-protocol-cas/releases/download/18.0.2/keycloak-protocol-cas-18.0.2.jar
-O /opt/keycloak-protocol-cas-18.0.2.jar
+
+ # generating SSL certificates for Kafka
+ sudo keytool -genkeypair -keyalg RSA -dname "CN=127.0.0.1" -alias
127.0.0.1 -keystore ./ci/pod/kafka/kafka-server/selfsigned.jks -validity 365
-keysize 2048 -storepass changeit
}
case $1 in
diff --git a/ci/pod/docker-compose.plugin.yml b/ci/pod/docker-compose.plugin.yml
index e66877ca6..81b2da0df 100644
--- a/ci/pod/docker-compose.plugin.yml
+++ b/ci/pod/docker-compose.plugin.yml
@@ -175,11 +175,15 @@ services:
restart: unless-stopped
ports:
- "9092:9092"
+ - "9093:9093"
depends_on:
- zookeeper-server1
- zookeeper-server2
networks:
kafka_net:
+ volumes:
+ -
./ci/pod/kafka/kafka-server/selfsigned.jks:/opt/bitnami/kafka/config/certs/kafka.keystore.jks:ro
+ -
./ci/pod/kafka/kafka-server/selfsigned.jks:/opt/bitnami/kafka/config/certs/kafka.truststore.jks:ro
kafka-server2:
image: bitnamilegacy/kafka:2.8.1
diff --git a/ci/pod/kafka/kafka-server/env/common.env
b/ci/pod/kafka/kafka-server/env/common.env
index 06200b9b0..2414fd2c5 100644
--- a/ci/pod/kafka/kafka-server/env/common.env
+++ b/ci/pod/kafka/kafka-server/env/common.env
@@ -1,3 +1,8 @@
ALLOW_PLAINTEXT_LISTENER=yes
KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
-KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092
+KAFKA_CFG_LISTENERS=PLAINTEXT://0.0.0.0:9092,SSL://0.0.0.0:9093
+KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://127.0.0.1:9092,SSL://127.0.0.1:9093
+KAFKA_CFG_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=
+KAFKA_CFG_SSL_KEYSTORE_LOCATION=/opt/bitnami/kafka/config/certs/kafka.keystore.jks
+KAFKA_CFG_SSL_KEYSTORE_PASSWORD=changeit
+KAFKA_CFG_SSL_KEY_PASSWORD=changeit
diff --git a/docs/en/latest/plugins/error-log-logger.md
b/docs/en/latest/plugins/error-log-logger.md
index 2af9dac59..bd46666d6 100644
--- a/docs/en/latest/plugins/error-log-logger.md
+++ b/docs/en/latest/plugins/error-log-logger.md
@@ -66,6 +66,8 @@ There are no attributes to configure this Plugin on Routes or
Services. All conf
| kafka.brokers[].sasl_config.mechanism | string | False | PLAIN
| ["PLAIN"]
| The mechanism of SASL configuration.
|
| kafka.brokers[].sasl_config.user | string | True |
|
| The user of SASL configuration. Required if
`sasl_config` is present.
|
| kafka.brokers[].sasl_config.password | string | True |
|
| The password of SASL configuration. Required if
`sasl_config` is present.
|
+| kafka.tls | object | False |
|
| TLS configuration for connecting to Kafka brokers.
|
+| kafka.tls.verify | boolean | False | false
|
| If true, verify the Kafka broker TLS certificate.
|
| kafka.kafka_topic | string | True |
|
| Target topic to push the logs for organization.
|
| kafka.producer_type | string | False | async
| ["async", "sync"]
| Message sending mode of the producer.
|
| kafka.required_acks | integer | False | 1
| [-1, 0, 1]
| Number of acknowledgements the leader needs to
receive for the producer to consider the request complete. See [Apache Kafka
documentation](https://kafka.apache.org/documentation/#producerconfigs_acks)
for more. |
@@ -238,3 +240,32 @@ curl
"http://127.0.0.1:9180/apisix/admin/plugin_metadata/error-log-logger"; -X PU
"inactive_timeout": 1
}'
```
+
+### Send Logs to TLS-Enabled Kafka Brokers
+
+The following example demonstrates how to configure the `error-log-logger`
Plugin to send error logs to TLS-enabled Kafka brokers.
+
+Configure the Plugin metadata with the Kafka broker details and TLS
configuration:
+
+```shell
+curl "http://127.0.0.1:9180/apisix/admin/plugin_metadata/error-log-logger"; -X
PUT \
+ -H "X-API-KEY: ${admin_key}" \
+ -d '{
+ "kafka": {
+ "brokers": [
+ {
+ "host": "kafka.example.com",
+ "port": 9093
+ }
+ ],
+ "kafka_topic": "apisix-error-logs",
+ "tls": {
+ "verify": true
+ }
+ },
+ "level": "ERROR",
+ "inactive_timeout": 1
+ }'
+```
+
+When using self-signed certificates, set `tls.verify` to `false` to skip
certificate verification.
diff --git a/docs/en/latest/plugins/kafka-logger.md
b/docs/en/latest/plugins/kafka-logger.md
index f57081183..22ecb3c3f 100644
--- a/docs/en/latest/plugins/kafka-logger.md
+++ b/docs/en/latest/plugins/kafka-logger.md
@@ -49,6 +49,8 @@ It might take some time to receive the log data. It will be
automatically sent a
| brokers.sasl_config.mechanism | string | False | "PLAIN" |
["PLAIN", "SCRAM-SHA-256", "SCRAM-SHA-512"] | The mechanism of SASL
config.
|
| brokers.sasl_config.user | string | True | |
| The user of `sasl_config`.
Required if `sasl_config` is configured.
|
| brokers.sasl_config.password | string | True | |
| The password of `sasl_config`.
Required if `sasl_config` is configured.
|
+| tls | object | False | |
| TLS configuration for
connecting to Kafka brokers.
|
+| tls.verify | boolean | False | false |
| If true, verify the Kafka
broker TLS certificate.
|
| kafka_topic | string | True | |
| Target topic to push the logs.
|
| producer_type | string | False | async |
["async", "sync"] | Message sending mode of the
producer.
|
| required_acks | integer | False | 1 | [1,
-1] | Number of acknowledgements the
leader needs to receive for the producer to consider the request complete. This
controls the durability of the sent records. The attribute follows the same
configuration as the Kafka `acks` attribute. `required_acks` cannot be 0. See
[Apache Kafka
documentation](https://kafka.apache.org/documentation/#producerconfigs_acks)
for more. |
@@ -487,3 +489,50 @@ If you have customized the `log_format` in addition to
setting `include_req_body
```
:::
+
+### Log to TLS-Enabled Kafka Brokers
+
+The following example demonstrates how to connect to TLS-enabled Kafka
brokers, such as AWS MSK.
+
+Create a Route with `kafka-logger` and configure the `tls` attribute to
connect to the TLS-enabled Kafka broker:
+
+```shell
+curl "http://127.0.0.1:9180/apisix/admin/routes"; -X PUT \
+ -H "X-API-KEY: ${admin_key}" \
+ -d '{
+ "id": "kafka-logger-tls-route",
+ "uri": "/get",
+ "plugins": {
+ "kafka-logger": {
+ "brokers": [
+ {
+ "host": "kafka.example.com",
+ "port": 9093
+ }
+ ],
+ "kafka_topic": "test2",
+ "key": "key1",
+ "batch_max_size": 1,
+ "tls": {
+ "verify": true
+ }
+ }
+ },
+ "upstream": {
+ "nodes": {
+ "httpbin.org:80": 1
+ },
+ "type": "roundrobin"
+ }
+ }'
+```
+
+When using self-signed certificates, set `tls.verify` to `false` to skip
certificate verification:
+
+```json
+{
+ "tls": {
+ "verify": false
+ }
+}
+```
diff --git a/t/plugin/error-log-logger-kafka.t
b/t/plugin/error-log-logger-kafka.t
index 0ae6a52ac..1ede14cd1 100644
--- a/t/plugin/error-log-logger-kafka.t
+++ b/t/plugin/error-log-logger-kafka.t
@@ -270,3 +270,104 @@ true
true
--- no_error_log
[alert]
+
+
+
+=== TEST 8: tls schema validation - valid tls config
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local plugin = require("apisix.plugins.error-log-logger")
+ local ok, err = plugin.check_schema(
+ {
+ kafka = {
+ brokers = {
+ {
+ host = "127.0.0.1",
+ port = 9093
+ }
+ },
+ kafka_topic = "test2",
+ tls = { verify = false }
+ }
+ },
+ core.schema.TYPE_METADATA
+ )
+ if not ok then
+ ngx.say(err)
+ end
+ ngx.say("done")
+ }
+ }
+--- response_body
+done
+
+
+
+=== TEST 9: tls schema validation - wrong type for verify
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local plugin = require("apisix.plugins.error-log-logger")
+ local ok, err = plugin.check_schema(
+ {
+ kafka = {
+ brokers = {
+ {
+ host = "127.0.0.1",
+ port = 9093
+ }
+ },
+ kafka_topic = "test2",
+ tls = { verify = "abc" }
+ }
+ },
+ core.schema.TYPE_METADATA
+ )
+ if not ok then
+ ngx.say(err)
+ end
+ ngx.say("done")
+ }
+ }
+--- response_body
+property "kafka" validation failed: property "tls" validation failed: property
"verify" validation failed: wrong type: expected boolean, got string
+done
+
+
+
+=== TEST 10: put metadata with kafka tls and log an error message - send via
TLS
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local t = require("lib.test_admin").test
+ local code, body =
t('/apisix/admin/plugin_metadata/error-log-logger',
+ ngx.HTTP_PUT,
+ [[{
+ "kafka": {
+ "brokers": [{
+ "host": "127.0.0.1",
+ "port": 9093
+ }],
+ "kafka_topic": "test2",
+ "meta_refresh_interval": 1,
+ "tls": {"verify": false}
+ },
+ "level": "ERROR",
+ "inactive_timeout": 1
+ }]]
+ )
+ ngx.sleep(2)
+ core.log.error("this is a error message for tls test.")
+ }
+ }
+--- error_log eval
+[qr/this is a error message for tls test/,
+qr/sending a batch logs to kafka brokers:
\[\{"host":"127.0.0.1","port":9093\}\]/,
+qr/send data to kafka: .*this is a error message for tls test/]
+--- no_error_log
+failed to do SSL handshake
+--- wait: 3
diff --git a/t/plugin/kafka-logger2.t b/t/plugin/kafka-logger2.t
index c1fccb7eb..8ba42999b 100644
--- a/t/plugin/kafka-logger2.t
+++ b/t/plugin/kafka-logger2.t
@@ -1123,3 +1123,180 @@ hello world
--- error_log eval
qr/send data to kafka: \{.*"body":"abcdef"/
--- wait: 2
+
+
+
+=== TEST 28: tls schema validation - valid tls config
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = { verify = false }
+ })
+ if not ok then
+ ngx.say(err)
+ end
+ ngx.say("done")
+ }
+ }
+--- response_body
+done
+
+
+
+=== TEST 29: tls schema validation - without tls (backward compatibility)
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9092}},
+ kafka_topic = "test"
+ })
+ if not ok then
+ ngx.say(err)
+ end
+ ngx.say("done")
+ }
+ }
+--- response_body
+done
+
+
+
+=== TEST 30: tls schema validation - wrong type for verify
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = { verify = "abc" }
+ })
+ if not ok then
+ ngx.say(err)
+ end
+ ngx.say("done")
+ }
+ }
+--- response_body
+property "tls" validation failed: property "verify" validation failed: wrong
type: expected boolean, got string
+done
+
+
+
+=== TEST 31: set route with tls config to SSL port
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins":{
+ "kafka-logger":{
+ "brokers":[
+ {
+ "host":"127.0.0.1",
+ "port":9093
+ }],
+ "kafka_topic":"test2",
+ "producer_type":"sync",
+ "key":"key1",
+ "timeout":1,
+ "batch_max_size":1,
+ "include_req_body": true,
+ "tls":{"verify":false}
+ }
+ },
+ "upstream":{
+ "nodes":{
+ "127.0.0.1:1980":1
+ },
+ "type":"roundrobin"
+ },
+ "uri":"/hello"
+ }]]
+ )
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 32: hit route, send data to kafka via TLS successfully
+--- request
+POST /hello?name=qwerty
+abcdef
+--- response_body
+hello world
+--- error_log eval
+qr/send data to kafka: \{.*"body":"abcdef"/
+--- no_error_log
+[error]
+--- wait: 2
+
+
+
+=== TEST 33: set route with tls verify enabled against self-signed broker
+--- config
+ location /t {
+ content_by_lua_block {
+ local t = require("lib.test_admin").test
+ local code, body = t('/apisix/admin/routes/1',
+ ngx.HTTP_PUT,
+ [[{
+ "plugins":{
+ "kafka-logger":{
+ "brokers":[
+ {
+ "host":"127.0.0.1",
+ "port":9093
+ }],
+ "kafka_topic":"test2",
+ "producer_type":"sync",
+ "key":"key1",
+ "timeout":1,
+ "batch_max_size":1,
+ "include_req_body": true,
+ "tls":{"verify":true}
+ }
+ },
+ "upstream":{
+ "nodes":{
+ "127.0.0.1:1980":1
+ },
+ "type":"roundrobin"
+ },
+ "uri":"/hello"
+ }]]
+ )
+ if code >= 300 then
+ ngx.status = code
+ end
+ ngx.say(body)
+ }
+ }
+--- response_body
+passed
+
+
+
+=== TEST 34: hit route, tls verify rejects the self-signed certificate
+--- request
+POST /hello?name=qwerty
+abcdef
+--- response_body
+hello world
+--- error_log
+failed to do SSL handshake with 127.0.0.1:9093
+--- wait: 2
diff --git a/t/plugin/security-warning.t b/t/plugin/security-warning.t
index 0dca62e7b..ed22f1506 100644
--- a/t/plugin/security-warning.t
+++ b/t/plugin/security-warning.t
@@ -574,3 +574,114 @@ Using loki-logger endpoint_addrs with no TLS is a
security risk
done
--- no_error_log
Using loki-logger endpoint_addrs with no TLS is a security risk
+
+
+
+=== TEST 21: kafka-logger with tls verify disabled
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = { verify = false }
+ })
+ ngx.say(ok and "done" or err)
+ }
+ }
+--- response_body
+done
+--- error_log
+Keeping tls.verify disabled in kafka-logger configuration is a security risk
+
+
+
+=== TEST 22: kafka-logger with tls verify enabled
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = { verify = true }
+ })
+ ngx.say(ok and "done" or err)
+ }
+ }
+--- response_body
+done
+--- no_error_log
+Keeping tls.verify disabled in kafka-logger configuration is a security risk
+
+
+
+=== TEST 23: error-log-logger with kafka tls verify disabled (metadata)
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local plugin = require("apisix.plugins.error-log-logger")
+
+ local ok, err = plugin.check_schema({
+ kafka = {
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = { verify = false }
+ }
+ }, core.schema.TYPE_METADATA)
+ ngx.say(ok and "done" or err)
+ }
+ }
+--- response_body
+done
+--- error_log
+Keeping kafka.tls.verify disabled in error-log-logger configuration is a
security risk
+
+
+
+=== TEST 24: error-log-logger with kafka tls but verify omitted (defaults to
false)
+--- config
+ location /t {
+ content_by_lua_block {
+ local core = require("apisix.core")
+ local plugin = require("apisix.plugins.error-log-logger")
+
+ local ok, err = plugin.check_schema({
+ kafka = {
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = {}
+ }
+ }, core.schema.TYPE_METADATA)
+ ngx.say(ok and "done" or err)
+ }
+ }
+--- response_body
+done
+--- error_log
+Keeping kafka.tls.verify disabled in error-log-logger configuration is a
security risk
+
+
+
+=== TEST 25: kafka-logger with tls but verify omitted (defaults to false)
+--- config
+ location /t {
+ content_by_lua_block {
+ local plugin = require("apisix.plugins.kafka-logger")
+
+ local ok, err = plugin.check_schema({
+ brokers = {{host = "127.0.0.1", port = 9093}},
+ kafka_topic = "test",
+ tls = {}
+ })
+ ngx.say(ok and "done" or err)
+ }
+ }
+--- response_body
+done
+--- error_log
+Keeping tls.verify disabled in kafka-logger configuration is a security risk
