birb_latency_collector/birb_latency_collector/utils/healthchecks.py

20 lines
527 B
Python

#!/usr/bin/env python3
from .influx import influxdb_instance
from healthcheck import HealthCheck
from flask import Flask
def health_database_status():
try:
version = influxdb_instance.connection.ping()
except Exception as e:
return False, str(e)
else:
return True, f'influxdb ({version}) is ok'
def register_health_checks(app: Flask):
health = HealthCheck()
health.add_check(health_database_status)
app.add_url_rule("/healthz", "healthcheck", view_func=lambda: health.run())