CommonHealth Metric Impl. / CLI command json output

health_check/backends.py

@@ -53,9 +53,9 @@ def get_prometheus_metric(self, metric_type: type, name, description: str = None
     def check_status(self):
         raise NotImplementedError
 
-    def run_check(self):
+    def run_check(self, external_errors=None):
         start = timer()
-        self.errors = []
+        self.errors = external_errors or []
         try:
             self.check_status()
         except HealthCheckException as e:
@@ -77,15 +77,18 @@ def add_error(self, error, cause=None):
         else:
             msg = _("unknown error")
             error = HealthCheckException(msg)
-        if isinstance(cause, BaseException):
-            logger.exception(str(error))
-        else:
-            logger.error(str(error))
+
+        if HEALTH_CHECK['VERBOSE']:
+            if isinstance(cause, BaseException):
+                logger.exception(str(error))
+            else:
+                logger.error(str(error))
+
         self.errors.append(error)
 
     def pretty_status(self):
         if self.errors:
-            return "\n".join(str(e) for e in self.errors)
+            return "; ".join(str(e) for e in self.errors)
         return _('working')
 
     @property
@@ -94,3 +97,9 @@ def status(self):
 
     def identifier(self):
         return self.__class__.__name__
+
+
+class CommonHealth(BaseHealthCheckBackend):
+
+    def check_status(self):
+        pass

health_check/conf.py

@@ -6,3 +6,4 @@
 HEALTH_CHECK.setdefault('WARNINGS_AS_ERRORS', True)
 HEALTH_CHECK.setdefault('USE_PROMETHEUS', False)
 HEALTH_CHECK.setdefault('PROMETHEUS_METRIC_NAMESPACE', 'app')
+HEALTH_CHECK.setdefault('VERBOSE', True)

health_check/management/commands/health_check.py

@@ -1,17 +1,34 @@
 import sys
+import json
 
 from django.core.management.base import BaseCommand
 
+from health_check.conf import HEALTH_CHECK
 from health_check.mixins import CheckMixin
 
 
 class Command(CheckMixin, BaseCommand):
     help = "Run health checks and exit 0 if everything went well."
 
+    def add_arguments(self, parser):
+        parser.add_argument("--json-output", action="store_true", required=False)
+        parser.add_argument("--verbose", action="store_true", required=False)
+
     def handle(self, *args, **options):
+        if not options["verbose"]:
+            HEALTH_CHECK["VERBOSE"] = False
+
         # perform all checks
         errors = self.errors
+        if options["json_output"]:
+
+            self.json_output()
+        else:
+            self.plain_output()
+            if errors:
+                sys.exit(1)
 
+    def plain_output(self):
         for plugin in self.plugins:
             style_func = self.style.SUCCESS if not plugin.errors else self.style.ERROR
             self.stdout.write(
@@ -21,5 +38,9 @@ def handle(self, *args, **options):
                 )
             )
 
-        if errors:
-            sys.exit(1)
+    def json_output(self):
+        metrics = {
+            p.identifier(): p.status
+            for p in self.plugins
+        }
+        self.stdout.write(json.dumps(metrics))

health_check/mixins.py

@@ -1,6 +1,7 @@
 import copy
 from concurrent.futures import ThreadPoolExecutor
 
+from health_check.backends import CommonHealth
 from health_check.conf import HEALTH_CHECK
 from health_check.exceptions import ServiceWarning
 from health_check.plugins import plugin_dir
@@ -9,6 +10,7 @@
 class CheckMixin:
     _errors = None
     _plugins = None
+    _common_health = None
 
     @property
     def errors(self):
@@ -25,6 +27,13 @@ def plugins(self):
             ), key=lambda plugin: plugin.identifier())
         return self._plugins
 
+    @property
+    def common_health(self):
+        if not self._common_health:
+            self._common_health = CommonHealth()
+            self._plugins.append(self._common_health)
+        return self._common_health
+
     def run_check(self):
         errors = []
 
@@ -36,8 +45,12 @@ def _run(plugin):
                 from django.db import connections
                 connections.close_all()
 
+        error_plugins = []
         with ThreadPoolExecutor(max_workers=len(self.plugins) or 1) as executor:
             for plugin in executor.map(_run, self.plugins):
+                if plugin.errors:
+                    error_plugins.append(plugin.identifier())
+
                 if plugin.critical_service:
                     if not HEALTH_CHECK['WARNINGS_AS_ERRORS']:
                         errors.extend(
@@ -47,4 +60,6 @@ def _run(plugin):
                     else:
                         errors.extend(plugin.errors)
 
+        self.common_health.run_check(external_errors=error_plugins)
+
         return errors