rogierslag/kafkalytics repository overview

⁠Kafkalytics

Kafkalytics enables you to analyze, log, and optimize your usage of Kafka⁠ topics within your application infrastructure.

⁠The problem

Within a microservice architecture, firing events turns out to be the most scalable way to let applications communicate with each other without introducing a large amount of coupling. However, one cannot easily infer how many messages are fired on a certain Kafka topic. Even worse, it is very likely that at some point you are unsure whether certain messages are even used. This causes too much traffic over Kafka, requiring infrastructure scaling.

In order to reduce this possible maintenance problem, Kafkalytics aims to analyze and log your messages. It does this by listening to some topics you have defined. Additionally clients will send events to Kafkalytics to notify when they consumed certain messages. This gives you an indication which types of messages are actually used, and which topics might need to be split due to the message count.

⁠Kafkalytics to the rescue!

Kafkalytics aims to solve this problem. Each message you send to a topic will be logged to Elasticsearch by Kafkalytics (Elasticsearch here is used as a poor mans Hadoop). Each time a message is actually used/consumed by a consumer, it will send a message to Kafkalytics to indicate this. This message includes the service name, the datetime of consumption, and the topic of which the message was consumed.

Kafkalytics will analyze your topics and messages, and give an indication on how your Kafka topics and messages can be optimized. Examples of optimization can be the following:

• Topic splitting (if a very small percentage of message of consumed by a few clients, and many messages are discarded as irrelevant),
• Message consumption delays (if messages are consumed relatively slowly, this might indicate an infrastructure problem)
• Unused messages (no consumers exist anymore for this type of message)

Kafkalytics will do a form of deep inspection on the package contents in order to perform these analyses.

⁠Performance

In order to improve performance, and reduce network overhead Kafkalytics (and its API clients) will always bulk send events. The thresholds for this are configurable.

Kafkalytics will bulk send ES events after a number of events has been saved in the internal queue. Once this threshold is reached it will bulk index these to Elasticsearch. You can continue sending new data into Kafkalytics while this is happening. The Elasticsearch threshold needs some tuning, based on the size of your messages: a large number of very large messages may cause the bulk to have a significant size, which can timeout when sending this. Generally you should aim to keep your messages small.

Kafkalytics clients will save a number of kafka events, and bulk send these over Kafka once the threshold is reached. Also here you can continue adding new events while the actual sending takes place.

⁠Prerequisites

Kafkalytics makes just a few assumptions about your message data.

1. You should use Kafka,
2. Your messages should be JSON encoded,
3. A message should include a message metadata⁠ object.

Apart from that Elasticsearch⁠ is used to save all data.

⁠Message metadata

Since messages can be fired over multiple topics, we want to log the id of any message. Therefore the producer should use a unique identifier. Apart from that, we want to compute any delays in message consumption, so additionally a datetime object is included.

The entire object has the following form:

{
  "id": "unique message id",
  "datetime": "iso8601 formatted date"
}

This means your entire message looks like:

{
  "message": {
    "id": "unique message id",
    "datetime": "iso8601 formatted date"
  },
  "someNumber": 7,
  "somestring": "...actual message content..."
}

⁠Elasticsearch

To log and aggregate Elasticsearch is a required backend. Kafkalytics needs one index to work on, which is configurable.

⁠Usage

Kafkalytics uses a client/server model. This is done in order to keep the client for your applications as light as possible. Apart from that, dependencies are minimized.

The client and the server communicate over Kafka, which you are already using (or this project makes little sense for you :wink:). You might need to modify your Kafka messages a bit in order for Kafkalytics to process them effectively. See the message metadata⁠ section on this.

⁠The server

1. Create an Elasticsearch index and set this in the configuration,
2. Create a Kafkalytics topic and set this in the configuration,
3. Pull the Docker image,
4. Create a configuration file called default.json with the options mentioned before,
5. Run the Docker image as docker run -v <your config file dir>:/opt/kafkalytics/config rogierslag/kafkalytics
6. Profit!

⁠The clients

Clients should push consumed messages to a certain kafka topic, which is configurable. API clients are available.

• NodeJS: https://github.com/rogierslag/kafkalytics-node⁠

⁠Gradual adoption

In order to support gradual adoption Kafkalytics will ignore messages which are not ready for Kafkalytics yet. This will allow your organization to start using Kafkalytics, but slowly add more message types over time.

• For producers: Any message not containing the message.id parameter will be ignored.
• For consumers: You can check whether the message.id property is present, and skip sending if that is not the case.

⁠Data analysis

still have to build this, see https://github.com/rogierslag/kafkalytics/issues/7⁠

⁠Running outside of Docker

You need the following packages on Ubuntu: make gcc g++.