import logging
from data.model import health
from app import build_logs
logger = logging.getLogger(__name__)
def _check_registry_gunicorn(app):
""" Returns the status of the registry gunicorn workers. """
# Compute the URL for checking the registry endpoint. We append a port if and only if the
# hostname contains one.
client = app.config['HTTPCLIENT']
hostname_parts = app.config['SERVER_HOSTNAME'].split(':')
port = ''
if len(hostname_parts) == 2:
port = ':' + hostname_parts[1]
registry_url = '%s://localhost%s/v1/_internal_ping' % (app.config['PREFERRED_URL_SCHEME'], port)
try:
return client.get(registry_url, verify=False, timeout=2).status_code == 200
except Exception:
logger.exception('Exception when checking registry health: %s', registry_url)
return False
def _check_database(app):
""" Returns the status of the database, as accessed from this instance. """
return health.check_health(app.config)
def _check_redis(app):
""" Returns the status of Redis, as accessed from this instance. """
return build_logs.check_health()
_SERVICES = {
'registry_gunicorn': _check_registry_gunicorn,
'database': _check_database,
'redis': _check_redis
}
def check_all_services(app, skip):
""" Returns a dictionary containing the status of all the services defined. """
status = {}
for name in _SERVICES:
if name in skip:
continue
status[name] = _SERVICES[name](app)
return status
