Metadata-Version: 2.1
Name: kafkareport
Version: 0.0.1
Summary: Reporting tool for kafka topics with retentions, sizes, earliest and latest message timestamps
Home-page: https://github.com/newvoll/kafkareport
License: MIT
Author: olmouse
Author-email: olmouse@newvoll.net
Requires-Python: >=3.9,<4.0
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Dist: confluent_kafka (>=2.4.0,<3.0.0)
Requires-Dist: kafka-python (>=2.0.2,<3.0.0)
Requires-Dist: prettytable (>=3.10.0,<4.0.0)
Project-URL: Documentation, https://kafkareport.readthedocs.io/en/latest/index.html
Project-URL: Issues, https://github.com/newvoll/kafkareport/issues
Project-URL: Repository, https://github.com/newvoll/kafkareport
Description-Content-Type: text/markdown

A reporting tool for kafka topics.

Show any or all of the following attributes for each topic in a cluster:
* sizes (`DescribeLogDirsRequest`)
* retention policies
* earliest/latest message timestamps (watermarks)

With larger teams and installations, misconfiguration often results in
lots of resource waste. Auditing with this tool has helped.

# Installation

`pip install kafkareport`

# Usage

All times should be UTC.

## Configuration
A dict passed to confluent_kafka's consumer, so it should follow the
[same
format](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#pythonclient-configuration)
except for group.id. It is automatically set. See `tests/helper_files/env.json` for a sample conf file. `-h` for help.

> [!NOTE]
> Tested only with `SASL_SSL` and `SCRAM_SHA_512` u/p on AWS MSK. See
> [Addenda](#Addenda) for details


## CLI
```
% kafkareport ~/aws.json --csv report.csv
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+
| topic          | bytes | earliest                         | latest                           | retention.ms | retention.bytes | delete.retention.ms |
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+
| kafkareportdue | 2330  | 2024-06-30 20:04:44.486000+00:00 | 2024-06-30 20:04:46.041000+00:00 | 16800001     | 16800001        | 16800001            |
| kafkareportuno | 2268  | 2024-06-30 20:04:42.880000+00:00 | 2024-06-30 20:04:44.431000+00:00 | 16800001     | 16800001        | 16800001            |
+----------------+-------+----------------------------------+----------------------------------+--------------+-----------------+---------------------+

% cat report.csv
topic,bytes,earliest,latest,retention.ms,retention.bytes,delete.retention.ms
kafkareportdue,2330,2024-06-30 20:04:44.486000+00:00,2024-06-30 20:04:46.041000+00:00,16800001,16800001,16800001
kafkareportuno,2268,2024-06-30 20:04:42.880000+00:00,2024-06-30 20:04:44.431000+00:00,16800001,16800001,16800001
```

## Lib
```
>>> from kafkareport import KafkaReport

>>> conf = {"bootstrap.servers": "localhost:9092", "ssl.endpoint.identification.algorithm": "none"}

>>> report = KafkaReport(conf)

>>> report.get_topicnames()
dict_keys(['kafkareportdue', 'kafkareportuno', '__consumer_offsets'])

>>> for topic in report.get_topicnames():
...     print(topic)
...     report.retentions(topic)
...     report.watermarks(topic)
...
kafkareportdue
{'retention.ms': 16800001, 'retention.bytes': 16800001, 'delete.retention.ms': 16800001}
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 57, 554000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 59, 99000, tzinfo=datetime.timezone.utc)}
kafkareportuno
{'retention.ms': 16800001, 'retention.bytes': 16800001, 'delete.retention.ms': 16800001}
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 55, 975000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 57, 503000, tzinfo=datetime.timezone.utc)}
__consumer_offsets
{'retention.ms': 604800000, 'retention.bytes': -1, 'delete.retention.ms': 86400000}
{'earliest': '', 'latest': ''}

>>> report.topic_sizes()
[{'topic': 'kafkareportuno', 'bytes': 2690}, {'topic': 'kafkareportdue', 'bytes': 2328}, {'topic': '__consumer_offsets', 'bytes': 0}]

>>> report.watermarks("kafkareportuno")
{'earliest': datetime.datetime(2024, 6, 30, 20, 8, 55, 975000, tzinfo=datetime.timezone.utc), 'latest': datetime.datetime(2024, 6, 30, 20, 8, 57, 503000, tzinfo=datetime.timezone.utc)}

```

# Development
This is a [poetry](https://python-poetry.org/) project, so it should
be butter once you get that sorted. Install
[pre-commit](https://pre-commit.com/) for black on commit, lint on
push. Couldn't figure `pytype` into `pre-commit`.

# Testing
Testing uses [localstack](https://www.localstack.cloud/), as you can
see in the [Github
actions](https://github.com/newvoll/kafkareport/actions). `docker-compose
up` should do the trick.

`pytest` can use `--conf=/some/file.json` instead of default for
localstack. This will manipulate pytest.topics on the kafka servers.

## Testing Gotchas
* Occasionally, e.g. on laptop wake, the kafka container will be in a
  weird state. `docker-compose down --remove-orphans && docker-compose
  up` and always did the trick.

# Addenda
* Watermarks use a thread for each topic partition, but it can still take a while. `-v` for gory details along the way, `KafkaReport(debug=True)` for the lib.
* This project started with
  [confluent_kafka](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/index.html#). At
  time of first writing, neither
  [kafka-python](https://kafka-python.readthedocs.io/en/master/) nor
  confluent_kafka implemented the `DescribeLogDirsRequest`, available
  in java. A [recent
  PR](https://github.com/dpkp/kafka-python/pull/2278) for kafka-python
  supports it, so I used that experimental code. Not pretty, but it
  works.
  - As a result, there is a janky confluent_kafka to kafka-python
    AdminClient conf map in `logdirs.doit()`. It's only tested for
    sasl auth in MSK on AWS.

