Function get_cluster_manifests

gpustack/routes/clusters.py:728–781

(
    request: Request,
    session: SessionDep,
    id: int,
    runtime: Optional[List[ManufacturerEnum]] = Query(
        None,
        description=(
            "GPU vendor runtimes to include in the manifest. Repeat the "
            "parameter for multiple vendors (e.g. ?runtime=nvidia&runtime=ascend). "
            "The CPU worker DaemonSet is always rendered regardless of this "
            "parameter."
        ),
    ),
)

@router.get("/{id}/manifests")
async def get_cluster_manifests(
    request: Request,
    session: SessionDep,
    id: int,
    runtime: Optional[List[ManufacturerEnum]] = Query(
        None,
        description=(
            "GPU vendor runtimes to include in the manifest. Repeat the "
            "parameter for multiple vendors (e.g. ?runtime=nvidia&runtime=ascend). "
            "The CPU worker DaemonSet is always rendered regardless of this "
            "parameter."
        ),
    ),
):
    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"Cannot get manifests for cluster {cluster.name}(id: {id}) with provider {cluster.provider}"
        )
    # TODO: Redundant principal names at the cluster level to reduce multiple queries.
    principal = await Principal.one_by_id(session, cluster.owner_principal_id)
    if not principal:
        raise NotFoundException(
            message=(
                f"Owner principal (id: {cluster.owner_principal_id}) of cluster "
                f"{cluster.name}(id: {id}) not found"
            )
        )

    # Resolve server-wide defaults onto a copy of k8s_options so the render
    # model only ever reads k8s_options. Copy (not mutate) the loaded cluster
    # so we don't risk persisting these derived values back to the DB.
    cfg = get_global_config()
    k8s_options = (cluster.k8s_options or K8sOptions()).model_copy()
    if not k8s_options.namespace:
        k8s_options.namespace = cfg.namespace
    if not k8s_options.operator_image:
        k8s_options.operator_image = cfg.operator_image

    config = TemplateConfig(
        registration=get_registration_from_cluster(request, cluster),
        cluster_owner_principal_identifier=principal_namespace_identifier(principal),
        runtimes=runtime,
        k8s_options=k8s_options,
        system_default_container_registry=cluster.system_default_container_registry,
    )
    yaml_content = config.render()
    return Response(
        content=yaml_content,
        media_type="application/x-yaml",
        headers={"Content-Disposition": "attachment; filename=manifest.yaml"},
    )


@router.get("/{id}/dashboard")