Initial commit

.gitignore

@@ -0,0 +1,62 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*,cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Gedit
+*~

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3-slim
+
+WORKDIR /usr/src/app
+
+COPY prometheus_kafka_consumer_group_exporter/*.py /usr/src/app/prometheus_kafka_consumer_group_exporter/
+COPY setup.py /usr/src/app/
+COPY LICENSE /usr/src/app/
+COPY README.md /usr/src/app/
+COPY MANIFEST.in /usr/src/app/
+
+RUN pip install -e .
+
+EXPOSE 8080
+
+ENTRYPOINT ["python", "-u", "/usr/local/bin/prometheus-kafka-consumer-group-exporter"]

LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2016 Braedon Vickers
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

MANIFEST.in

@@ -0,0 +1,2 @@
+include LICENSE
+include README.md

prometheus_kafka_consumer_group_exporter/__init__.py

@@ -0,0 +1,125 @@
+import argparse
+import logging
+
+from kafka import KafkaConsumer, KafkaProducer
+from prometheus_client import start_http_server, Gauge
+from struct import unpack_from, unpack
+
+gauges = {}
+
+def update_gauge(metric_name, label_dict, value):
+    label_keys = tuple(label_dict.keys())
+    label_values = tuple(label_dict.values())
+
+    if metric_name not in gauges:
+        gauges[metric_name] = Gauge(metric_name, '', label_keys)
+
+    gauge = gauges[metric_name]
+
+    if label_values:
+        gauge.labels(*label_values).set(value)
+    else:
+        gauge.set(value)
+
+def main():
+    parser = argparse.ArgumentParser(description='Export Kafka consumer offsets to Prometheus.')
+    parser.add_argument('-b', '--bootstrap-brokers', default='localhost',
+        help='addresses of brokers in a Kafka cluster to read the offsets topic of. Brokers should be separated by commas e.g. broker1,broker2. Ports can be provided if non-standard (9092) e.g. brokers1:9999 (default: localhost)')
+    parser.add_argument('-p', '--port', type=int, default=8080,
+        help='port to serve the metrics endpoint on. (default: 8080)')
+    parser.add_argument('-v', '--verbose', action='store_true',
+        help='turn on verbose logging.')
+    args = parser.parse_args()
+
+    logging.basicConfig(
+        format='[%(asctime)s] %(name)s.%(levelname)s %(threadName)s %(message)s',
+        level=logging.DEBUG if args.verbose else logging.INFO
+    )
+    logging.captureWarnings(True)
+
+    port = args.port
+    bootstrap_brokers = args.bootstrap_brokers.split(',')
+
+    consumer = KafkaConsumer(
+        '__consumer_offsets',
+         bootstrap_servers=bootstrap_brokers,
+         auto_offset_reset='earliest',
+         group_id=None
+    )
+
+    logging.info('Starting server...')
+    start_http_server(port)
+    logging.info('Server started on port %s', port)
+
+    def read_short(bytes):
+        num = unpack_from('>h', bytes)[0]
+        remaining = bytes[2:]
+        return (num, remaining)
+
+    def read_int(bytes):
+        num = unpack_from('>i', bytes)[0]
+        remaining = bytes[4:]
+        return (num, remaining)
+
+    def read_long_long(bytes):
+        num = unpack_from('>q', bytes)[0]
+        remaining = bytes[8:]
+        return (num, remaining)
+
+    def read_string(bytes):
+        length, remaining = read_short(bytes)
+        string = remaining[:length].decode('utf-8')
+        remaining = remaining[length:]
+        return (string, remaining)
+
+    def parse_key(bytes):
+        (version, remaining_key) = read_short(bytes)
+        if version == 1 or version == 0:
+            (group, remaining_key) = read_string(remaining_key)
+            (topic, remaining_key) = read_string(remaining_key)
+            (partition, remaining_key) = read_int(remaining_key)
+            return (version, group, topic, partition)
+
+    def parse_value(bytes):
+        (version, remaining_key) = read_short(bytes)
+        if version == 0:
+            (offset, remaining_key) = read_long_long(remaining_key)
+            (metadata, remaining_key) = read_string(remaining_key)
+            (timestamp, remaining_key) = read_long_long(remaining_key)
+            return (version, offset, metadata, timestamp)
+        elif version == 1:
+            (offset, remaining_key) = read_long_long(remaining_key)
+            (metadata, remaining_key) = read_string(remaining_key)
+            (commit_timestamp, remaining_key) = read_long_long(remaining_key)
+            (expire_timestamp, remaining_key) = read_long_long(remaining_key)
+            return (version, offset, metadata, commit_timestamp, expire_timestamp)
+
+    try:
+        for message in consumer:
+
+            update_gauge(
+                metric_name='exporter_offset',
+                label_dict={
+                    'partition': message.partition
+                },
+                value=message.offset
+            )
+
+            key = parse_key(message.key)
+            if key:
+                value = parse_value(message.value)
+
+                update_gauge(
+                    metric_name='consumer_group_offset',
+                    label_dict={
+                        'group': key[1],
+                        'topic': key[2],
+                        'partition': key[3]
+                    },
+                    value=value[1]
+                )
+
+    except KeyboardInterrupt:
+        pass
+
+    logging.info('Shutting down')

prometheus_kafka_consumer_group_exporter/__main__.py

@@ -0,0 +1,4 @@
+from prometheus_kafka_consumer_group_exporter import main
+
+if __name__ == '__main__':
+    main()

setup.py

@@ -0,0 +1,31 @@
+from setuptools import setup, find_packages
+
+setup(
+    name='prometheus-kafka-consumer-group-exporter',
+    version='0.1.0.dev1',
+    description='Kafka consumer group Prometheus exporter',
+    url='https://github.com/Braedon/prometheus-kafka-consumer-group-exporter',
+    author='Braedon Vickers',
+    author_email='braedon.vickers@gmal.com',
+    license='MIT',
+    classifiers=[
+        'Development Status :: 4 - Beta',
+        'Intended Audience :: Developers',
+        'Intended Audience :: System Administrators',
+        'Topic :: System :: Monitoring',
+        'License :: OSI Approved :: MIT License',
+        'Programming Language :: Python :: 3',
+        'Programming Language :: Python :: 3.4',
+    ],
+    keywords='monitoring prometheus exporter kafka consumer group',
+    packages=find_packages(),
+    install_requires=[
+        'kafka-python',
+        'prometheus-client'
+    ],
+    entry_points={
+        'console_scripts': [
+            'prometheus-kafka-consumer-group-exporter=prometheus_kafka_consumer_group_exporter:main',
+        ],
+    },
+)