outposts: implement .run_wuth_logs() which returns logs, add task monitoring

passbook/outposts/controllers/base.py

@@ -1,7 +1,8 @@
 """Base Controller"""
-from typing import Dict
+from typing import Dict, List
 
 from structlog import get_logger
+from structlog.testing import capture_logs
 
 from passbook.lib.sentry import SentryIgnoredException
 from passbook.outposts.models import Outpost
@@ -29,6 +30,12 @@ class BaseController:
         """Called by scheduled task to reconcile deployment/service/etc"""
         raise NotImplementedError
 
+    def run_with_logs(self) -> List[str]:
+        """Call .run() but capture all log output and return it."""
+        with capture_logs() as logs:
+            self.run()
+        return logs
+
     def get_static_deployment(self) -> str:
         """Return a static deployment configuration"""
         raise NotImplementedError

passbook/outposts/settings.py

@@ -3,7 +3,7 @@ from celery.schedules import crontab
 
 CELERY_BEAT_SCHEDULE = {
     "outposts_controller": {
-        "task": "passbook.outposts.tasks.outpost_controller",
+        "task": "passbook.outposts.tasks.outpost_controller_all",
         "schedule": crontab(minute="*/5"),
         "options": {"queue": "passbook_scheduled"},
     },

passbook/outposts/tasks.py

@@ -6,7 +6,9 @@ from channels.layers import get_channel_layer
 from django.db.models.base import Model
 from structlog import get_logger
 
+from passbook.lib.tasks import MonitoredTask, TaskResult, TaskResultStatus
 from passbook.lib.utils.reflection import path_to_class
+from passbook.outposts.controllers.base import ControllerException
 from passbook.outposts.models import (
     Outpost,
     OutpostDeploymentType,
@@ -22,24 +24,30 @@ LOGGER = get_logger()
 
 
 @CELERY_APP.task()
-def outpost_controller():
+def outpost_controller_all():
     """Launch Controller for all Outposts which support it"""
     for outpost in Outpost.objects.exclude(
         deployment_type=OutpostDeploymentType.CUSTOM
     ):
-        outpost_controller_single.delay(
-            outpost.pk.hex, outpost.deployment_type, outpost.type
-        )
+        outpost_controller.delay(outpost.pk.hex, outpost.deployment_type, outpost.type)
 
 
-@CELERY_APP.task()
-def outpost_controller_single(outpost_pk: str, deployment_type: str, outpost_type: str):
+@CELERY_APP.task(bind=True, base=MonitoredTask)
+def outpost_controller(
+    self: MonitoredTask, outpost_pk: str, deployment_type: str, outpost_type: str
+):
     """Launch controller and reconcile deployment/service/etc"""
-    if outpost_type == OutpostType.PROXY:
-        if deployment_type == OutpostDeploymentType.KUBERNETES:
-            ProxyKubernetesController(outpost_pk).run()
-        if deployment_type == OutpostDeploymentType.DOCKER:
-            ProxyDockerController(outpost_pk).run()
+    logs = []
+    try:
+        if outpost_type == OutpostType.PROXY:
+            if deployment_type == OutpostDeploymentType.KUBERNETES:
+                logs = ProxyKubernetesController(outpost_pk).run_with_logs()
+            if deployment_type == OutpostDeploymentType.DOCKER:
+                logs = ProxyDockerController(outpost_pk).run_with_logs()
+    except ControllerException as exc:
+        self.set_status(TaskResult(TaskResultStatus.ERROR, [str(exc)], exc))
+    else:
+        self.set_status(TaskResult(TaskResultStatus.SUCCESSFUL, logs))
 
 
 @CELERY_APP.task()