Add basic exporter

README.md

@@ -0,0 +1,44 @@
+# Fluent Bit Storage Metric Exporter
+
+Fluent Bit provides storage layer metrics under `api/v1/storage`, but unfortunately these are only available as JSON and therefore can't be scraped by Prometheus.
+
+This exporter runs as own web server, parses the storage layer metrics and provides them in a Prometheus compatible format.  
+
+## Setup
+
+This exporter ist written to run with Python 2.7.
+
+### Installation
+
+In order to install the exporter run `pip install .`
+
+### Start Exporter
+
+To start the exporter run `python -m fb_storage_metric_exporter  <EXPORTER-PORT> <FB-HOST> <FB-PORT>`
+
+`python -m fb_storage_metric_exporter 8080 127.0.0.1 2020` will start the exporter on port `8080` and the original storage metrics are requested from `127.0.0.1:2020/api/v1/storage`.
+
+## Provided Metrics
+
+All metrics are provided as gauge.
+
+### Storage Layer Metrics
+
+- fluentbit_storage_chunks
+- fluentbit_storage_chunks_mem
+- fluentbit_storage_chunks_fs
+- fluentbit_storage_chunks_fs_up
+- fluentbit_storage_chunks_fs_down
+
+### Input Metrics
+
+These metrics are provided for each running input plugin. You have to configure an alias in your Fluent Bit settings to set proper names for the plugins.
+
+- fluentbit_storage_input_overlimit
+- fluentbit_storage_input_mem_bytes
+- fluentbit_storage_input_limit_bytes
+- fluentbit_storage_input_chunks
+- fluentbit_storage_input_chunks_fs_up
+- fluentbit_storage_input_chunks_fs_down
+- fluentbit_storage_input_chunks_busy
+- fluentbit_storage_input_busy_bytes

fb_storage_metric_exporter/__main__.py

@@ -0,0 +1,33 @@
+import time
+import sys
+
+from prometheus_client import start_http_server, REGISTRY
+from .collector import StoreCollector
+
+
+def main():
+    if len(sys.argv) < 4:
+        print "Usage: <EXPORTER-PORT> <FB-HOST> <FB-PORT>"
+        sys.exit(1)
+
+    fb_host = sys.argv[2]
+    try:
+        fb_port = int(sys.argv[3])
+    except ValueError:
+        print "<FB-PORT> must be an integer"
+        sys.exit(1)
+    try:
+        server_port = int(sys.argv[1])
+    except ValueError:
+        print "<EXPORTER-PORT> must be an integer"
+        sys.exit(1)
+
+    collector = StoreCollector(fluent_bit_host=fb_host, fluent_bit_port=fb_port)
+    REGISTRY.register(collector)
+    start_http_server(server_port)
+    while True:
+        time.sleep(30)  # Server runs in daemon thread
+
+
+if __name__ == '__main__':
+    main()

fb_storage_metric_exporter/collector.py

@@ -0,0 +1,71 @@
+from prometheus_client.core import GaugeMetricFamily, CounterMetricFamily
+
+from .util import request_fb_storage_metrics
+from .parser import byte_str_to_int
+
+
+class StoreCollector(object):
+    def __init__(self, fluent_bit_host, fluent_bit_port):
+        self.fb_host = fluent_bit_host
+        self.fb_port = fluent_bit_port
+
+    def collect(self):
+        """
+        Collects Fluent Bit storage layer metrics available under /api/v1/storage
+        :return: Gauge metrics for storage layer
+        """
+
+        data = request_fb_storage_metrics(fb_host=self.fb_host, fb_port=self.fb_port)
+
+        # storage_layer
+        if "storage_layer" in data:
+            yield GaugeMetricFamily("fluentbit_storage_chunks", "Amount of currently used chunks",
+                                    value=data["storage_layer"]["chunks"]["total_chunks"])
+            yield GaugeMetricFamily("fluentbit_storage_chunks_mem", "Amount of chunks currently in memory",
+                                    value=data["storage_layer"]["chunks"]["mem_chunks"])
+            yield GaugeMetricFamily("fluentbit_storage_chunks_fs", "Amount of chunks currently in filesystem",
+                                    value=data["storage_layer"]["chunks"]["fs_chunks"])
+            yield GaugeMetricFamily("fluentbit_storage_chunks_fs_up", "Amount of chunks currently up",
+                                    value=data["storage_layer"]["chunks"]["fs_chunks_up"])
+            yield GaugeMetricFamily("fluentbit_storage_chunks_fs_down", "Amount of chunks currently down",
+                                    value=data["storage_layer"]["chunks"]["fs_chunks_down"])
+
+        # input_chunks
+        if "input_chunks" in data:
+            overlimit = GaugeMetricFamily("fluentbit_storage_input_overlimit",
+                                          "Memory buffer limit reached for input", labels=["name"])
+            mem_bytes = GaugeMetricFamily("fluentbit_storage_input_mem_bytes",
+                                          "Currently used memory buffer for input in bytes", labels=["name"])
+            limit_bytes = GaugeMetricFamily("fluentbit_storage_input_limit_bytes",
+                                            "Memory buffer limit for input in bytes", labels=["name"])
+
+            chunks = GaugeMetricFamily("fluentbit_storage_input_chunks",
+                                       "Amount of chunks currently used for input", labels=["name"])
+            chunks_fs_up = GaugeMetricFamily("fluentbit_storage_input_chunks_fs_up",
+                                             "Amount of chunks for input currently up", labels=["name"])
+            chunks_fs_down = GaugeMetricFamily("fluentbit_storage_input_chunks_fs_down",
+                                               "Amount of chunks for input currently down", labels=["name"])
+            chunks_busy = GaugeMetricFamily("fluentbit_storage_input_chunks_busy",
+                                            "Amount of chunks for input currently busy", labels=["name"])
+            busy_bytes = GaugeMetricFamily("fluentbit_storage_input_busy_bytes",
+                                           "Size of chunks for input currently busy in bytes", labels=["name"])
+
+            for input_name, stats in data["input_chunks"].iteritems():
+                overlimit.add_metric([input_name], int(stats["status"]["overlimit"]))
+                mem_bytes.add_metric([input_name], byte_str_to_int(stats["status"]["mem_size"]))
+                limit_bytes.add_metric([input_name], byte_str_to_int(stats["status"]["mem_limit"]))
+
+                chunks.add_metric([input_name], stats["chunks"]["total"])
+                chunks_fs_up.add_metric([input_name], stats["chunks"]["up"])
+                chunks_fs_down.add_metric([input_name], stats["chunks"]["down"])
+                chunks_busy.add_metric([input_name], stats["chunks"]["busy"])
+                busy_bytes.add_metric([input_name], byte_str_to_int(stats["chunks"]["busy_size"]))
+
+            yield overlimit
+            yield mem_bytes
+            yield limit_bytes
+            yield chunks
+            yield chunks_fs_up
+            yield chunks_fs_down
+            yield chunks_busy
+            yield busy_bytes

fb_storage_metric_exporter/parser.py

@@ -0,0 +1,17 @@
+from humanfriendly import parse_size, InvalidSize
+
+
+def byte_str_to_int(raw_str):
+    """
+    Converts data size strings like "4.8M" to an integer representation
+    :param raw_str: Human friendly string of data size
+    :return: Integer representation of data size
+    """
+
+    try:
+        res = parse_size(raw_str)
+    except InvalidSize:
+        print "Could not parse {}".format(raw_str)
+        res = 0
+    return res
+

fb_storage_metric_exporter/util.py

@@ -0,0 +1,22 @@
+import requests
+
+FB_ENDPOINT = "api/v1/storage"
+
+
+def request_fb_storage_metrics(fb_host, fb_port):
+    """
+    Requests Fluent Bit /api/v1/storage API endpoint and parses JSON
+    :param fb_host: Host of Fluent Bit
+    :param fb_port: Port of Fluent Bit
+    :return: Parsed JSON of /api/v1/storage
+    """
+    conn_url = "http://{}:{}/{}".format(fb_host, fb_port, FB_ENDPOINT)
+    try:
+        resp = requests.get(url=conn_url)
+    except requests.exceptions.ConnectionError as e:
+        print "Could not establish connection to FB: {}".format(e.message)
+        return {}
+    if resp.status_code != 200:
+        print "Received unexpected status code {} requesting /api/v1/storage".format(resp.status_code)
+        return {}
+    return resp.json()

setup.py

@@ -0,0 +1,16 @@
+from setuptools import setup
+
+
+setup(
+    name="fb_storage_metric_exporter",
+    version="0.1.0",
+    packages=[
+        "fb_storage_metric_exporter",
+    ],
+    python_requires=">=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*",
+    install_requires=[
+        "prometheus-client",
+        "humanfriendly",
+        "requests"
+    ],
+)