quay/health/services.py

import logging
from data.model import health
from app import build_logs, storage


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()

def _check_storage(app):
  """ Returns the status of storage, as accessed from this instance. """
  try:
    storage.validate(storage.preferred_locations, app.config['HTTPCLIENT'])
    return True
  except Exception as ex:
    logger.exception('Storage check failed with exception %s', ex)
    return False


_SERVICES = {
  'registry_gunicorn': _check_registry_gunicorn,
  'database': _check_database,
  'redis': _check_redis,
  'storage': _check_storage,
}

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