kafka-logger

Sends request logs to a Kafka topic.

Category ObservabilityAdd it to Route, Service, Consumer, Consumer group, Global rule, Plugin configEngine plugin kafka-loggerPriority 403

What it does

kafka-logger sends the records of the requests that reach the route or other object it is added to, to a Kafka topic. It needs the brokers’ addresses and the topic; the records are collected on the gateway node and sent in batches, so one message can carry several records.

Without a Log format, the record carries the request with its headers, the response status and headers, the latencies, the route, the service, the consumer and the client address. A Log format of your own, here or in the environment-wide settings, replaces that record with the fields you name, and Log format extra adds fields to the default record only.

When to use it

• Request records have to reach the platform the rest of the company already reads from Kafka, without a collector in between.
• Each record needs fields of your own, a header or the consumer name for example, instead of the default record.
• The brokers ask for a SASL user and password, or for an encrypted connection.

Not for a plain HTTP collector or a webhook — see http-logger (Sends request logs to an HTTP endpoint in batches), a RocketMQ topic — see rocketmq-logger (Sends request logs to a RocketMQ topic) or a syslog server — see syslog (Sends request logs to a syslog server).

Guides

Pages that walk through a job with kafka-logger, step by step.

• Send request records to KafkaSend a record of every request a route takes to a Kafka topic.

Set it up

Write request records to a Kafka topic

Under Basics fill Brokers as in the JSON below and set Kafka topic to gateway-access. Under Advanced leave Required acks at 1.

The console's Add plugin panel, filled with this example.

kafka-logger · as the console saves it

{
  "brokers": [
    {
      "host": "kafka-1.internal",
      "port": 9092
    },
    {
      "host": "kafka-2.internal",
      "port": 9092
    }
  ],
  "kafka_topic": "gateway-access",
  "required_acks": 1
}

More scenarios

Choose the fields every record carries and how many may wait for the brokersenvironment-wide settings

Fill Log format as in the JSON below and set Pending entry limit to 16384.

kafka-logger environment-wide settings · as the console saves it

{
  "log_format": {
    "client": "$remote_addr",
    "route": "$route_id",
    "status": "$status",
    "at": "$time_iso8601"
  },
  "max_pending_entries": 16384
}

Things to know

• Batch size — A batch goes out when it holds Batch size records, when its first record has waited Buffer duration, or when nothing new arrived for Inactive timeout.
• Cluster name — Every kafka-logger with the same Cluster name sends through one set of brokers, the first the gateway connected to, so give each set of brokers its own number.
• Log format — To keep the request's headers, and the keys or cookies they carry, out of the topic, set a Log format of your own: it replaces the default record with the fields you name.
• Pending entry limit — While more entries than this wait to be sent, new ones are dropped rather than queued.
• Brokers — Name the brokers either under Brokers or under the older Broker list, never both and never neither; either way the save is refused.
• Broker list — A port here is a number: a quoted port or a $env:// reference is refused when you save.
• Meta format — Under origin the request is sent as raw text and no Log format is read.
• Include request body when — The console’s form draws one level of conditions; a value that nests deeper is kept as it is and edited in Raw JSON.
• Include response body when — The console’s form draws one level of conditions; a value that nests deeper is kept as it is and edited in Raw JSON.
• Log format extra — Added to the default record only: while a log format is set, on this object or in the plugin's metadata, the engine writes that format alone and reads nothing here.

Settings

The fields as the console shows them, section by section.

Basics

Brokersbrokersarray of objectsRequired, or Broker list instead

at least 1 item

Each item:

Hostbrokers[].hoststringRequired

example kafka-1.internal

Portbrokers[].portintegerRequired

1 to 65535 · example 9092

SASL configbrokers[].sasl_configobject

Passwordbrokers[].sasl_config.passwordstringRequired

Stored on the engine and shown only to roles that manage this object.

Userbrokers[].sasl_config.userstringRequired

Mechanismbrokers[].sasl_config.mechanismstringdefault PLAIN

one of PLAIN, SCRAM-SHA-256, SCRAM-SHA-512

Kafka topickafka_topicstringRequired

Behavior

Keykeystring

Meta formatmeta_formatstringdefault default

one of default, origin

Timeouttimeoutintegerdefault 3

at least 1 · in seconds

TLStlsobject

Turn on Verify to connect to the brokers over TLS.

Verifytls.verifybooleandefault false

On, the gateway connects to the brokers over TLS and checks their certificate; a certificate it does not trust stops every record.

Log formatlog_formatobject

example {"host":"$host","client":"$remote_addr","status":"$status","at":"$time_iso8601"}

Log format extralog_format_extraobject

example {"zone":"eu-west","gateway":"$hostname"}

Include request bodyinclude_req_bodybooleandefault false

Include request body wheninclude_req_body_exprarray

at least 1 item · conditions on request variables: see Conditions on request variables

Include response bodyinclude_resp_bodybooleandefault false

Include response body wheninclude_resp_body_exprarray

at least 1 item · conditions on request variables: see Conditions on request variables

Request body limitmax_req_body_bytesintegerdefault 524288

at least 1

Response body limitmax_resp_body_bytesintegerdefault 524288

at least 1

Advanced

Batch sizebatch_max_sizeintegerdefault 1000

at least 1

Buffer durationbuffer_durationintegerdefault 60

at least 1 · in seconds

Inactive timeoutinactive_timeoutintegerdefault 5

at least 1 · in seconds

Retriesmax_retry_countintegerdefault 0

at least 0

Retry delayretry_delayintegerdefault 1

at least 0 · in seconds

Producer typeproducer_typestringdefault async

one of async, sync

Required acksrequired_acksintegerdefault 1

one of -1, 1

Producer batch messagesproducer_batch_numintegerdefault 200

at least 1

Producer batch bytesproducer_batch_sizeintegerdefault 1048576

at least 0

Producer buffer limitproducer_max_bufferingintegerdefault 50000

at least 1

Producer time lingerproducer_time_lingerintegerdefault 1

at least 1 · in seconds

API versionapi_versionintegerdefault 1

one of 0, 1, 2

Cluster namecluster_nameintegerdefault 1

The number that names this set of brokers on the gateway. Every kafka-logger with the same Cluster name sends through one set of brokers, the first the gateway connected to, so give each set of brokers its own number.

at least 1

Meta refresh intervalmeta_refresh_intervalintegerdefault 30

at least 1 · in seconds

Broker listbroker_listobjectRequired, or Brokers instead

An object whose keys are any name, each with an integer as its value.

at least 1 key

Namenamestringdefault kafka logger

Environment-wide settings

Set once per environment, under Plugins › kafka-logger. Every kafka-logger in the environment uses these.

Log formatlog_formatobject

example {"host":"$host","client":"$remote_addr","status":"$status","at":"$time_iso8601"}

Log format extralog_format_extraobject

example {"zone":"eu-west","gateway":"$hostname"}

Pending entry limitmax_pending_entriesintegerdefault 8192

How many log entries may wait to be sent. While the backlog is above this number, new entries are dropped rather than queued.

at least 1

Every plugin also has Enabled, Priority, Filter, Error response and Pre-function: see Plugin execution settings.