Added health checks with flask-healthz

mirrormanager2/app.py

@@ -29,6 +29,7 @@
 
 import flask
 from flask_admin import Admin
+from flask_healthz import healthz
 from flask_oidc import OpenIDConnect
 from sqlalchemy.orm import configure_mappers
 
@@ -130,6 +131,8 @@ def create_app(config=None):
     app.register_blueprint(api_views, url_prefix="/api")
     from mirrormanager2.xml_rpc import XMLRPC
 
+    app.register_blueprint(healthz, url_prefix="/healthz")
+
     XMLRPC.connect(app, "/xmlrpc")
 
     return app

mirrormanager2/default_config.py

@@ -195,3 +195,8 @@
 UMDL_PREFIX = ""
 
 UMDL_MASTER_DIRECTORIES = []
+
+HEALTHZ = {
+    "live": "mirrormanager2.health_checks.liveness",
+    "ready": "mirrormanager2.health_checks.readiness",
+}

mirrormanager2/health_checks.py

@@ -0,0 +1,19 @@
+from flask_healthz import HealthError
+from sqlalchemy_helpers import DatabaseStatus
+
+from mirrormanager2.database import DB
+
+
+def liveness():
+    pass
+
+
+def readiness():
+    try:
+        status = DB.manager.get_status()
+    except Exception as e:
+        raise HealthError(f"Can't get the database status: {e}") from e
+    if status is DatabaseStatus.NO_INFO:
+        raise HealthError("Can't connect to the database")
+    if status is DatabaseStatus.UPGRADE_AVAILABLE:
+        raise HealthError("The database schema needs to be updated")

pyproject.toml

@@ -47,6 +47,7 @@ python = "^3.9"
 email-validator = ">=1.1.3"
 flask = "^2.2.3 || ^3.0.0"
 flask-admin = "^1.6.0"
+flask-healthz = "^1.0.0"
 flask-oidc = "^2.0.0"
 flask-xml-rpc-re = "^0.1.4"
 flask-wtf = "^1.1.1"

tests/test_health_checks.py

@@ -0,0 +1,21 @@
+from alembic import command
+
+from mirrormanager2.database import DB
+
+
+def test_healthz_readiness_ok(client, db):
+    result = client.get("/healthz/ready")
+    assert result.status_code == 200
+
+
+def test_healthz_readiness_not_ok_when_old_schema(client, db):
+    command.downgrade(DB.manager.alembic_cfg, "-1")
+
+    result = client.get("/healthz/ready")
+    assert result.status_code != 200
+    assert "The database schema needs to be updated" in result.get_data(as_text=True)
+
+
+def test_healthz_readiness_not_ok(client):
+    result = client.get("/healthz/ready")
+    assert result.status_code != 200