diff --git a/docs/components/frontend/configuration.md b/docs/components/frontend/configuration.md index 2d81352b81af..874a92eb022f 100644 --- a/docs/components/frontend/configuration.md +++ b/docs/components/frontend/configuration.md @@ -161,8 +161,19 @@ The frontend exposes the following HTTP endpoints: | `GET` | `/metrics` | Prometheus metrics | | `GET` | `/openapi.json` | OpenAPI specification | | `GET` | `/docs` | Swagger UI | -| `POST` | `/busy_threshold` | Set busy thresholds | -| `GET` | `/busy_threshold` | Get current busy thresholds | +| `POST` | `/busy_threshold` | Set busy thresholds (gated by `DYN_ENABLE_FRONTEND_ADMIN_API`, see below) | +| `GET` | `/busy_threshold` | Get current busy thresholds (gated by `DYN_ENABLE_FRONTEND_ADMIN_API`, see below) | + +### Frontend feature switches + +Environment variables controlling frontend extensions. Extensions are enabled by default. When deploying, consider whether each is needed for your use case; if not, disable it to prevent accidental abuse. + +Set an env value of `0` / `false` / `no` / `off` (case-insensitive) to disable. + +| Env Var | Default | Behavior when `false` | +|---------|---------|------------------------| +| `DYN_ENABLE_FRONTEND_NVEXT` | `true` | Frontend drops `request.nvext` at handler entry on `/v1/chat/completions`, `/v1/completions`, `/v1/responses`, and `/v1/embeddings`; ignores routing-override headers (`x-worker-instance-id`, `x-prefill-instance-id`, `x-dp-rank`, `x-data-parallel-rank`, `x-prefill-dp-rank`); silently ignores the response-side `nvext.extra_fields` opt-in. Note: disabling this breaks EPP / GAIE serving, Prime-RL-style training that uses `nvext.cache_salt`, multi-tenant agent platforms that forward `nvext.agent_hints` / `nvext.agent_context`, and clients that opt into response disclosure via `nvext.extra_fields`. | +| `DYN_ENABLE_FRONTEND_ADMIN_API` | `true` | `GET /busy_threshold` and `POST /busy_threshold` are not registered (404 instead of 503). Inference, metrics, models, health, and liveness routes are unaffected. | ### Endpoint Path Customization diff --git a/lib/llm/src/http/service/clear_kv_blocks.rs b/lib/llm/src/http/service/clear_kv_blocks.rs deleted file mode 100644 index dcd4352ffb17..000000000000 --- a/lib/llm/src/http/service/clear_kv_blocks.rs +++ /dev/null @@ -1,238 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -use super::{service_v2, RouteDoc}; -use axum::{http::Method, response::IntoResponse, routing::post, Json, Router}; -use serde_json::json; -use std::sync::Arc; - -use dynamo_runtime::{discovery::DiscoveryQuery, pipeline::PushRouter, stream::StreamExt}; - -pub const CLEAR_KV_ENDPOINT: &str = "clear_kv_blocks"; - -pub fn clear_kv_blocks_router( - state: Arc, - path: Option, -) -> (Vec, Router) { - let path = path.unwrap_or_else(|| "/clear_kv_blocks".to_string()); - - let docs: Vec = vec![RouteDoc::new(Method::POST, &path)]; - - let router = Router::new() - .route(&path, post(clear_kv_blocks_handler)) - .with_state(state); - - (docs, router) -} - -async fn clear_kv_blocks_handler( - axum::extract::State(state): axum::extract::State>, -) -> impl IntoResponse { - let model_entries = state.manager().get_model_entries(); - - // if there are no active workers - if model_entries.is_empty() { - return Json(serde_json::json!({ - "message": "No active worker groups found" - })); - } - - let distributed = match state.runtime() { - Some(runtime) => runtime, - None => { - return Json(serde_json::json!({ - "message": "Failed to create distributed runtime", - })); - } - }; - - let mut cleared_workers = Vec::new(); - let mut failed_workers = Vec::new(); - - // update cleared and failed workers - let mut add_worker_result = |success: bool, - name: String, - status: &str, - ns: &str, - comp: &str, - message: Option| { - let mut result = json!({ - "name": name, - "endpoint": format!("{}/{}/{}", ns, comp, CLEAR_KV_ENDPOINT), - "status": status, - }); - if success { - if let Some(m) = message { - result["response"] = json!(m); - } - cleared_workers.push(result); - } else { - if let Some(m) = message { - result["error"] = json!(m); - } - failed_workers.push(result); - } - }; - - // create client for each model entry - for entry in &model_entries { - let namespace = &entry.endpoint_id.namespace; - let component = &entry.endpoint_id.component; - let entry_name = entry.name.to_string(); - - tracing::debug!("Processing worker group: {}/{}", namespace, component); - - let namespace_obj = match distributed.namespace(namespace) { - Ok(ns) => ns, - Err(e) => { - add_worker_result( - false, - entry_name, - "Failed to get namespace", - namespace, - component, - Some(e.to_string()), - ); - continue; - } - }; - - let component_obj = match namespace_obj.component(component) { - Ok(comp) => comp, - Err(e) => { - add_worker_result( - false, - entry_name, - "Failed to get component", - namespace, - component, - Some(e.to_string()), - ); - continue; - } - }; - - let endpoint: dynamo_runtime::component::Endpoint = - component_obj.endpoint(CLEAR_KV_ENDPOINT); - - let client = match endpoint.client().await { - Ok(c) => c, - Err(e) => { - add_worker_result( - false, - entry_name, - "Failed to get client", - namespace, - component, - Some(e.to_string()), - ); - continue; - } - }; - - let router = match PushRouter::<(), serde_json::Value>::from_client( - client.clone(), - Default::default(), - ) - .await - { - Ok(r) => r, - Err(e) => { - add_worker_result( - false, - entry_name, - "Failed to create router", - namespace, - component, - Some(e.to_string()), - ); - continue; - } - }; - - let discovery_client = distributed.discovery(); - let discovery_key = DiscoveryQuery::Endpoint { - namespace: namespace.clone(), - component: component.clone(), - endpoint: CLEAR_KV_ENDPOINT.to_string(), - }; - - let discovery_instances = match discovery_client.list(discovery_key).await { - Ok(instances) => instances, - Err(e) => { - add_worker_result( - false, - entry_name, - "Failed to get instances for worker group", - namespace, - component, - Some(e.to_string()), - ); - continue; - } - }; - - if discovery_instances.is_empty() { - add_worker_result( - false, - entry_name, - "No instances found for clear_kv_blocks endpoint", - namespace, - component, - None, - ); - continue; - } - - let instances_filtered: Vec = discovery_instances - .into_iter() - .filter_map(|di| match di { - dynamo_runtime::discovery::DiscoveryInstance::Endpoint(instance) => Some(instance), - _ => None, - }) - .collect(); - - for instance in &instances_filtered { - let instance_name = format!("{}-instance-{}", entry.name, instance.id()); - match router.direct(().into(), instance.id()).await { - Ok(mut stream) => match stream.next().await { - Some(response) => { - add_worker_result( - true, - instance_name, - "Successfully cleared kv blocks for instance", - namespace, - component, - Some(response.to_string()), - ); - } - None => { - add_worker_result( - false, - instance_name, - "No response from instance", - namespace, - component, - None, - ); - } - }, - Err(e) => { - add_worker_result( - false, - instance_name, - "Failed to send request for instance", - namespace, - component, - Some(e.to_string()), - ); - } - } - } - } - - Json(serde_json::json!({ - "cleared_workers": cleared_workers, - "failed_workers": failed_workers - })) -} diff --git a/lib/llm/src/http/service/openai.rs b/lib/llm/src/http/service/openai.rs index 4f14cb6845c1..4620941095fd 100644 --- a/lib/llm/src/http/service/openai.rs +++ b/lib/llm/src/http/service/openai.rs @@ -534,6 +534,31 @@ fn copy_x_request_id( } } +/// Warn (once per request) when nvext data is dropped because the extension is +/// disabled. Only called from the disabled branch, so the default path is free. +fn warn_nvext_disabled(endpoint: &str, nvext_present: bool, headers: &HeaderMap) { + use crate::protocols::openai::nvext::{ + HEADER_DP_RANK, HEADER_DP_RANK_ALIAS, HEADER_PREFILL_DP_RANK, HEADER_PREFILL_INSTANCE_ID, + HEADER_WORKER_INSTANCE_ID, + }; + let header_present = [ + HEADER_WORKER_INSTANCE_ID, + HEADER_PREFILL_INSTANCE_ID, + HEADER_DP_RANK, + HEADER_DP_RANK_ALIAS, + HEADER_PREFILL_DP_RANK, + ] + .iter() + .any(|h| headers.contains_key(*h)); + + if nvext_present || header_present { + tracing::warn!( + endpoint, + "request carried nvext data but DYN_ENABLE_FRONTEND_NVEXT is disabled; dropping it" + ); + } +} + /// OpenAI Completions Request Handler /// /// This method will handle the incoming request for the `/v1/completions endpoint`. The endpoint is a "source" @@ -551,7 +576,12 @@ async fn handler_completions( check_ready(&state)?; check_model_serving_ready(&state, &request.inner.model)?; - request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); + request.nvext = if state.nvext_enabled() { + apply_header_routing_overrides(request.nvext.take(), &headers) + } else { + warn_nvext_disabled("completions", request.nvext.is_some(), &headers); + None + }; // create the context for the request let request_id = get_or_create_request_id(&headers); @@ -971,12 +1001,17 @@ async fn completions_batch( async fn embeddings( State(state): State>, headers: HeaderMap, - Json(request): Json, + Json(mut request): Json, ) -> Result { // return a 503 if the service or per-model topology is not ready check_ready(&state)?; check_model_serving_ready(&state, &request.inner.model)?; + if !state.nvext_enabled() { + warn_nvext_disabled("embeddings", request.nvext.is_some(), &headers); + request.nvext = None; + } + let request_id = get_or_create_request_id(&headers); let request = context_from_headers(request, request_id, &headers)?; let request_id = request.id().to_string(); @@ -1142,7 +1177,12 @@ async fn handler_chat_completions( check_model_serving_ready(&state, resolved_model)?; } - request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); + request.nvext = if state.nvext_enabled() { + apply_header_routing_overrides(request.nvext.take(), &headers) + } else { + warn_nvext_disabled("chat_completions", request.nvext.is_some(), &headers); + None + }; // create the context for the request let request_id = get_or_create_request_id(&headers); @@ -1891,7 +1931,12 @@ async fn handler_responses( check_model_serving_ready(&state, resolved_model)?; } - request.nvext = apply_header_routing_overrides(request.nvext.take(), &headers); + request.nvext = if state.nvext_enabled() { + apply_header_routing_overrides(request.nvext.take(), &headers) + } else { + warn_nvext_disabled("responses", request.nvext.is_some(), &headers); + None + }; // create the context for the request let request_id = get_or_create_request_id(&headers); diff --git a/lib/llm/src/http/service/service_v2.rs b/lib/llm/src/http/service/service_v2.rs index f07b309224c9..7a297252c546 100644 --- a/lib/llm/src/http/service/service_v2.rs +++ b/lib/llm/src/http/service/service_v2.rs @@ -26,8 +26,8 @@ use anyhow::Result; use axum_server::tls_rustls::RustlsConfig; use derive_builder::Builder; use dynamo_runtime::DistributedRuntime; -use dynamo_runtime::config::env_is_truthy; use dynamo_runtime::config::environment_names::llm as env_llm; +use dynamo_runtime::config::{env_is_falsey, env_is_truthy}; use dynamo_runtime::discovery::Discovery; use dynamo_runtime::logging::{make_inference_request_span, make_system_request_span}; use dynamo_runtime::metrics::{ @@ -61,6 +61,7 @@ pub struct State { discovery_client: Arc, flags: StateFlags, cancel_token: CancellationToken, + nvext_enabled: bool, } #[derive(Default, Debug)] @@ -131,11 +132,21 @@ impl State { manager: Arc, discovery_client: Arc, cancel_token: CancellationToken, + ) -> Self { + Self::new_with_nvext_enabled(manager, discovery_client, cancel_token, true) + } + + pub fn new_with_nvext_enabled( + manager: Arc, + discovery_client: Arc, + cancel_token: CancellationToken, + nvext_enabled: bool, ) -> Self { Self { manager, metrics: Arc::new(Metrics::default()), discovery_client, + nvext_enabled, flags: StateFlags { chat_endpoints_enabled: AtomicBool::new(false), cmpl_endpoints_enabled: AtomicBool::new(false), @@ -173,6 +184,13 @@ impl State { self.cancel_token.is_cancelled() } + /// Master switch for the `nvext` extension protocol (see + /// [`environment_names::llm::DYN_ENABLE_FRONTEND_NVEXT`]). + #[inline] + pub fn nvext_enabled(&self) -> bool { + self.nvext_enabled + } + /// Get the cancellation token pub fn cancel_token(&self) -> &CancellationToken { &self.cancel_token @@ -279,6 +297,17 @@ pub struct HttpServiceConfig { #[builder(default = "false")] enable_rl: bool, + /// Master switch for the `nvext` extension protocol. Default `true`, + /// env-falsey on `DYN_ENABLE_FRONTEND_NVEXT` overrides to `false`. + #[builder(default = "true")] + enable_nvext: bool, + + /// Master switch for the frontend admin API surface (`GET` / + /// `POST /busy_threshold`). Default `true`, env-falsey on + /// `DYN_ENABLE_FRONTEND_ADMIN_API` overrides to `false`. + #[builder(default = "true")] + enable_admin_api: bool, + /// Port for the RL worker discovery listener. Defaults to `DYN_RL_PORT` or 8001. #[builder(default = "default_rl_port()")] rl_port: u16, @@ -567,7 +596,18 @@ impl HttpServiceConfigBuilder { cancel_token.child_token(), )) as Arc }); - let state = Arc::new(State::new(model_manager, discovery_client, cancel_token)); + // Env-falsey overrides the builder; unset preserves the builder default. + let nvext_enabled = + config.enable_nvext && !env_is_falsey(env_llm::DYN_ENABLE_FRONTEND_NVEXT); + let admin_api_enabled = + config.enable_admin_api && !env_is_falsey(env_llm::DYN_ENABLE_FRONTEND_ADMIN_API); + + let state = Arc::new(State::new_with_nvext_enabled( + model_manager, + discovery_client, + cancel_token, + nvext_enabled, + )); state .flags .set(&EndpointType::Chat, config.enable_chat_endpoints); @@ -641,7 +681,7 @@ impl HttpServiceConfigBuilder { }; // System routes (health, metrics, models) — debug-level spans - let system_routes = vec![ + let mut system_routes = vec![ metrics::router( registry, var(HTTP_SVC_METRICS_PATH_ENV).ok(), @@ -657,8 +697,18 @@ impl HttpServiceConfigBuilder { }, super::health::health_check_router(state.clone(), var(HTTP_SVC_HEALTH_PATH_ENV).ok()), super::health::live_check_router(state.clone(), var(HTTP_SVC_LIVE_PATH_ENV).ok()), - super::busy_threshold::busy_threshold_router(state.clone(), None), ]; + if admin_api_enabled { + system_routes.push(super::busy_threshold::busy_threshold_router( + state.clone(), + None, + )); + } else { + tracing::info!( + env = env_llm::DYN_ENABLE_FRONTEND_ADMIN_API, + "frontend admin API disabled — busy_threshold routes not registered" + ); + } let mut system_router = axum::Router::new(); for (route_docs, route) in system_routes { system_router = system_router.merge(route); @@ -861,4 +911,107 @@ mod tests { // Clean up handle.abort(); } + + /// `enable_admin_api=false` ⇒ `GET /busy_threshold` is not registered and + /// returns 404, not 503 or 405. Inference is unaffected (covered by other + /// tests). + #[tokio::test] + async fn test_admin_api_disabled_404s_busy_threshold() { + let cancel_token = Arc::new(CancellationToken::new()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let service = HttpService::builder() + .port(port) + .enable_admin_api(false) + .build() + .unwrap(); + + let service_token = cancel_token.clone(); + let handle = tokio::spawn(async move { + service + .run_with_listener((*service_token).clone(), listener) + .await + .unwrap(); + }); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + + let resp = reqwest::Client::new() + .get(format!("http://localhost:{}/busy_threshold", port)) + .send() + .await + .expect("request failed"); + assert_eq!(resp.status(), reqwest::StatusCode::NOT_FOUND); + + // And /live still works (sanity: only the admin surface is gated). + let live = reqwest::Client::new() + .get(format!("http://localhost:{}/live", port)) + .send() + .await + .expect("request failed"); + assert_eq!(live.status(), reqwest::StatusCode::OK); + + cancel_token.cancel(); + handle.abort(); + } + + /// `enable_nvext` is wired from the builder onto `State.nvext_enabled` and + /// exposed via the accessor used by the openai handlers. + #[test] + fn test_enable_nvext_propagates_through_builder_to_state() { + let on = HttpService::builder().enable_nvext(true).build().unwrap(); + assert!(on.state.nvext_enabled()); + + let off = HttpService::builder().enable_nvext(false).build().unwrap(); + assert!(!off.state.nvext_enabled()); + + let default = HttpService::builder().build().unwrap(); + assert!( + default.state.nvext_enabled(), + "default should preserve current behavior (nvext on)" + ); + } + + /// `DYN_ENABLE_FRONTEND_NVEXT` is the env-var mirror of the builder + /// flag. Unset → builder default wins (on). Truthy strings → on. + /// Falsey strings (`0` / `false` / `no` / `off`, case-insensitive) → + /// off, regardless of what the builder asked for. + #[test] + fn test_dyn_enable_frontend_nvext_env_var_mirror() { + use dynamo_runtime::config::environment_names::llm::DYN_ENABLE_FRONTEND_NVEXT; + + // Unset → builder default (true) wins. + temp_env::with_var_unset(DYN_ENABLE_FRONTEND_NVEXT, || { + let svc = HttpService::builder().build().unwrap(); + assert!( + svc.state.nvext_enabled(), + "unset env + default builder = on" + ); + }); + + // Explicit truthy → on (builder default also on; env doesn't flip it off). + temp_env::with_var(DYN_ENABLE_FRONTEND_NVEXT, Some("true"), || { + let svc = HttpService::builder().build().unwrap(); + assert!(svc.state.nvext_enabled(), "env=true + default builder = on"); + }); + + // Explicit falsey → off, even though the builder default is on. + for falsey in ["false", "0", "no", "off", "FALSE"] { + temp_env::with_var(DYN_ENABLE_FRONTEND_NVEXT, Some(falsey), || { + let svc = HttpService::builder().build().unwrap(); + assert!( + !svc.state.nvext_enabled(), + "env={falsey:?} should override builder default to off" + ); + }); + } + + // Builder=false short-circuits regardless of env. + temp_env::with_var(DYN_ENABLE_FRONTEND_NVEXT, Some("true"), || { + let svc = HttpService::builder().enable_nvext(false).build().unwrap(); + assert!( + !svc.state.nvext_enabled(), + "builder=false wins even if env=true" + ); + }); + } } diff --git a/lib/llm/tests/http-service.rs b/lib/llm/tests/http-service.rs index 398c3e71e934..792bfab7b7bd 100644 --- a/lib/llm/tests/http-service.rs +++ b/lib/llm/tests/http-service.rs @@ -908,3 +908,57 @@ async fn test_request_id_annotation() { cancel_token.cancel(); task.await.unwrap().unwrap(); } + +/// With nvext disabled, a request asking for response `extra_fields` must not +/// produce any `nvext` field in the response. +#[tokio::test] +async fn test_nvext_disabled_strips_request_and_response() { + dynamo_runtime::logging::init(); + + let (listener, port) = bind_random_port().await; + let service = HttpService::builder() + .enable_chat_endpoints(true) + .enable_nvext(false) + .port(port) + .build() + .unwrap(); + let state = service.state_clone(); + let manager = state.manager(); + + let token = CancellationToken::new(); + let cancel_token = token.clone(); + let task = tokio::spawn(async move { service.run_with_listener(token, listener).await }); + wait_for_service_ready(port).await; + + let card = ModelDeploymentCard::with_name_only("test-model"); + manager + .add_chat_completions_model("test-model", card.mdcsum(), Arc::new(CounterEngine {})) + .unwrap(); + + let response = reqwest::Client::new() + .post(format!("http://localhost:{port}/v1/chat/completions")) + .header("x-worker-instance-id", "42") + .json(&serde_json::json!({ + "model": "test-model", + "messages": [{"role": "user", "content": "hi"}], + "stream": true, + "max_tokens": 1, + "nvext": { + "extra_fields": ["worker_id", "timing", "engine_data"], + "backend_instance_id": 99 + } + })) + .send() + .await + .expect("request should succeed"); + assert!(response.status().is_success()); + + let body = response.text().await.expect("read body"); + assert!( + !body.contains("\"nvext\""), + "nvext gate off: response must not contain an `nvext` field, got: {body}" + ); + + cancel_token.cancel(); + task.await.unwrap().unwrap(); +} diff --git a/lib/runtime/src/config/environment_names.rs b/lib/runtime/src/config/environment_names.rs index de60b53e46c3..5463921436f1 100644 --- a/lib/runtime/src/config/environment_names.rs +++ b/lib/runtime/src/config/environment_names.rs @@ -284,6 +284,21 @@ pub mod llm { /// Enable the experimental Anthropic Messages API endpoint (/v1/messages) pub const DYN_ENABLE_ANTHROPIC_API: &str = "DYN_ENABLE_ANTHROPIC_API"; + /// Master switch for the `nvext` extension protocol on the frontend. + /// Default `true`. Falsy values (`0` / `false` / `no` / `off`, + /// case-insensitive) cause the frontend to drop `request.nvext` at + /// handler entry, ignore the routing-override headers + /// (`x-worker-instance-id`, `x-prefill-instance-id`, `x-dp-rank`, + /// `x-prefill-dp-rank`), and silently ignore the response-side + /// `extra_fields` opt-in. + pub const DYN_ENABLE_FRONTEND_NVEXT: &str = "DYN_ENABLE_FRONTEND_NVEXT"; + + /// Master switch for the frontend's HTTP admin API surface. + /// Default `true`. Falsy values prevent registration of `GET` / + /// `POST /busy_threshold`. Inference, metrics, models, health, and + /// liveness routes are unaffected. + pub const DYN_ENABLE_FRONTEND_ADMIN_API: &str = "DYN_ENABLE_FRONTEND_ADMIN_API"; + /// Strip the Claude Code billing preamble (`x-anthropic-billing-header: ...`) /// from the system prompt before forwarding to the target model. The preamble /// varies per session and per release, wasting tokens and breaking prompt caching.