spark_on_k8s.api.webserver

Implementation reference. These symbols describe how the project works; their presence here does not make them supported application imports. Use the documented entry points and support policy.

current_dir = Path(__file__).parent.absolute() module-attribute

router = APIRouter(prefix='/webserver', tags=['spark-apps', 'webserver'], include_in_schema=False) module-attribute

templates = Jinja2Templates(directory=str(current_dir / 'templates')) module-attribute

app_logs(request, namespace, app_id, tail=Query(default=1000, ge=-1, le=10000)) async

Display a bounded, searchable driver-log stream.

Source code in spark_on_k8s/api/webserver/__init__.py

@router.get("/logs/{namespace}/{app_id}", response_class=HTMLResponse)
async def app_logs(
    request: Request, namespace: str, app_id: str, tail: int = Query(default=1000, ge=-1, le=10000)
):
    """Display a bounded, searchable driver-log stream."""
    namespace_scope().require(namespace)
    return console_page(request, view="logs", namespace=namespace, app_id=app_id, tail=tail)

app_logs_websocket(websocket, namespace, app_id, tail=Query(default=-1, ge=-1, le=10000)) async

Stream literal log text and signal failures through the WebSocket close state.

Source code in spark_on_k8s/api/webserver/__init__.py

@router.websocket("/ws/logs/{namespace}/{app_id}")
async def app_logs_websocket(
    websocket: WebSocket, namespace: str, app_id: str, tail: int = Query(default=-1, ge=-1, le=10000)
):
    """Stream literal log text and signal failures through the WebSocket close state."""
    if not _logs_origin_allowed(websocket):
        await websocket.close(code=1008, reason="Browser origin is not permitted.")
        return
    if not namespace_scope().allows(namespace):
        note(code="namespace_forbidden")
        await websocket.close(code=1008, reason="The namespace is outside this API's namespace scope.")
        return
    await websocket.accept()
    manager = AsyncSparkAppManager(k8s_client_manager=KubernetesClientSingleton.client_manager)

    async def send_logs():
        async with aclosing(
            manager.logs_streamer(namespace=namespace, app_id=app_id, tail_lines=tail)
        ) as logs:
            async for log in logs:
                await websocket.send_text(log)

    async def wait_for_disconnect():
        while True:
            message = await websocket.receive()
            if message["type"] == "websocket.disconnect":
                return

    producer = asyncio.create_task(send_logs())
    disconnected = asyncio.create_task(wait_for_disconnect())
    code, reason = 1000, ""
    try:
        finished, _ = await asyncio.wait({producer, disconnected}, return_when=asyncio.FIRST_COMPLETED)
        if disconnected in finished:
            await disconnected
            return
        await producer
    except WebSocketDisconnect:
        return
    except ApiException as error:
        count_kubernetes_error(error)
        code = 1008 if error.status in (403, 404) else 1011
        reason = "Driver logs are unavailable."
    except ValueError:
        code, reason = 1008, "The application is no longer available."
    except Exception:
        code, reason = 1011, "Unable to stream driver logs."
    finally:
        producer.cancel()
        disconnected.cancel()
        await asyncio.gather(producer, disconnected, return_exceptions=True)
    try:
        await websocket.close(code=code, reason=reason)
    except WebSocketDisconnect:
        pass

apps(request) async

Display the application console, with namespace preserved in the URL.

Source code in spark_on_k8s/api/webserver/__init__.py

@router.get("/apps", response_class=HTMLResponse)
async def apps(request: Request):
    """Display the application console, with namespace preserved in the URL."""
    scope = namespace_scope()
    namespace = scope.require(request.query_params.get("namespace") or scope.default)
    return console_page(request, view="apps", namespace=namespace)

console_page(request, *, view, namespace, app_id=None, tail=1000)

Serve the packaged UI; fetch application data without reloading the page.

Source code in spark_on_k8s/api/webserver/__init__.py

def console_page(request: Request, *, view: str, namespace: str, app_id: str | None = None, tail: int = 1000):
    """Serve the packaged UI; fetch application data without reloading the page."""
    scope = namespace_scope()
    config = {
        "basePath": request.scope.get("root_path", "").rstrip("/"),
        "namespace": namespace,
        # A listed scope becomes a picker; a single namespace is not editable at all.
        "namespaces": list(scope.namespaces) if scope.namespaces is not None else None,
        "namespaceEditable": scope.namespaces is None or len(scope.namespaces) > 1,
        "refreshInterval": APIConfiguration.SPARK_ON_K8S_API_WEBSERVER_APPS_REFRESH_INTERVAL,
        "allowKill": APIConfiguration.SPARK_ON_K8S_API_WEBSERVER_APPS_KILL_BUTTON,
        "allowDelete": APIConfiguration.SPARK_ON_K8S_API_WEBSERVER_APPS_DELETE_BUTTON,
        "historyEnabled": bool(APIConfiguration.SPARK_ON_K8S_API_SPARK_HISTORY_HOST),
        "runsEnabled": get_history_settings().enabled,
        "version": __version__,
        "view": view,
        "appId": app_id,
        "tail": tail,
    }
    return templates.TemplateResponse(
        request=request,
        name="console.html",
        context={"config": config},
        headers={"Cache-Control": "no-cache"},
    )

handle_k8s_errors(func)

Source code in spark_on_k8s/api/webserver/__init__.py

def handle_k8s_errors(func):
    @wraps(func)
    async def wrapper(*args, request: Request, **kwargs):
        try:
            return await func(*args, request=request, **kwargs)
        except ApiException as e:
            count_kubernetes_error(e)
            title = f"{e.status} {e.reason}"
            message = e.body
            return templates.TemplateResponse(
                request=request,
                name="error.html",
                context={
                    "title": title,
                    "message": message,
                },
                status_code=e.status or 500,
            )
        except httpx.HTTPError as e:
            # Finished drivers and restarting history servers are upstream failures, not API errors.
            # Exception text can contain internal resolver/upstream details; report only its type.
            logger.warning("Spark UI proxy upstream request failed: %s", type(e).__name__)
            if isinstance(e, httpx.TimeoutException):
                raise HTTPException(504, detail="The Spark UI did not respond in time.") from e
            raise HTTPException(502, detail="The Spark UI is unavailable.") from e

    return wrapper

runs_page(request, namespace=None) async

Source code in spark_on_k8s/api/webserver/__init__.py

@router.get("/runs", response_class=HTMLResponse)
async def runs_page(request: Request, namespace: str | None = None):
    settings = get_history_settings()
    if not settings.enabled:
        raise HTTPException(status_code=404, detail="Run history is disabled.")
    # The console links here from any namespace in the API scope, which may be wider than the
    # recorded ones; open a recorded namespace instead of an error page. /runs/v1 still refuses it.
    if namespace not in settings.namespaces:
        namespace = settings.namespaces[0]
    return console_page(request, view="runs", namespace=namespace)

spark_history_reverse_proxy(request, path) async

Source code in spark_on_k8s/api/webserver/__init__.py

@router.get("/ui-history/{path:path}")
@handle_k8s_errors
async def spark_history_reverse_proxy(request: Request, path: str):
    spark_history_url = APIConfiguration.SPARK_ON_K8S_API_SPARK_HISTORY_HOST
    if not spark_history_url:
        raise HTTPException(status_code=412, detail="SPARK_ON_K8S_API_SPARK_HISTORY_HOST is not set.")
    # Match the console listing: a scheme is explicit, not inferred from a host name starting "http".
    spark_history_url = spark_history_url if "://" in spark_history_url else f"http://{spark_history_url}"
    url = _upstream_url(spark_history_url, f"/{path}", request)
    async_http_client = AsyncHttpClientSingleton.client()
    reverse_proxy_req = async_http_client.build_request(
        request.method, url=url, headers=_proxy_headers(request), content=request.stream()
    )
    reverse_proxy_resp = await async_http_client.send(reverse_proxy_req, stream=True)
    return await _proxy_response(reverse_proxy_resp, url, request, "/webserver/ui-history")

ui_reverse_proxy(request, path) async

Source code in spark_on_k8s/api/webserver/__init__.py

@router.get("/ui/{path:path}")
@handle_k8s_errors
async def ui_reverse_proxy(request: Request, path: str):
    parts = path.split("/", 2)
    if len(parts) < 2 or not all(parts[:2]):
        raise HTTPException(status_code=404, detail="A namespace and Spark service are required.")
    namespace, service_name = parts[:2]
    if not all(_DNS_LABEL.fullmatch(value) for value in (namespace, service_name)):
        raise HTTPException(status_code=400, detail="Invalid Kubernetes namespace or service name.")
    # The proxy connects to Services directly, so Kubernetes RBAC cannot bound it; the scope does.
    namespace_scope().require(namespace)
    path = "/" + parts[2] if len(parts) == 3 else "/"
    url = _upstream_url(f"http://{service_name}.{namespace}.svc.cluster.local:4040", path, request)
    async_http_client = AsyncHttpClientSingleton.client()
    reverse_proxy_req = async_http_client.build_request(
        request.method, url=url, headers=_proxy_headers(request), content=request.stream()
    )
    reverse_proxy_resp = await async_http_client.send(reverse_proxy_req, stream=True)
    return await _proxy_response(
        reverse_proxy_resp, url, request, f"/webserver/ui/{namespace}/{service_name}"
    )