Flask and Prometheus custom metrics examples

prometheus_flask.py

import random

import argparse
import time
import traceback
from concurrent.futures.thread import ThreadPoolExecutor
from flask import Flask
from prometheus_flask_exporter import PrometheusMetrics
from prometheus_client import Gauge, Counter, Histogram, Info, Enum

app = Flask(__name__)


class WebServer:
    def __init__(self, pool):
        self.pool = pool
        self._app = Flask(__name__, static_folder='static')
        self.metrics = PrometheusMetrics(self._app)
        self.metrics.info('app_info', 'Application info', version='1.0.3')
        self.my_gauge = Gauge('my_gauge', 'Description of gauge')
        self.my_counter = Counter('my_counter', 'Description of counter')
        self.my_hist = Histogram('my_histogram', 'Description of counter')
        self.my_info = Info('my_info', 'Description of info')
        self.my_info.info({"my_key": "my val"})
        self.my_enum = Enum('my_task_state', 'Description of enum',
                            states=['starting', 'running', 'stopped'])
        self.my_enum.state('starting')

    def _setup_routes(self):
        @self._app.route("/")
        def index():
            return "hi"

    def update_metrics(self):
        while True:
            try:
                self.my_gauge.set(random.random())
                self.my_counter.inc()
                self.my_hist.observe(random.random())
                time.sleep(0.5)
            except Exception:
                traceback.print_exc()

    def start(self, ip, port):
        self._setup_routes()
        self.pool.submit(self._app.run,
                         host=ip,
                         port=port,
                         debug=False,
                         threaded=True,
                         use_reloader=False)
        self.my_enum.state('running')


if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    ap.add_argument("-i", "--ip", type=str, required=True,
                    help="ip address of the device")
    ap.add_argument("-o", "--port", type=int, required=True,
                    help="ephemeral port number of the server (1024 to 65535)")
    args = vars(ap.parse_args())
    pool = ThreadPoolExecutor(10)
    web_server = WebServer(pool)
    pool.submit(web_server.update_metrics)
    web_server.start(args["ip"], args["port"])