host/online: Start Ansible job to provision host

The _POST /host/online_ webhook now creates a Kubernetes Job to run the
host provisioner.  The Job resource is defined in a YAML document, and
will be created in the Kubernetes namespace specified by the
`ANSIBLE_JOB_NAMESPACE` environment variable (defaults to `ansible`).

dch_webhooks.py

@@ -7,8 +7,9 @@ import json
 import logging
 import os
 import re
+from pathlib import Path
 from types import TracebackType
-from typing import Optional, Self, Type
+from typing import Any, Optional, Self, Type
 
 import fastapi
 import httpx
@@ -16,6 +17,7 @@ import pika
 import pika.channel
 import pydantic
 import pyrfc6266
+import ruamel.yaml
 from fastapi import Form
 from pika.adapters.asyncio_connection import AsyncioConnection
 
@@ -60,6 +62,21 @@ PAPERLESS_URL = os.environ.get(
     'http://paperless-ngx',
 )
 
+ANSIBLE_JOB_YAML = Path(os.environ.get('ANSIBLE_JOB_YAML', 'ansible-job.yaml'))
+ANSIBLE_JOB_NAMESPACE = os.environ.get('ANSIBLE_JOB_NAMESPACE', 'ansible')
+KUBERNETES_TOKEN_PATH = Path(
+    os.environ.get(
+        'KUBERNETES_TOKEN_PATH',
+        '/run/secrets/kubernetes.io/serviceaccount/token',
+    )
+)
+KUBERNETES_CA_CERT = Path(
+    os.environ.get(
+        'KUBERNETES_CA_CERT',
+        '/run/secrets/kubernetes.io/serviceaccount/ca.crt',
+    )
+)
+
 
 class FireflyIIITransactionSplit(pydantic.BaseModel):
     type: str
@@ -112,17 +129,15 @@ class HttpxClientMixin:
             self._client = self._get_client()
         return self._client
 
-    def _get_client(self) -> httpx.AsyncClient:
+    def _get_client(self, **kwargs) -> httpx.AsyncClient:
-        return httpx.AsyncClient(
+        headers = kwargs.setdefault('headers', {})
-            headers={
+        headers['User-Agent'] = f'{DIST["Name"]}/{DIST["Version"]}'
-                'User-Agent': f'{DIST["Name"]}/{DIST["Version"]}',
+        return httpx.AsyncClient(**kwargs)
-            },
-        )
 
 
 class Firefly(HttpxClientMixin):
-    def _get_client(self) -> httpx.AsyncClient:
+    def _get_client(self, **kwargs) -> httpx.AsyncClient:
-        client = super()._get_client()
+        client = super()._get_client(**kwargs)
         if token_file := os.environ.get('FIREFLY_AUTH_TOKEN'):
             try:
                 f = open(token_file, encoding='utf-8')
@@ -166,8 +181,8 @@ class Firefly(HttpxClientMixin):
 
 
 class Paperless(HttpxClientMixin):
-    def _get_client(self) -> httpx.AsyncClient:
+    def _get_client(self, **kwargs) -> httpx.AsyncClient:
-        client = super()._get_client()
+        client = super()._get_client(**kwargs)
         if token_file := os.environ.get('PAPERLESS_AUTH_TOKEN'):
             try:
                 f = open(token_file, encoding='utf-8')
@@ -374,6 +389,34 @@ class AMQPContext:
         self._channel = None
 
 
+class Kubernetes(HttpxClientMixin):
+    @functools.cached_property
+    def base_url(self) -> str:
+        https = True
+        port = os.environ.get('KUBERNETES_SERVICE_PORT_HTTPS')
+        if not port:
+            https = False
+            port = os.environ.get('KUBERNETES_SERVICE_PORT', 8001)
+        host = os.environ.get('KUBERNETES_SERVICE_HOST', '127.0.0.1')
+        url = f'{"https" if https else "http"}://{host}:{port}'
+        log.info('Using Kubernetes URL: %s', url)
+        return url
+
+    @functools.cached_property
+    def token(self) -> str:
+        return KUBERNETES_TOKEN_PATH.read_text().strip()
+
+    def _get_client(self, **kwargs) -> httpx.AsyncClient:
+        if KUBERNETES_CA_CERT.exists():
+            kwargs.setdefault('verify', str(KUBERNETES_CA_CERT))
+        client = super()._get_client(**kwargs)
+        try:
+            client.headers['Authorization'] = f'Bearer {self.token}'
+        except (OSError, UnicodeDecodeError) as e:
+            log.warning('Failed to read k8s auth token: %s', e)
+        return client
+
+
 class Context:
 
     def __init__(self):
@@ -500,7 +543,24 @@ def rfc2047_base64encode(
     return f"=?UTF-8?B?{encoded}?="
 
 
-async def start_ansible_job(): ...
+def load_job_yaml(path: Optional[Path] = None) -> dict[str, Any]:
+    if path is None:
+        path = ANSIBLE_JOB_YAML
+    yaml = ruamel.yaml.YAML()
+    with path.open(encoding='utf-8') as f:
+        return yaml.load(f)
+
+
+async def start_ansible_job():
+    async with Kubernetes() as kube:
+        url = (
+            f'{kube.base_url}/apis/batch/v1/namespaces/'
+            f'{ANSIBLE_JOB_NAMESPACE}/jobs'
+        )
+        job = load_job_yaml()
+        r = await kube.client.post(url, json=job)
+        if r.status_code > 299:
+            raise Exception(r.read())
 
 
 async def publish_host_info(hostname: str, sshkeys: str):
@@ -526,6 +586,12 @@ async def handle_host_online(hostname: str, sshkeys: str):
     except Exception:
         log.exception('Failed to publish host info:')
         return
+    try:
+        await start_ansible_job()
+    except asyncio.CancelledError:
+        raise
+    except Exception:
+        log.exception('Failed to start Ansible job:')
 
 
 app = fastapi.FastAPI(

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "pika>=1.3.2",
     "pyrfc6266~=1.0.2",
     "python-multipart>=0.0.20",
+    "ruamel-yaml>=0.18.10",
 ]
 dynamic = ["version"]