wikimedia / analytics-kafkatee

Github mirror of "analytics/kafkatee" - our actual code is hosted with Gerrit (please see https://www.mediawiki.org/wiki/Developer_access for contributing

kafkatee - Apache Kafka consumer with multiple inputs and outputs

Copyright (c) 2014 Wikimedia Foundation

Copyright (c) 2014 Magnus Edenhill

Description

kafkatee consumes messages from one or more Kafka topics and writes the messages to one or more outputs - either command pipes or files.

It provides simple transformation from JSON to arbitrary string output controlled through configuration.

Each output has a configurable sample rate.

Features:

• Supported input types: Kafka consumer or piped command.
• Supported output types: Piped command or file.
• Configuration file syntax is backwards compatible with Wikimedia's udp2log
• Configurable output queue size
• Memory frugal: message payload is shared by all output queues
• Configured with configuration file
• Operates as a daemon (daemonization, pidfile)
• Closes/stops and reopens/restarts all inputs and outputs on SIGHUP.

Documentation

See kafkatee.conf.example for an annotated configuration file explaining available configuration properties.

Build

Dependencies

• librdkafka-dev >= 0.8.3
• libyajl-dev (yajl1 or yajl2)

Compile

make

Install

make install
# or:
DESTDIR=/alternate/buildroot make install

Run tests

make test

analytics-kafkatee's Issues

No new pipe input means kafkatee will not shutdown properly on SIGTERM

If there is no new input on a configured piped input, kafkatee will hang on an fgets() call in input.c This causes kafkatee to not clean up children and pidfile properly.

We should probably use select() with a timeout instead of fgets() to poll for new data to read. I just had a quick look at implementing this, but since the code in input_pipe_main() uses FILE * instead of fds for all of its error checking, it requires more than just modification of the while loop for reading.

I suppose this would be difficult, because select() might indicate that there is data ready to be read before a full line has been written to the piped input?