Smoke Test Routing for Microservice DR Drills

#!/usr/bin/env python3
"""Header-scoped smoke test router for microservice DR drills.

Provisions an Istio VirtualService that steers a signed drill-context header to
an isolated restore sandbox, runs a smoke test through the mesh ingress, and
tears the route down regardless of outcome.

Exit codes (consumed by the DR drill runner):
    0  the route applied, the sandbox answered, and the smoke test passed
    1  the smoke test failed, or the sandbox was unreachable during the drill
    2  usage / configuration / cluster-access error -> abort the drill
"""
import hashlib
import hmac
import logging
import os
import sys
import uuid
from enum import IntEnum
from typing import Any, Dict

import requests
from kubernetes import client, config
from kubernetes.client.rest import ApiException

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("dr_smoke_router")

GROUP = "networking.istio.io"
VERSION = "v1beta1"
PLURAL = "virtualservices"


class SmokeStatus(IntEnum):
    PASS = 0
    FAIL = 1
    ABORT = 2


class DrillTrafficRouter:
    def __init__(self, namespace: str, service: str, sandbox_host: str, signing_key: bytes):
        try:
            config.load_incluster_config()
        except config.ConfigException:
            config.load_kube_config()
        self.api = client.CustomObjectsApi()
        self.namespace = namespace
        self.service = service
        self.sandbox_host = sandbox_host
        self.drill_id = str(uuid.uuid4())
        # HMAC the UUID so an unprivileged client cannot forge a routable header.
        self.drill_context = hmac.new(
            signing_key, self.drill_id.encode(), hashlib.sha256
        ).hexdigest()
        self.route_name = f"{service}-drill-route-{self.drill_id[:8]}"

    @property
    def prod_host(self) -> str:
        return f"{self.service}.{self.namespace}.svc.cluster.local"

    def _manifest(self) -> Dict[str, Any]:
        return {
            "apiVersion": f"{GROUP}/{VERSION}",
            "kind": "VirtualService",
            "metadata": {
                "name": self.route_name,
                "namespace": self.namespace,
                "labels": {
                    "app.kubernetes.io/managed-by": "drill-orchestrator",
                    "drill-id": self.drill_id,
                },
            },
            "spec": {
                "hosts": [self.prod_host],
                "http": [
                    {
                        "match": [
                            {"headers": {"x-drill-context": {"exact": self.drill_context}}}
                        ],
                        "route": [
                            {"destination": {"host": self.sandbox_host, "port": {"number": 8080}}}
                        ],
                        "timeout": "30s",
                        "retries": {"attempts": 2, "perTryTimeout": "10s", "retryOn": "5xx"},
                    },
                    {
                        "route": [
                            {"destination": {"host": self.prod_host, "port": {"number": 8080}}}
                        ]
                    },
                ],
            },
        }

    def apply_route(self) -> None:
        body = self._manifest()
        try:
            self.api.create_namespaced_custom_object(
                GROUP, VERSION, self.namespace, PLURAL, body
            )
            logger.info("Route applied: %s (drill %s)", self.route_name, self.drill_id)
        except ApiException as exc:
            if exc.status == 409:  # a retried drill left the route in place
                logger.warning("Route exists; patching to converge state")
                self.api.patch_namespaced_custom_object(
                    GROUP, VERSION, self.namespace, PLURAL, self.route_name, body
                )
            else:
                raise

    def teardown_route(self) -> None:
        try:
            self.api.delete_namespaced_custom_object(
                GROUP, VERSION, self.namespace, PLURAL, self.route_name
            )
            logger.info("Route removed: %s", self.route_name)
        except ApiException as exc:
            if exc.status == 404:
                logger.warning("Route already absent; nothing to tear down")
            else:
                raise

    def run_smoke_test(self, ingress_url: str) -> bool:
        headers = {
            "X-Drill-Context": self.drill_context,
            "Host": self.prod_host,
            "Content-Type": "application/json",
        }
        payload = {"test": "backup_validation", "drill_id": self.drill_id}
        try:
            resp = requests.post(
                f"{ingress_url}/health", json=payload, headers=headers, timeout=30
            )
        except requests.RequestException as exc:
            logger.error("Smoke test transport error: %s", exc)
            return False
        if resp.status_code == 503:
            # Isolation guarantee: an unreachable sandbox must NOT fail over to prod.
            logger.error("Sandbox unreachable (503); no failover to production")
            return False
        if resp.status_code != 200:
            logger.error("Smoke test returned HTTP %s", resp.status_code)
            return False
        logger.info("Smoke test passed: HTTP 200 from restore sandbox")
        return True


def main() -> int:
    if len(sys.argv) != 5:
        logger.error(
            "Usage: dr_smoke_router.py <namespace> <service> <sandbox_host> <ingress_url>"
        )
        return int(SmokeStatus.ABORT)

    signing_key = os.environ.get("DRILL_SIGNING_KEY")
    if not signing_key:
        logger.error("DRILL_SIGNING_KEY is not set; cannot sign the drill context")
        return int(SmokeStatus.ABORT)

    namespace, service, sandbox_host, ingress_url = sys.argv[1:5]
    try:
        router = DrillTrafficRouter(namespace, service, sandbox_host, signing_key.encode())
    except config.ConfigException as exc:
        logger.error("Cluster access error: %s", exc)
        return int(SmokeStatus.ABORT)

    try:
        router.apply_route()
    except ApiException as exc:
        logger.error("Failed to apply route: %s", exc)
        return int(SmokeStatus.ABORT)

    try:
        passed = router.run_smoke_test(ingress_url)
    finally:
        router.teardown_route()

    if not passed:
        logger.critical("Smoke test failed; drill red")
        return int(SmokeStatus.FAIL)

    logger.info("Route validated and torn down; drill green")
    return int(SmokeStatus.PASS)


if __name__ == "__main__":
    sys.exit(main())

2026-07-05 05:14:02 | INFO | dr_smoke_router | Route applied: api-gateway-drill-route-3f9a1c04 (drill 3f9a1c04-...)
2026-07-05 05:14:03 | INFO | dr_smoke_router | Smoke test passed: HTTP 200 from restore sandbox
2026-07-05 05:14:03 | INFO | dr_smoke_router | Route removed: api-gateway-drill-route-3f9a1c04
2026-07-05 05:14:03 | INFO | dr_smoke_router | Route validated and torn down; drill green

2026-07-05 05:14:02 | INFO | dr_smoke_router | Route applied: api-gateway-drill-route-8b1d77e2 (drill 8b1d77e2-...)
2026-07-05 05:14:32 | ERROR | dr_smoke_router | Sandbox unreachable (503); no failover to production
2026-07-05 05:14:32 | INFO | dr_smoke_router | Route removed: api-gateway-drill-route-8b1d77e2
2026-07-05 05:14:32 | CRITICAL | dr_smoke_router | Smoke test failed; drill red

#!/usr/bin/env bash
set -euo pipefail

python3 dr_smoke_router.py dr-drill-sandbox api-gateway \
  api-gateway.dr-sandbox.svc.cluster.local \
  http://istio-ingressgateway.istio-system.svc.cluster.local
case $? in
  0) echo "[$(date -u)] smoke test green; advancing drill" ;;
  1) curl -s -X POST "$PAGERDUTY_WEBHOOK" -d '{"event":"dr_smoke_test_failed"}'
     exit 1 ;;
  *) echo "[$(date -u)] router misconfigured; aborting"; exit 2 ;;
esac