Ensure Flask-Monitor is singleton (#4)

* Teste

Signed-off-by: Gustavo Coelho <[email protected]>

* make metrics registry singleton and stops Timer thread when close

Signed-off-by: Luiz Oliveira <[email protected]>

* update docs to watch_dependencies

Signed-off-by: Luiz Oliveira <[email protected]>

* improve tests

Signed-off-by: Luiz Oliveira <[email protected]>

* install nose to run testes

Signed-off-by: Luiz Oliveira <[email protected]>

* use default prom registry if none passed

Signed-off-by: Luiz Oliveira <[email protected]>

* include tests for watch_dependencies

Signed-off-by: Luiz Oliveira <[email protected]>

* isError receives conditional result

Signed-off-by: Luiz Oliveira <[email protected]>

* using extensions to hold prometheus registry

Signed-off-by: Luiz Oliveira <[email protected]>

* default status code value to exception

Signed-off-by: Luiz Oliveira <[email protected]>

* splited watch_dependencies in watch_dependencies and collect_dependency_time

Signed-off-by: Luiz Oliveira <[email protected]>

* improved simple_example

Signed-off-by: Luiz Oliveira <[email protected]>

* using status code as result in example

Signed-off-by: Luiz Oliveira <[email protected]>

* removed global statement

Signed-off-by: Luiz Oliveira <[email protected]>

* removed use of kwargs

Signed-off-by: Luiz Oliveira <[email protected]>

* removed start time from cdt

Signed-off-by: Luiz Oliveira <[email protected]>

Co-authored-by: Gustavo Coelho <[email protected]>

Author: Ziul

.github/workflows/continuous-integration-workflow.yml

@@ -21,5 +21,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
       - name: Install dependencies
         run: pip3 install -r requirements.txt
+      - name: Install tests dependencies
+        run: pip3 install nose coverage
       - name: Run tests
-        run: pytest
+        run: nosetests --with-coverage --cover-package=flask_monitor -v tests

README.md

@@ -106,7 +106,7 @@ def check_db():
         traceback.print_stack()
     return 0
 
-watch_dependencies("Bd", check_db)
+watch_dependencies("Bd", check_db, app=app)
 ```
 
 Other optional parameters are also:

example/simple_example.py

@@ -3,9 +3,14 @@
 from werkzeug.middleware.dispatcher import DispatcherMiddleware
 from werkzeug.serving import run_simple
 import traceback
-from flask_monitor import register_metrics, watch_dependencies
+from flask_monitor import register_metrics, watch_dependencies, collect_dependency_time
 from flask import Flask
 import requests as req
+from prometheus_client import CollectorRegistry
+from time import time, sleep
+from random import random
+
+registry = CollectorRegistry()
 
 ## create a flask app
 app = Flask(__name__)
@@ -23,27 +28,27 @@ def is_error200(code):
 # buckets is the internavals for histogram parameter. buckets is a optional parameter
 # error_fn is a function to define what http status code is a error. By default errors are
 # 400 and 500 status code. error_fn is a option parameter
-register_metrics(app, buckets=[0.3, 0.6], error_fn=is_error200)
+register_metrics(app, buckets=[0.3, 0.6], error_fn=is_error200, registry=registry)
 
 # Plug metrics WSGI app to your main app with dispatcher
-dispatcher = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app()})
+dispatcher = DispatcherMiddleware(app.wsgi_app, {"/metrics": make_wsgi_app(registry=registry)})
 
 # a dependency healthcheck
 def check_db():
     try:
         response = req.get("http://localhost:5000/database")
-        if response.status_code == 200:
-            return 1
+        app.logger.info(response)
+        return response.status_code < 400
     except:
-        traceback.print_stack()
-    return 0
+        return 0
+    
 
 # watch dependency
 # first parameter is the dependency's name. It's a mandatory parameter.
 # second parameter is the health check function. It's a mandatory parameter.
 # time_execution is used to set the interval of running the healthchec function.
 # time_execution is a optional parameter
-watch_dependencies("database", check_db, time_execution=1)
+scheduler = watch_dependencies('database', check_db, app=app, time_execution=500)
 
 # endpoint
 @app.route('/teste')
@@ -58,6 +63,22 @@ def hello_world():
 # endpoint
 @app.route('/database')
 def bd_running():
+    start = time()
+    # checks the database
+    sleep(random()/10)
+    # compute the elapsed time
+    elapsed = time() - start
+    # register the dependency time
+    collect_dependency_time(
+        app=app,
+        name='database',
+        rtype='http',
+        status=200,
+        is_error= 'False',
+        method='GET',
+        addr='external/database',
+        elapsed=elapsed
+    )
     return 'I am a database working.'
 
 if __name__ == "__main__":

flask_monitor/__init__.py

@@ -1 +1,2 @@
-from .metrics import register_metrics, watch_dependencies
+""" Functions for define and register metrics """
+from .metrics import register_metrics, watch_dependencies, collect_dependency_time

flask_monitor/metrics.py

@@ -1,26 +1,14 @@
 """ Functions for define and register metrics """
 import time
-import threading
-from flask import request
-from prometheus_client import Counter, Histogram, Gauge
+import atexit
+from flask import request, current_app
+from prometheus_client import Counter, Histogram, Gauge, CollectorRegistry
+from apscheduler.schedulers.background import BackgroundScheduler
 
 #
 # Request callbacks
 #
-
-METRICS_INFO = Gauge(
-    "application_info",
-    "records static application info such as it's semantic version number",
-    ["version"]
-)
-
-DEPENDENCY_UP = Gauge(
-    'dependency_up',
-    'records if a dependency is up or down. 1 for up, 0 for down',
-    ["name"]
-)
-
-def is_error(code):
+def _is_error_(code):
     """
     Default status error checking
     """
@@ -30,7 +18,7 @@ def is_error(code):
 #
 # Metrics registration
 #
-def register_metrics(app, buckets=None, error_fn=None):
+def register_metrics(app=current_app, buckets=None, error_fn=None, registry=None):
     """
     Register metrics middlewares
 
@@ -42,24 +30,46 @@ def register_metrics(app, buckets=None, error_fn=None):
     Before CPython 3.6 dictionaries didn't guarantee keys order, so callbacks
     could be executed in arbitrary order.
     """
+
+    if app.config.get("METRICS_ENABLED", False):
+        return app, app.extensions.get("registry", registry)
+    app.config["METRICS_ENABLED"] = True
+    if not registry:
+        registry = app.extensions.get("registry", CollectorRegistry())
+    app.extensions["registry"] = registry
+    app.logger.info('Metrics enabled')
+
     buckets = [0.1, 0.3, 1.5, 10.5] if buckets is None else buckets
+
+
     # pylint: disable=invalid-name
-    METRICS_REQUEST_LATENCY = Histogram(
+    metrics_info = Gauge(
+        "application_info",
+        "records static application info such as it's semantic version number",
+        ["version", "name"],
+        registry=registry
+    )
+
+    # pylint: disable=invalid-name
+    metrics_request_latency = Histogram(
         "request_seconds",
         "records in a histogram the number of http requests and their duration in seconds",
-        ["type", "status", "isError", "method", "addr"],
-        buckets=buckets
+        ["type", "status", "isError", "errorMessage", "method", "addr"],
+        buckets=buckets,
+        registry=registry
     )
 
-    METRICS_REQUEST_SIZE = Counter(
+    # pylint: disable=invalid-name
+    metrics_request_size = Counter(
         "response_size_bytes",
         "counts the size of each http response",
-        ["type", "status", "isError", "method", "addr"],
+        ["type", "status", "isError", "errorMessage", "method", "addr"],
+        registry=registry
     )
     # pylint: enable=invalid-name
 
     app_version = app.config.get("APP_VERSION", "0.0.0")
-    METRICS_INFO.labels(app_version).set(1)
+    metrics_info.labels(app_version, app.name).set(1)
 
     def before_request():
         """
@@ -77,29 +87,91 @@ def after_request(response):
         # pylint: disable=protected-access
         request_latency = time.time() - request._prometheus_metrics_request_start_time
         # pylint: enable=protected-access
-        error_status = is_error(response.status_code)
-        METRICS_REQUEST_LATENCY \
-            .labels("http", response.status_code, error_status, request.method, request.path) \
+        error_status = _is_error_(response.status_code)
+        metrics_request_latency \
+            .labels("http", response.status_code, error_status, "", request.method, request.path) \
             .observe(request_latency)
-        METRICS_REQUEST_SIZE.labels(
-            "http", response.status_code, error_status, request.method, request.path
+        metrics_request_size.labels(
+            "http", response.status_code, error_status, "", request.method, request.path
         ).inc(size_request)
         return response
+
     if error_fn is not None:
-        is_error.__code__ = error_fn.__code__
+        _is_error_.__code__ = error_fn.__code__
     app.before_request(before_request)
     app.after_request(after_request)
+    return app, registry
+
 
-def watch_dependencies(dependency, func, time_execution=1500):
+def watch_dependencies(dependency, func, time_execution=15000, registry=None, app=current_app):
+    """
+    Register dependencies metrics up
+    """
+
+    if not registry:
+        registry = app.extensions.get("registry", CollectorRegistry())
+        app.extensions["registry"] = registry
+
+    # pylint: disable=invalid-name
+    DEPENDENCY_UP = Gauge(
+        'dependency_up',
+        'records if a dependency is up or down. 1 for up, 0 for down',
+        ["name"],
+        registry=registry
+    )
+    def register_dependecy():
+        DEPENDENCY_UP.labels(dependency).set(func())
+
+    scheduler = BackgroundScheduler()
+    scheduler.add_job(
+        func=register_dependecy,
+        trigger="interval",
+        seconds=time_execution/1000,
+        max_instances=1,
+        name='dependency',
+        misfire_grace_time=2,
+        replace_existing=True
+        )
+    scheduler.start()
+
+    # Shut down the scheduler when exiting the app
+    atexit.register(scheduler.shutdown)
+    return scheduler
+
+# pylint: disable=too-many-arguments
+def collect_dependency_time(
+    app, name, rtype='http', status=200,
+    is_error=False, error_message='',
+    method='GET', addr='/',
+    elapsed=0,
+    registry=None
+):
     """
     Register dependencies metrics
     """
-    def thread_function():
-        thread = threading.Timer(time_execution, lambda x: x + 1, args=(1,))
-        thread.start()
-        thread.join()
-        response = func()
-        DEPENDENCY_UP.labels(dependency).set(response)
-        thread_function()
-    thread = threading.Timer(time_execution, thread_function)
-    thread.start()
+
+    if not registry:
+        registry = app.extensions.get("registry", CollectorRegistry())
+        app.extensions["registry"] = registry
+
+    dependency_up_latency = app.extensions.get(
+        "dependency_latency"
+    )
+    if not dependency_up_latency:
+        app.extensions['dependency_latency'] = dependency_up_latency = Histogram(
+                "dependency_request_seconds",
+                "records in a histogram the number of requests to dependency",
+                ["name", "type", "status", "isError", "errorMessage", "method", "addr"],
+                registry=registry
+            )
+
+    dependency_up_latency \
+        .labels(
+            name,
+            rtype.lower(),
+            status,
+            "False" if is_error else "True",
+            error_message,
+            method.upper(),
+            addr) \
+        .observe(elapsed)