aks

azure_bootstrap.aks

AKS runtime helpers — graceful shutdown, build info, probes, pod context.

Modules:

Name	Description
leader_election	Generic leader election — K8s ConfigMap lock or soft no-op.

Functions:

Name	Description
install_sigterm_handler	Register SIGTERM handler that sets stop_event for graceful drain.
setup_async_sigterm_handler	Return an asyncio.Event set on SIGTERM.
build_info	Return version/build metadata from downward-API or CI env vars.
keda_metric_value	Scalar metric for KEDA metrics-api scaler (from counters snapshot).
mount_build_info_route	Mount a FastAPI route returning :func:build_info.
pod_context_extra	Correlation/log extras from Kubernetes downward API env vars.

install_sigterm_handler

install_sigterm_handler(stop_event: Event) -> None

Register SIGTERM handler that sets stop_event for graceful drain.

Source code in azure_bootstrap/aks/__init__.py

def install_sigterm_handler(stop_event: threading.Event) -> None:
    """Register SIGTERM handler that sets *stop_event* for graceful drain."""

    def _handler(signum: int, frame: Any) -> None:
        _logger.info("SIGTERM received — setting stop event")
        stop_event.set()

    signal.signal(signal.SIGTERM, _handler)

setup_async_sigterm_handler

setup_async_sigterm_handler() -> Event

Return an asyncio.Event set on SIGTERM.

Source code in azure_bootstrap/aks/__init__.py

def setup_async_sigterm_handler() -> asyncio.Event:
    """Return an asyncio.Event set on SIGTERM."""
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()

    def _handler() -> None:
        _logger.info("SIGTERM received — setting async stop event")
        stop.set()

    loop.add_signal_handler(signal.SIGTERM, _handler)
    return stop

build_info

build_info() -> dict[str, str | None]

Return version/build metadata from downward-API or CI env vars.

Source code in azure_bootstrap/aks/__init__.py

def build_info() -> dict[str, str | None]:
    """Return version/build metadata from downward-API or CI env vars."""
    return {
        "version": os.environ.get("BUILD_VERSION") or os.environ.get("APP_VERSION"),
        "git_sha": os.environ.get("GIT_SHA"),
        "build_time": os.environ.get("BUILD_TIME"),
        "image_tag": os.environ.get("IMAGE_TAG"),
        "pod_name": os.environ.get("POD_NAME"),
        "pod_namespace": os.environ.get("POD_NAMESPACE"),
        "node_name": os.environ.get("NODE_NAME"),
    }

keda_metric_value

keda_metric_value(name: str = 'queue_depth') -> float

Scalar metric for KEDA metrics-api scaler (from counters snapshot).

Source code in azure_bootstrap/aks/__init__.py

def keda_metric_value(name: str = "queue_depth") -> float:
    """Scalar metric for KEDA metrics-api scaler (from counters snapshot)."""
    snap = counter_snapshot()
    return float(snap.get(name, 0))

mount_build_info_route

mount_build_info_route(app: Any, path: str = '/api/version') -> None

Mount a FastAPI route returning :func:build_info.

Source code in azure_bootstrap/aks/__init__.py

def mount_build_info_route(app: Any, path: str = "/api/version") -> None:
    """Mount a FastAPI route returning :func:`build_info`."""

    @app.get(path)
    def _version() -> dict[str, str | None]:
        return build_info()

pod_context_extra

pod_context_extra() -> dict[str, str | None]

Correlation/log extras from Kubernetes downward API env vars.

Source code in azure_bootstrap/aks/__init__.py

def pod_context_extra() -> dict[str, str | None]:
    """Correlation/log extras from Kubernetes downward API env vars."""
    return {
        "pod_name": os.environ.get("POD_NAME"),
        "pod_namespace": os.environ.get("POD_NAMESPACE"),
        "node_name": os.environ.get("NODE_NAME"),
    }