## ProFTPD module `mod_kafka`

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

Syntax: Kafka host[:port] ...
Default: None
Context: server config, `<VirtualHost>`, `<Global>`
Module: mod_kafka
Compatibility: 1.3.6rc5 and later

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

or, for an IPv6 address, make sure the IPv6 address is enclosed in square brackets:

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

### KafkaEngine

Syntax: KafkaEngine on|off
Default: KafkaEngine off
Context: server config, `<VirtualHost>`, `<Global>`
Module: mod_kafka
Compatibility: 1.3.6rc5 and later

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

### KafkaLog

Syntax: KafkaLog path|"none"
Default: None
Context: server config, `<VirtualHost>`, `<Global>`
Module: mod_kafka
Compatibility: 1.3.6rc5 and later

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

Syntax: KafkaLogOnEvent "none"|events format-name [topic ...]
Default: None
Context: server config, `<VirtualHost>`, `<Global>`, `<Anonymous>`, `<Directory>`
Module: mod_kafka
Compatibility: 1.3.7rc1 and later

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

Syntax: KafkaProperty name value
Default: None
Context: server config, `<VirtualHost>`, `<Global>`
Module: mod_kafka
Compatibility: 1.3.7rc1 and later

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

To install `mod_kafka`, copy the `mod_kafka` files into:

      proftpd-dir/contrib/

after unpacking the latest proftpd-1.3.x source code. For including `mod_kafka` as a staticly linked module:

      $ ./configure --with-modules=mod_kafka

To build `mod_kafka` as a DSO module:

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

Then follow the usual steps:

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

`LogFormat` Variable | Key
`%A` | anon_password
`%a` | remote_ip
`%b` | bytes_sent
`%c` | connection_class
`%D` | dir_path
`%d` | dir_name
`%E` | session_end_reason
`%{epoch}` | Unix timestamp, in seconds since Jan 1 1970.
`%{name}e` | ENV:name
`%F` | transfer_path
`%f` | file
`%{file-modified}` | file_modified
`%g` | group
`%{gid}` | gid
`%H` | server_ip
`%h` | remote_dns
`%I` | session_bytes_rcvd
`%{iso8601}` | timestamp
`%J` | command_params
`%L` | local_ip
`%l` | identd_user
`%m` | command
`%{microsecs}` | microsecs
`%{millisecs}` | millisecs
`%{note:name}` | NOTE:name
`%O` | session_bytes_sent
`%P` | pid
`%p` | local_port
`%{protocol}` | protocol
`%r` | raw_command
`%S` | response_msg
`%s` | response_code
`%T` | transfer_secs
`%t` | local_time
`%{transfer-failure}` | transfer_failure
`%{transfer-status}` | transfer_status
`%U` | original_user
`%u` | user
`%{uid}` | uid
`%V` | server_dns
`%v` | server_name
`%{version}` | server_version
`%w` | rename_from

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

Notice that for a given event, not all of the `LogFormat` variables are filled in. If `mod_kafka` determines that a given `LogFormat` variable has no value for the logged event, it will simply omit that variable from the JSON object.

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.

© Copyright 2017-2022 TJ Saunders
All Rights Reserved