The mod_kafka module enables ProFTPD support for sending log messages, as JSON, to Kafka brokers using the librdkafka client library.

This module is contained in the mod_kafka files for ProFTPD 1.3.x, and is not compiled by default. Installation instructions are discussed here. More examples of mod_kafka usage can be found here.

The most current version of mod_kafka can be found at:

  https://github.com/Castaglia/proftpd-mod_kafka

Author

Please contact TJ Saunders <tj at castaglia.org> with any questions, concerns, or suggestions regarding this module.

Directives

• KafkaBroker
• KafkaEngine
• KafkaLog
• KafkaLogOnEvent
• KafkaProperty

KafkaBroker

The KafkaServer directive is used to configure the addresses/ports of the initial Kafka brokers contacted by mod_kafka. For example:

  KafkaBroker 1.2.3.4 5.6.7.8:19092

  KafkaBroker [::ffff:1.2.3.4]:9092
  KafkaProperty broker.address.family any

KafkaEngine

The KafkaEngine directive enables or disables the mod_kafka module, and thus the configuration of Kafka support for the proftpd daemon.

KafkaLog

The KafkaLog directive is used to specify a log file for mod_kafka's reporting on a per-server basis. The file parameter given must be the full path to the file to use for logging.

Note that this path must not be to a world-writable directory and, unless AllowLogSymlinks is explicitly set to on (generally a bad idea), the path must not be a symbolic link.

KafkaLogOnEvent

The KafkaLogOnEvent directive configures the use of Kafka for logging. Whenever one of the comma-separated list of events occurs, mod_kafka will compose a JSON object, using the LogFormat named by format-name as a template for the fields to include in the JSON object. The JSON object of that event will then be published to a Kafka topic. Multiple KafkaLogOnEvent directives can be used, for different log formats for different events and different topics.

The optional topic parameter, if present, specifies the value to use as the topic name. If the topic name is not provided explicitly, the configured format-name is used as the topic name.

More on the use of Kafka logging, including a table showing how LogFormat variables are mapped to JSON object keys can be found here.

Example:

  LogFormat sessions "%{iso8601} %a"
  KafkaLogOnEvent CONNECT,DISCONNECT sessions

In addition to specific FTP commands, the events list can specify "ALL", for logging on all commands. Or it can include the "CONNECT" and "DISCONNECT" events, which can be useful for logging the start and end times of a session. Note that KafkaLogOnEvent does support the logging classes that the ExtendedLog directive supports.

KafkaProperty

The KafkaProperty directive is used to configure the property name and value of the common Kafka properties; see here.

Example:

  KafkaProperty socket.timeout.ms 30000
  KafkaProperty socket.keepalive.enable true
  KafkaProperty socket.nagle.disable true

Installation

  proftpd-dir/contrib/

  $ ./configure --with-modules=mod_kafka

  $ ./configure --enable-dso --with-shared=mod_kafka

  $ make
  $ make install

You may also need to tell configure how to find the librdkafka header and library files:

  $ ./configure --with-modules=mod_kafka \
    --with-includes=/path/to/librdkafka/include \
    --with-libraries=/path/to/librdkafka/lib

Usage

This example shows the use of Kafka logging for all commands:

  <IfModule mod_kafka.c>
    KafkaEngine on
    KafkaLog /var/log/ftpd/kafka.log
    KafkaBroker kafka:9092

    LogFormat kafka "%h %l %u %t \"%r\" %s %b"
    KafkaLogOnEvent ALL kafka
  </IfModule>

For cases where you need to use TLS when talking to your Kafka brokers, you configure the necessary TLS files via the KafkaProperty directive:

  <IfModule mod_kafka.c>
    KafkaEngine on
    KafkaLog /var/log/ftpd/kafka.log

    KafkaProperty ssl.ca.location /usr/local/etc/kafka/ca.pem
    KafkaProperty ssl.certificate.location /usr/local/etc/kafka/client.pem
    KafkaProperty ssl.key.location /usr/local/etc/kafka/client.pem

    # Set this to false if necessary
    KafkaProperty enable.ssl.certificate.verification true

    # Necessary for telling librdkafka to use TLS for the broker
    KafkaProperty security.protocol ssl

    # Kafka uses TLS on port 9093
    KafkaBroker ssl://kafka:9093

    LogFormat kafka "%h %l %u %t \"%r\" %s %b"
    KafkaLogOnEvent ALL kafka
  </IfModule>

Kafka Logging
When using Kafka logging, the following table shows how mod_kafka converts a LogFormat variable into the key names in the JSON logging objects:

In addition to the standard LogFormat variables, the mod_kafka module also adds a "connecting" key for events generated when a client first connects, and a "disconnecting" key for events generated when a client disconnects. These keys can be used for determining the start/finish events for a given session.

Here is an example of the JSON-formatted records generated, using the above example configuration:

  {"connecting":true,"timestamp":"2013-08-21 23:08:22,171"}
  {"command":"USER","timestamp":"2013-08-21 23:08:22,278"}
  {"user":"proftpd","command":"PASS","timestamp":"2013-08-21 23:08:22,305"}
  {"user":"proftpd","command":"PASV","timestamp":"2013-08-21 23:08:22,317"}
  {"user":"proftpd","command":"LIST","bytes_sent":432,"transfer_secs":4.211,"timestamp":"2013-08-21 23:08:22,329"}
  {"user":"proftpd","command":"QUIT","timestamp":"2013-08-21 23:08:22,336"}
  {"disconnecting":true,"user":"proftpd","timestamp":"2013-08-21 23:08:22,348"}

Another thing to notice is that the generated JSON object ignores the textual delimiters configured by the LogFormat directive; all that matters are the LogFormat variables which appear in the directive.