Prometheus exporter in Python with background metrics gathering

prometheus_flask_exporter.py

"""Prometheus exporter with background metrics gathering."""
import signal
import threading
import time
from typing import TYPE_CHECKING

import schedule
from flask import Flask
from prometheus_client import Gauge, make_wsgi_app
from werkzeug.middleware.dispatcher import DispatcherMiddleware

if TYPE_CHECKING:
    from types import FrameType


class MetricsGatherer(threading.Thread):
    """Metrics gathering thread manager."""
    
    def __init__(self: MetricsGatherer) -> None:
        """Initialize object."""
        threading.Thread.__init__(self)

        self.shutdown_flag = threading.Event()
        schedule.every(10).seconds.do(update_metrics)
        schedule.run_all()
    
    def run(self: MetricsGatherer) -> None:
        """Run the thread."""
        print(f"Thread #{self.ident} started")

        while True:
            if self.shutdown_flag.is_set():
                schedule.clear()
                print(f"Thread #{self.ident} stopped")
                break
            schedule.run_pending()
            time.sleep(1)



class ServiceExit(Exception):
    """Custom exception which is used to trigger the clean exit of all running threads and the main program."""


def update_metrics() -> None:
    """Update your metrics."""


def service_shutdown(signum: int, _: FrameType) -> None:
    """Handle shutdown signals."""
    print("Caught signal %d" % signum)
    raise ServiceExit


signal.signal(signal.SIGTERM, service_shutdown)
signal.signal(signal.SIGINT, service_shutdown)

app = Flask(__name__)

@app.route("/")
def hello_world() -> str:
    """Serve the main app."""
    return "Hello world!"

app.wsgi_app = DispatcherMiddleware(app.wsgi_app, {
    "/metrics": make_wsgi_app(),
})

my_metric = Gauge("my_metric", "My beautiful metric")

metrics_gatherer = MetricsGatherer()
metrics_gatherer.start()


try:
    app.run()

except ServiceExit:
    metrics_gatherer.shutdown_flag.set()
    metrics_gatherer.join()