feat(gitops_server): Update github deployment status

william chu · william chu · commit fbe88119814f · 2021-07-30T15:51:49.000+10:00
diff --git a/Dockerfile b/Dockerfile
@@ -35,7 +35,7 @@ RUN pip3 install poetry
 COPY pyproject.toml poetry.lock /app/
 
 # Install dependencies
-RUN poetry install -E server --no-dev
+RUN poetry install --extras server --no-dev
 
 COPY cluster.key /app/
 COPY gitops /app/gitops/
diff --git a/charts/gitops/Chart.yaml b/charts/gitops/Chart.yaml
@@ -2,4 +2,4 @@ apiVersion: v1
 appVersion: "1.0"
 description: GitOps Server Helm chart.
 name: gitops
-version: 0.3.0
+version: 0.4.0
diff --git a/gitops/common/app.py b/gitops/common/app.py
@@ -46,6 +46,17 @@ def __eq__(self, other):
     def is_inactive(self):
         return "inactive" in self.values.get("tags", [])
 
+    def set_value(self, path: str, value: any):
+        """Sets the value on the path of self.values
+        Usage:
+            app.set_value("deployments.label.GITHUB_DEPLOYMENT_KEY", "1")
+        """
+        keys = path.split(".")
+        current_dict = self.values
+        for key in keys[:-1]:
+            current_dict = current_dict.setdefault(key, {})
+        current_dict[keys[-1]] = value
+
     def _make_values(self, deployments: Dict, secrets: Dict) -> Dict:
         values = {
             **deployments,
diff --git a/gitops_server/deploy.py b/gitops_server/deploy.py
@@ -3,14 +3,14 @@
 import logging
 import os
 import tempfile
+import uuid
 from typing import List, Optional
 
 from gitops.common.app import App
 
-from . import settings
+from . import github, settings, slack
 from .app_definitions import AppDefinitions
 from .git import temp_repo
-from .slack import post
 from .types import UpdateAppResult
 from .utils import get_repo_name_from_url, run
 
@@ -27,25 +27,37 @@ async def post_init_summary(
     for typ, d in [("Adding", added_apps), ("Updating", updated_apps), ("Removing", removed_apps)]:
         if d:
             deltas += f"\n\t• {typ}: {', '.join(f'`{app}`' for app in sorted(d))}"
-    await post(
+    await slack.post(
         f"A deployment from `{source}` has been initiated by *{username}* for cluster"
         f" `{settings.CLUSTER_NAME}`, the following apps will be updated:{deltas}\nCommit Message:"
         f" {commit_message}"
     )
 
 
-async def post_result(source: str, result: UpdateAppResult):
+async def post_result(app: App, source: str, result: UpdateAppResult):
+    github_deployment_url = str(app.values.get("github/deployment_url", ""))
     if result["exit_code"] != 0:
-        await post(
+        await github.update_deployment(
+            github_deployment_url,
+            status=github.STATUSES.failure,
+            description=f"Failed to deploy app. {result['output']}",
+        )
+        await slack.post(
             f"Failed to deploy app `{result['app_name']}` from `{source}` for cluster"
             f" `{settings.CLUSTER_NAME}`:\n>>>{result['output']}"
         )
+    else:
+        await github.update_deployment(
+            github_deployment_url,
+            status=github.STATUSES.in_progress,
+            description="Helm installed app into cluster. Waiting for pods to deploy.",
+        )
 
 
 async def post_result_summary(source: str, results: List[UpdateAppResult]):
     n_success = sum([r["exit_code"] == 0 for r in results])
     n_failed = sum([r["exit_code"] != 0 for r in results])
-    await post(
+    await slack.post(
         f"Deployment from `{source}` for `{settings.CLUSTER_NAME}` results summary:\n"
         f"\t• {n_success} succeeded\n"
         f"\t• {n_failed} failed"
@@ -72,6 +84,7 @@ def __init__(
         self.commit_message = commit_message
         self.current_app_definitions = current_app_definitions
         self.previous_app_definitions = previous_app_definitions
+        self.deploy_id = str(uuid.uuid4())
 
         # Max parallel helm installs at a time
         # Kube api may rate limit otherwise
@@ -130,10 +143,15 @@ async def uninstall_app(self, app: App) -> UpdateAppResult:
                 f"helm uninstall {app.name} -n {app.values['namespace']}", suppress_errors=True
             )
             update_result = UpdateAppResult(app_name=app.name, **result)
-            await post_result(self.current_app_definitions.name, update_result)
+            await post_result(app, self.current_app_definitions.name, update_result)
         return update_result
 
     async def update_app_deployment(self, app: App) -> Optional[UpdateAppResult]:
+        app.set_value("deployment.labels.gitops/deploy_id", self.deploy_id)
+        app.set_value("deployment.labels.gitops/status", github.STATUSES.in_progress)
+        if github_deployment_url := app.values.get("github/deployment_url"):
+            app.set_value("deployment.annotations.github/deployment_url", github_deployment_url)
+
         async with self.semaphore:
             logger.info(f"Deploying app {app.name!r}.")
             if app.chart.type == "git":
@@ -176,7 +194,8 @@ async def update_app_deployment(self, app: App) -> Optional[UpdateAppResult]:
                 return None
 
             update_result = UpdateAppResult(app_name=app.name, **result)
-            await post_result(self.current_app_definitions.name, update_result)
+
+            await post_result(app, self.current_app_definitions.name, update_result)
         return update_result
 
     def calculate_app_deltas(self):
diff --git a/gitops_server/deployment_checker.py b/gitops_server/deployment_checker.py
@@ -0,0 +1,134 @@
+"""
+This async worker keeps tracks of ongoing deployments by polling k8s space (CLUSTER_NAMESPACE).
+
+The worker polls and checks for deployments with the following labels:  gitops/deploy_id   and  gitops/status=in_progress
+
+Any failing/successfully deployed deployment is notified to github.
+This mechanism will only work if the annotation: github/deployment_url is applied to the deployment.
+
+TODO
+- Update the slack summary message from deploy.py
+- @notify the user if the deployment failed
+"""
+import asyncio
+import logging
+
+import kubernetes_asyncio
+import kubernetes_asyncio.client
+import kubernetes_asyncio.config
+
+from . import github
+from .settings import CLUSTER_NAMESPACE
+
+logger = logging.getLogger("deployment_status")
+
+
+class DeploymentStore:
+    def __init__(self):
+        pass
+
+
+async def get_ingress_url(api, namespace: str, app: str):
+    """Attempts to get domain for the ingress associated with the app"""
+    ingresses = await kubernetes_asyncio.client.NetworkingV1beta1Api(api).list_namespaced_ingress(
+        namespace=namespace, label_selector=f"app={app}"
+    )
+    environment_url = ""
+    if ingresses.items:
+        try:
+            environment_url: str = "https://" + ingresses.items[0].spec.rules[0].host
+        except Exception:
+            logger.warning(f"Could not find ingress for {app=}")
+            pass
+    return environment_url
+
+
+class DeploymentStatusWorker:
+    """Watches for deployments and updates the github deployment status"""
+
+    _worker = None
+
+    @classmethod
+    def get_worker(
+        cls,
+    ):
+        if not cls._worker:
+            loop = asyncio.get_running_loop()
+            cls._worker = cls(loop)
+        return cls._worker
+
+    def __init__(self, loop):
+        self.loop = loop
+
+    async def load_config(self):
+        logger.info("Loading kubernetes asyncio api")
+        try:
+            kubernetes_asyncio.config.load_incluster_config()
+        except kubernetes_asyncio.config.config_exception.ConfigException:
+            await kubernetes_asyncio.config.load_kube_config()
+
+    async def process_work(self):
+        await self.load_config()
+
+        while True:
+            async with kubernetes_asyncio.client.ApiClient() as api:
+                apps_api = kubernetes_asyncio.client.AppsV1Api(api)
+                deployments = await apps_api.list_namespaced_deployment(
+                    # Only things that have gitops/deploy_id aka was deployed
+                    namespace=CLUSTER_NAMESPACE,
+                    label_selector="gitops/deploy_id,gitops/status=in_progress",
+                )
+                await asyncio.sleep(10)
+
+                for deployment in deployments.items:
+                    app = deployment.metadata.labels["app"]
+                    namespace = deployment.metadata.namespace
+                    github_deployment_url = deployment.metadata.annotations.get(
+                        "github/deployment_url"
+                    )
+                    conds = {}
+                    for x in deployment.status.conditions:
+                        conds[x.type] = x
+                    status = None
+                    if (
+                        len(conds) == 2
+                        and conds["Available"].status == "True"
+                        and conds["Progressing"].status == "True"
+                        and conds["Progressing"].reason == "NewReplicaSetAvailable"
+                    ):
+                        status = github.STATUSES.success
+                        await github.update_deployment(
+                            github_deployment_url,
+                            status=status,
+                            description="Deployed successfully",
+                            environment_url=await get_ingress_url(api, namespace, app),
+                        )
+                    elif (
+                        "Progressing" in conds
+                        and conds["Progressing"].status == "False"
+                        and conds["Progressing"].reason == "ProgressDeadlineExceeded"
+                    ):
+                        status = github.STATUSES.failure
+                        await github.update_deployment(
+                            github_deployment_url,
+                            status=status,
+                            description="Failed to deploy. Check the pod or migrations.",
+                        )
+                    if status:
+                        logger.info(
+                            f"Patching {deployment.metadata.name}.label.gitops/status to {status}"
+                        )
+                        deployment.metadata.labels["gitops/status"] = status
+                        try:
+                            await apps_api.patch_namespaced_deployment(
+                                deployment.metadata.name, deployment.metadata.namespace, deployment
+                            )
+                        except kubernetes_asyncio.client.exceptions.ApiException as e:
+                            logger.warning(e, exc_info=True)
+
+    async def run(self):
+        logger.info("Starting deployment status watching loop")
+        try:
+            await self.process_work()
+        except Exception as e:
+            logger.error(str(e), exc_info=True)
diff --git a/gitops_server/github.py b/gitops_server/github.py
@@ -0,0 +1,52 @@
+import logging
+import os
+
+import httpx
+
+logger = logging.getLogger("github")
+
+GITHUB_OAUTH_TOKEN = os.environ.get("GITHUB_OAUTH_TOKEN")
+
+
+class STATUSES:
+    pending = "pending"  # default status when created during gh workflow
+    in_progress = "in_progress"  # when helm installs
+    success = "success"  # when helm deployed
+    failure = "failure"  # when error during helm install
+    error = "error"  # when error during helm deploy
+
+
+async def update_deployment(deployment_url: str, status: str, description: str, environment_url=""):
+    # https://docs.github.com/en/rest/reference/repos#create-a-deployment-status
+    if not deployment_url:
+        return
+    status_url = deployment_url + "/statuses"
+    headers = {
+        "Authorization": f"token {GITHUB_OAUTH_TOKEN}",
+        "Content-Type": "application/json",
+        "Accept": (
+            "application/vnd.github.flash-preview+json, application/vnd.github.ant-man-preview+json"
+        ),
+    }
+
+    logger.info(f"Updating deployment status of: {deployment_url} to {status}")
+    async with httpx.AsyncClient() as client:
+        data = {
+            "state": status,
+            "description": description,
+            # https://github.com/chrnorm/deployment-status/issues/13
+            "environment_url": environment_url,
+        }
+        response = await client.post(
+            status_url,
+            json=data,
+            headers=headers,
+        )
+        if response.status_code >= 300:
+            try:
+                logger.warn(response.json())
+            except Exception:
+                pass
+            logger.exception(
+                "Failed to update github deployment", exc_info=True, extra=response.__dict__
+            )
diff --git a/gitops_server/logging_config.py b/gitops_server/logging_config.py
@@ -3,11 +3,11 @@
 
 class EndpointFilter(logging.Filter):
     def filter(self, record: logging.LogRecord) -> bool:
-        return record.args[2] != "/"
+        return record.args[2] != "/"  # type: ignore
 
 
-logging_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-logging.basicConfig(format=logging_format, level=logging.DEBUG)
+logging_format = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
+logging.basicConfig(format=logging_format, level=logging.INFO)
 
 # Filter out / from access logs (We don't care about these calls)
 logging.getLogger("uvicorn.access").addFilter(EndpointFilter())
diff --git a/gitops_server/main.py b/gitops_server/main.py
@@ -1,3 +1,4 @@
+import asyncio
 import hashlib
 import hmac
 import logging
@@ -6,8 +7,9 @@
 
 from gitops_server import settings
 from gitops_server.app import app
+from gitops_server.deployment_checker import DeploymentStatusWorker
 from gitops_server.logging_config import *  # noqa
-from gitops_server.worker import get_worker  # noqa
+from gitops_server.worker import Worker  # noqa
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("gitops")
@@ -28,12 +30,26 @@ async def webhook(request: Request):
 
     json = await request.json()
 
-    worker = get_worker()
+    worker = Worker.get_worker()
 
     await worker.enqueue(json)
     return {"enqueued": True}
 
 
+@app.on_event("startup")
+async def startup_event():
+    """Prepare the worker.
+
+    Creates a new worker object and launches it as a future task.
+    """
+    loop = asyncio.get_running_loop()
+    worker = Worker.get_worker()
+    worker.task = asyncio.ensure_future(worker.run(), loop=loop)
+
+    deployment_status_worker = DeploymentStatusWorker.get_worker()
+    deployment_status_worker.task = asyncio.ensure_future(deployment_status_worker.run(), loop=loop)
+
+
 def get_digest(data: bytes) -> str:
     """Calculate the digest of a webhook body.
 
diff --git a/gitops_server/settings.py b/gitops_server/settings.py
@@ -1,5 +1,7 @@
 import os
 
 CLUSTER_NAME = os.getenv("CLUSTER_NAME", "")
+# Namespace to search/deploy into
+CLUSTER_NAMESPACE = os.getenv("CLUSTER_NAMESPACE", "")
 ACCOUNT_ID = os.getenv("ACCOUNT_ID", "")
 GITHUB_WEBHOOK_KEY = os.getenv("GITHUB_WEBHOOK_KEY", "")
diff --git a/gitops_server/slack.py b/gitops_server/slack.py
@@ -18,5 +18,5 @@ async def post(message):
     async with httpx.AsyncClient() as client:
         response = await client.post(url, json=data)
         if response.status_code >= 300:
-            logger.error("Failed to post a message to slack (see below):")
+            logger.warning("Failed to post a message to slack (see below):")
             logger.error(f"{message}", exc_info=True)
diff --git a/gitops_server/worker.py b/gitops_server/worker.py
diff --git a/poetry.lock b/poetry.lock
diff --git a/pyproject.toml b/pyproject.toml
diff --git a/tests/test_deploy.py b/tests/test_deploy.py

-Original file line number
+Diff line change
 import asyncio
 import logging
 -from gitops_server.app import app
+-
 from .deploy import Deployer
 logger = logging.getLogger("gitops_worker")
 class Worker:
 -    """Simple syncrhonous background work queue.
 +    """Simple synchronous background work queue.
     Deployments need to be carried out one at a time to ensure the cluster
     doesn't get confused. The worker is based entirely on asyncio and runs
     alongside the server for maximum efficiency.
     """
 +    _worker = None
++
 +    @classmethod
 +    def get_worker(cls):
 +        if not cls._worker:
 +            loop = asyncio.get_running_loop()
 +            cls._worker = cls(loop)
 +        return cls._worker
++
     def __init__(self, loop):
         self.loop = loop
         self.queue = asyncio.Queue(loop=self.loop)
         awaited here to ensure synchronous operation.
         # TODO: Need to gracefully handle termination.
         """
 +        logger.info("Starting up deployer worker loop")
         while True:
             try:
                 await self.process_work()
         if ref == "refs/heads/master":
             deployer = await Deployer.from_push_event(work)
             await deployer.deploy()
+-
+-
 -def get_worker():
 -    global worker
 -    return worker
+-
+-
 -worker = None
+-
+-
 -@app.on_event("startup")
 -async def startup_event():
 -    """Prepare the worker.
+-
 -    Creates a new worker object and launches it as a future task.
 -    """
 -    loop = asyncio.get_running_loop()
 -    logger.info("Starting up worker")
 -    global worker
 -    worker = Worker(loop)
 -    worker.task = asyncio.ensure_future(worker.run(), loop=loop)
-Original file line number
+Diff line change
 humanize = "^3.5.0"
 tabulate = "^0.8.9"
 # Server requirements
 -fastapi = {version = "*", extras = ["server"]}
 -httpx = {version = "^0.18.1", extras = ["server"]}
 -uvicorn = {version = "^0.13.4", extras = ["server"]}
 -sentry-sdk = {version = "^1.3.0", extras = ["server"]}
 +fastapi = { version = "*", optional = true }
 +httpx = { version = "^0.18.1", optional = true }
 +uvicorn = { version = "^0.13.4", optional = true }
 +sentry-sdk = { version = "^1.3.0", optional = true }
 +kubernetes_asyncio = { version = "^12.1.2", optional = true }
++
 +[tool.poetry.extras]
 +server = [
 +    "fastapi",
 +    "uvicorn",
 +    "httpx",
 +    "uvicorn",
 +    "sentry-sdk",
 +    "kubernetes_asyncio",
 +]
 [tool.poetry.dev-dependencies]
 #Flake 9 is flake 8 with pyproject support =_=.
-Original file line number
+Diff line change
 class TestDeploy(TestCase):
     @patch("gitops_server.deploy.run")
 -    @patch("gitops_server.deploy.post")
 +    @patch("gitops_server.slack.post")
     @patch("gitops_server.deploy.load_app_definitions", mock_load_app_definitions)
     @patch("gitops_server.deploy.temp_repo")
     async def test_deployer_git(self, temp_repo_mock, post_mock, run_mock):