Function cluster_apiserver_proxy

gpustack/routes/clusters.py:838–907

Proxy a request to the Kubernetes API server of a Kubernetes-provider cluster, by forwarding it through one of the cluster's worker pods. The worker uses its in-pod ServiceAccount credentials to call the API server. Uses an inline session instead of SessionDep so the session is rel

(
    request: Request,
    id: int,
    path: str,
)

    include_in_schema=False,
)
async def cluster_apiserver_proxy(
    request: Request,
    id: int,
    path: str,
):
    """
    Proxy a request to the Kubernetes API server of a Kubernetes-provider
    cluster, by forwarding it through one of the cluster's worker pods. The
    worker uses its in-pod ServiceAccount credentials to call the API server.

    Uses an inline session instead of SessionDep so the session is released
    immediately after the initial lookup, preventing long-lived Kubernetes
    watch/log streams from holding a database connection.
    """
    async with async_session() as session:
        cluster = await Cluster.one_by_id(session, id)
        if not cluster or cluster.deleted_at is not None:
            raise NotFoundException(message=f"cluster {id} not found")
        if cluster.provider != ClusterProvider.Kubernetes:
            raise InvalidException(
                message=(
                    f"cluster {cluster.name}(id: {id}) provider is "
                    f"{cluster.provider.value}; API server proxy is only supported "
                    "for Kubernetes-provider clusters."
                )
            )

        workers = await Worker.all_by_fields(
            session,
            fields={"cluster_id": id, "state": WorkerStateEnum.READY},
        )
        if not workers:
            raise ServiceUnavailableException(
                message=f"No reachable workers in cluster {cluster.name}(id: {id})"
            )
        worker = random.choice(workers)
        session.expunge(worker)

    headers = {
        k: v
        for k, v in request.headers.items()
        if k.lower() not in _CLUSTER_PROXY_REQUEST_HEADER_SKIP
    }

    body = None
    if request.method not in ("GET", "HEAD", "OPTIONS"):
        body = await request.body()

    # request.query_params preserves order but a flat dict is sufficient for
    # the Kubernetes API surface we forward (no duplicate keys in practice).
    params = dict(request.query_params) or None

    # No total timeout — Kubernetes watch and log-follow streams may be open
    # indefinitely. Connect timeout still bounds the upstream connect step.
    timeout = aiohttp.ClientTimeout(total=None, sock_connect=10)

    return StreamingResponseWithStatusCode(
        stream_to_worker(