From 77be8e65a9685b01bf0d6a2e5ca7a89eb6383459 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Fri, 16 Oct 2026 03:02:54 -0400 Subject: [PATCH 1/2] api: split admin node_maintenance.rs by route family (pure move) node_maintenance.rs serves three admin route families beside the Direct node registry: node status, health and reconcile; node template pull and delete through the node-api, and listing from the cached template inventory; and the node-api client plumbing the registry and template routes use. Keep the registry handlers, the shared node row and response types, and the node reads in node_maintenance.rs, and move the rest into child modules: node_api, node_status and node_templates. Every line that touches a Direct-only table stays in node_maintenance.rs, so archcheck's direct_tables ratchet is unchanged. Generated by scripts/split_rust_module.py (#562) from the spec in the pull request. Items move verbatim. The only edits keep each item's reach: 137 private items, fields or methods become pub(super). The parent re-exports the moved items, so no caller changes. --- .../api/src/routes/admin/node_maintenance.rs | 1228 +---------------- .../routes/admin/node_maintenance/node_api.rs | 195 +++ .../admin/node_maintenance/node_status.rs | 504 +++++++ .../admin/node_maintenance/node_templates.rs | 550 ++++++++ 4 files changed, 1256 insertions(+), 1221 deletions(-) create mode 100644 crates/libs/api/src/routes/admin/node_maintenance/node_api.rs create mode 100644 crates/libs/api/src/routes/admin/node_maintenance/node_status.rs create mode 100644 crates/libs/api/src/routes/admin/node_maintenance/node_templates.rs diff --git a/crates/libs/api/src/routes/admin/node_maintenance.rs b/crates/libs/api/src/routes/admin/node_maintenance.rs index cdaf5df4b..fca500a31 100644 --- a/crates/libs/api/src/routes/admin/node_maintenance.rs +++ b/crates/libs/api/src/routes/admin/node_maintenance.rs @@ -37,11 +37,13 @@ use crate::{ routes::{AppState, admin::support::require_managing_org_admin}, }; -const NODE_API_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); -const NODE_API_FAST_RPC_TIMEOUT: Duration = Duration::from_secs(10); -const NODE_API_TEMPLATE_RPC_TIMEOUT: Duration = Duration::from_secs(600); -const NODE_TEMPLATE_INVENTORY_CONCURRENCY: usize = 16; -const DEFAULT_NODE_STATUS_CACHE_TTL_SECS: u64 = 30; +mod node_api; +mod node_status; +mod node_templates; + +use node_api::*; +pub use node_status::*; +pub use node_templates::*; #[derive(Debug, Clone, FromRow)] struct NodeRow { @@ -60,17 +62,6 @@ struct NodeRow { node_name: Option, } -#[derive(Debug, Clone, FromRow)] -struct NodeStatusRow { - node_id: NodeId, - stale_after: OffsetDateTime, - reachable: bool, - error: Option, - health_healthy: Option, - services: sqlx::types::Json>, - node_ctl: Option>, -} - #[derive(Debug, Clone, Serialize)] struct NodeResponse { id: NodeId, @@ -87,86 +78,6 @@ struct NodeResponse { deleted_at: Option, } -#[derive(Debug, Clone, Serialize, Deserialize)] -struct ServiceHealthResponse { - name: String, - reachable: bool, - error: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -struct ReconciliationStatusResponse { - name: String, - poll_interval_secs: u64, - last_started_at_unix_secs: u64, - last_completed_at_unix_secs: u64, - last_success_at_unix_secs: u64, - next_due_at_unix_secs: u64, - running: bool, - templates_seen: u64, - sandboxes_seen: u64, - templates_applied: u64, - sandboxes_applied: u64, - last_error: String, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -struct NodeCtlStatusResponse { - reachable: bool, - error: Option, - reconciliations: Vec, -} - -#[derive(Debug, Clone, Serialize)] -struct NodeHealthStatusResponse { - reachable: bool, - healthy: bool, - error: Option, - services: Vec, - node_ctl: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize)] -struct NodeTemplateSnapshotResponse { - vcpu: u32, - memory_mib: u32, - snapshot_hash: String, -} - -#[derive(Debug, FromRow)] -struct NodeTemplateInventoryRow { - name: String, - content_hash: Option, - build_date: Option, - artifact_size_bytes: Option, - cache_size_bytes: Option, - disk_size_mb: Option, - runtime: Option, - arch: Option, - snapshots: sqlx::types::Json>, -} - -#[derive(Debug, Serialize)] -struct NodeTemplateBuildResponse { - name: String, - content_hash: Option, - build_date: Option, - artifact_size_bytes: Option, - cache_size_bytes: Option, - disk_size_mb: Option, - runtime: Option, - arch: Option, - snapshots: Vec, -} - -#[derive(Debug, Serialize)] -struct NodeTemplateInventoryResponse { - node: NodeResponse, - reachable: bool, - error: Option, - builds: Vec, -} - #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] pub struct CreateNodeRequest { @@ -195,129 +106,10 @@ pub struct UpdateNodeRequest { verify: bool, } -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct RefreshNodeStatusRequest { - role: Option, - #[serde(default)] - force: bool, -} - -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ReconcileFacetRequest { - #[serde(deserialize_with = "deserialize_reconcile_facet")] - facet: ReconcileFacet, - element: Option, -} - -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct ReconcileNodesRequest { - role: Option, - #[serde(default)] - node_ids: Vec, - #[serde(default, deserialize_with = "deserialize_reconcile_facets")] - facets: Vec, - #[serde(default)] - refs: Vec, -} - -#[derive(Debug, Deserialize)] -pub struct NodeFilters { - role: Option, -} - -#[derive(Debug, Deserialize)] -#[serde(deny_unknown_fields)] -pub struct PullNodeTemplateRequest { - template_id: String, - content_hash: Option, - source_node_id: NodeId, - #[serde(default)] - include_snapshot: bool, - snapshot_vcpu: Option, - snapshot_memory_mib: Option, - #[serde(default = "default_true")] - skip_if_exists: bool, - local_tag: Option, - block_on_snapshot_policy_achieved: Option, -} - -#[derive(Debug, Deserialize)] -pub struct DeleteNodeTemplateQuery { - content_hash: Option, -} - -#[derive(Debug, Serialize)] -struct PullNodeTemplateResponse { - content_hash: String, - already_existed: bool, - bytes_transferred: u64, -} - -#[derive(Debug, Serialize)] -struct RefreshNodeStatusResponse { - refreshed: bool, - nodes_seen: usize, - nodes_updated: usize, - nodes_failed: usize, - cache_ttl_secs: u64, - error: Option, -} - -#[derive(Debug, Serialize)] -struct ReconcileNodeResponse { - node_id: NodeId, - node_name: Option, - role: String, - queued: bool, - error: Option, -} - -#[derive(Debug, Serialize)] -struct ReconcileNodesResponse { - nodes_seen: usize, - nodes_queued: usize, - nodes_failed: usize, - nodes: Vec, -} - -#[derive(Debug, Serialize)] -struct ReconcileNodeNetworkPolicyResponse { - queued: bool, -} - -fn deserialize_reconcile_facet<'de, D>(deserializer: D) -> Result -where - D: Deserializer<'de>, -{ - let value = String::deserialize(deserializer)?; - ReconcileFacet::parse(value.trim()) - .ok_or_else(|| de::Error::custom(format!("unsupported reconcile facet: {value}"))) -} - -fn deserialize_reconcile_facets<'de, D>(deserializer: D) -> Result, D::Error> -where - D: Deserializer<'de>, -{ - Vec::::deserialize(deserializer)? - .into_iter() - .map(|value| { - ReconcileFacet::parse(value.trim()) - .ok_or_else(|| de::Error::custom(format!("unsupported reconcile facet: {value}"))) - }) - .collect() -} - fn empty_metadata() -> serde_json::Value { serde_json::json!({}) } -fn default_true() -> bool { - true -} - fn format_time(value: OffsetDateTime) -> String { value .format(&time::format_description::well_known::Rfc3339) @@ -397,37 +189,6 @@ fn node_response(row: NodeRow) -> NodeResponse { } } -fn node_supports_state_reconciliation(role: NodeRole) -> bool { - matches!(role, NodeRole::Worker | NodeRole::Builder) -} - -fn template_capable_node_role( - node: &NodeRow, - label: &str, -) -> Result)> { - match node.role { - role @ (NodeRole::Worker | NodeRole::Builder) => Ok(role), - NodeRole::Cluster => Err(api_error( - StatusCode::BAD_REQUEST, - format!("{label} node must be a worker or builder"), - )), - } -} - -fn ensure_template_capable_node( - node: &NodeRow, - label: &str, -) -> Result<(), (StatusCode, Json)> { - if node.deleted_at.is_some() { - return Err(api_error( - StatusCode::NOT_FOUND, - format!("{label} node not found"), - )); - } - template_capable_node_role(node, label)?; - Ok(()) -} - fn require_admin_node_registry( state: &AppState, ) -> Result<(), (StatusCode, Json)> { @@ -444,188 +205,6 @@ fn require_admin_node_registry( Ok(()) } -fn normalize_node_api_endpoint(raw: &str) -> Result)> { - let mut endpoint = raw.trim().trim_end_matches('/').to_string(); - if let Some(stripped) = endpoint.strip_prefix("http://") { - endpoint = stripped.to_string(); - } - if endpoint.starts_with("https://") { - return Err(api_error( - StatusCode::BAD_REQUEST, - "node_api_endpoint must use plaintext node-api host:port", - )); - } - if endpoint.is_empty() || endpoint.contains('/') { - return Err(api_error( - StatusCode::BAD_REQUEST, - "node_api_endpoint must be a host:port endpoint", - )); - } - let uri: axum::http::Uri = endpoint_uri(&endpoint).parse().map_err(|err| { - api_error( - StatusCode::BAD_REQUEST, - format!("invalid node_api_endpoint: {err}"), - ) - })?; - if uri - .authority() - .and_then(axum::http::uri::Authority::port_u16) - .is_none() - { - return Err(api_error( - StatusCode::BAD_REQUEST, - "node_api_endpoint must include an explicit port", - )); - } - Ok(endpoint) -} - -fn endpoint_uri(endpoint: &str) -> String { - pplx_space_common_grpc::internal_transport::qualify_endpoint(endpoint) -} - -fn node_endpoint( - endpoint: &str, - request_timeout: Option, -) -> Result> { - // `node_api_endpoint` is scheme-aware: it supplies the internal RPC scheme for bare - // `host:port` and passes `http://`/`https://`-qualified inputs through - // untouched. Trim whitespace only — stripping `http://` here would mangle - // an `https://` input into `http://https://...`. - let endpoint = endpoint.trim(); - Ok(crate::controllers::substrate::node_api_endpoint( - endpoint, - NODE_API_CONNECT_TIMEOUT, - request_timeout.unwrap_or(crate::controllers::substrate::NODE_RPC_TIMEOUT), - )?) -} - -fn proto_role_name(role: i32) -> Option<&'static str> { - if role == ProtoNodeRole::Worker as i32 { - Some(NodeRole::Worker.as_str()) - } else if role == ProtoNodeRole::Builder as i32 { - Some(NodeRole::Builder.as_str()) - } else if role == ProtoNodeRole::Cluster as i32 { - Some(NodeRole::Cluster.as_str()) - } else { - None - } -} - -fn bearer_kind_for_role(role: NodeRole) -> SpaceBearerToken { - match role { - NodeRole::Cluster => SpaceBearerToken::ClusterApi, - NodeRole::Worker | NodeRole::Builder => SpaceBearerToken::NodeApi, - } -} - -async fn verify_node_endpoint( - endpoint: &str, - expected_role: NodeRole, -) -> Result)> { - let endpoint = node_endpoint(endpoint, Some(NODE_API_FAST_RPC_TIMEOUT)).map_err(|err| { - api_error( - StatusCode::BAD_REQUEST, - format!("invalid node_api_endpoint: {err}"), - ) - })?; - let service = match expected_role { - NodeRole::Cluster => pplx_space_common_grpc::internal_transport::Service::ClusterApi, - NodeRole::Worker | NodeRole::Builder => { - pplx_space_common_grpc::internal_transport::Service::NodeApi - } - }; - let endpoint = - pplx_space_common_grpc::internal_transport::client_connect_endpoint(service, endpoint) - .await - .map_err(|error| api_error(StatusCode::BAD_GATEWAY, error.to_string()))?; - let channel = endpoint.connect().await.map_err(|err| { - api_error( - StatusCode::BAD_GATEWAY, - format!("node-api endpoint is not reachable: {err}"), - ) - })?; - let mut client = NodeInfoClient::new(crate::controllers::substrate::node_api_channel( - channel, - bearer_kind_for_role(expected_role), - )); - let response = client - .get_node_info(GetNodeInfoRequest {}) - .await - .map_err(|err| { - api_error( - StatusCode::BAD_GATEWAY, - format!("node-api NodeInfo probe failed: {err}"), - ) - })? - .into_inner(); - - let actual_role = proto_role_name(response.role).ok_or_else(|| { - api_error( - StatusCode::BAD_GATEWAY, - format!("node-api returned unknown role {}", response.role), - ) - })?; - if actual_role != expected_role.as_str() { - return Err(api_error( - StatusCode::BAD_REQUEST, - format!("node-api role mismatch: expected {expected_role}, got {actual_role}"), - )); - } - - response.node_id.parse::().map_err(|err| { - api_error( - StatusCode::BAD_GATEWAY, - format!("node-api returned invalid node_id: {err}"), - ) - }) -} - -fn node_api_client( - node: &NodeRow, - label: &str, - request_timeout: Duration, -) -> Result)> { - let raw_endpoint = node.node_api_endpoint.as_deref().ok_or_else(|| { - api_error( - StatusCode::INTERNAL_SERVER_ERROR, - format!("{label} node has no node_api_endpoint"), - ) - })?; - let endpoint = node_endpoint(raw_endpoint, Some(request_timeout)).map_err(|err| { - api_error( - StatusCode::BAD_REQUEST, - format!("invalid {label} node_api_endpoint: {err}"), - ) - })?; - Ok(NodeApiClient::new(endpoint)) -} - -fn node_rpc_error(operation: &str, err: tonic::Status) -> (StatusCode, Json) { - let status = match err.code() { - tonic::Code::InvalidArgument => StatusCode::BAD_REQUEST, - tonic::Code::NotFound => StatusCode::NOT_FOUND, - tonic::Code::FailedPrecondition => StatusCode::CONFLICT, - _ => StatusCode::BAD_GATEWAY, - }; - api_error(status, format!("{operation} failed: {err}")) -} - -fn node_api_rpc_error_response( - node: &NodeRow, - command: &'static str, - status: tonic::Status, -) -> (StatusCode, Json) { - tracing::warn!( - node_id = %node.id, - endpoint = %node.node_api_endpoint.as_deref().unwrap_or("-"), - command, - grpc_code = ?status.code(), - "node-api RPC failed" - ); - node_rpc_error(command, status) -} - fn inventory_entry_to_row( entry: crate::controllers::substrate::NodeInventoryEntry, ) -> Option { @@ -730,114 +309,6 @@ async fn fetch_inventory_nodes( .map_err(db_error) } -async fn fetch_inventory_node( - state: &AppState, - id: NodeId, -) -> Result)> { - let nodes = fetch_inventory_nodes(state, None).await?; - nodes - .into_iter() - .find(|node| node.id == id && node.deleted_at.is_none()) - .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "Node not found")) -} - -async fn fetch_node_status_map( - pool: &sqlx::PgPool, - node_ids: &[NodeId], -) -> Result, sqlx::Error> { - if node_ids.is_empty() { - return Ok(Vec::new()); - } - sqlx::query_as::<_, NodeStatusRow>(sql!( - "SELECT node_id, - stale_after, - reachable, - error, - health_healthy, - services, - node_ctl - FROM node_status_snapshots - WHERE node_id = ANY($1)" - )) - .bind(node_ids) - .fetch_all(pool) - .await -} - -async fn node_status_refresh_needed( - pool: &sqlx::PgPool, - nodes: &[NodeRow], -) -> Result { - let node_ids = nodes - .iter() - .filter(|node| node_supports_state_reconciliation(node.role)) - .map(|node| node.id) - .collect::>(); - if node_ids.is_empty() { - return Ok(false); - } - let statuses = fetch_node_status_map(pool, &node_ids).await?; - let now = OffsetDateTime::now_utc(); - Ok(node_ids.iter().any(|node_id| { - statuses - .iter() - .find(|status| status.node_id == *node_id) - .map(|status| status.stale_after <= now) - .unwrap_or(true) - })) -} - -async fn fetch_cached_node_health(pool: &sqlx::PgPool, node: &NodeRow) -> NodeHealthStatusResponse { - if !node_supports_state_reconciliation(node.role) { - return NodeHealthStatusResponse { - reachable: false, - healthy: false, - error: Some("node health is only available for worker and builder nodes".to_string()), - services: Vec::new(), - node_ctl: None, - }; - } - - match fetch_node_status_map(pool, &[node.id]).await { - Ok(statuses) => { - let Some(status) = statuses - .into_iter() - .find(|status| status.node_id == node.id) - else { - return NodeHealthStatusResponse { - reachable: false, - healthy: false, - error: Some("cached node health is unavailable".to_string()), - services: Vec::new(), - node_ctl: None, - }; - }; - NodeHealthStatusResponse { - reachable: status.reachable, - healthy: status.health_healthy.unwrap_or(false), - error: status.error, - services: status.services.0, - node_ctl: status.node_ctl.map(|node_ctl| node_ctl.0), - } - } - Err(err) => NodeHealthStatusResponse { - reachable: false, - healthy: false, - error: Some(format!("cached node health query failed: {err}")), - services: Vec::new(), - node_ctl: None, - }, - } -} - -fn default_node_publish_refs() -> Vec { - vec![ - ReconcileRef::full(ReconcileFacet::NodeStatus), - ReconcileRef::full(ReconcileFacet::SandboxInventory), - ReconcileRef::full(ReconcileFacet::TemplateInventory), - ] -} - async fn enqueue_default_node_publish( pool: &sqlx::PgPool, node: &NodeRow, @@ -858,142 +329,6 @@ async fn enqueue_default_node_publish( } } -fn reconcile_refs_from_request( - req: &ReconcileNodesRequest, -) -> Result, (StatusCode, Json)> { - if !req.refs.is_empty() { - return req - .refs - .iter() - .map(|reference| { - ReconcileRef::from_storage(reference.facet, reference.element.as_deref()).map_err( - |error| { - api_error( - StatusCode::BAD_REQUEST, - format!("invalid reconcile ref element: {error}"), - ) - }, - ) - }) - .collect(); - } - - let facets = if req.facets.is_empty() { - vec![ - ReconcileFacet::SandboxInventory, - ReconcileFacet::TemplateInventory, - ReconcileFacet::NodeStatus, - ] - } else { - req.facets.clone() - }; - facets - .into_iter() - .map(|facet| Ok(ReconcileRef::full(facet))) - .collect() -} - -fn non_empty_optional(value: Option) -> Option { - value.and_then(|value| { - let value = value.trim().to_string(); - if value.is_empty() { None } else { Some(value) } - }) -} - -async fn delete_cached_node_template_inventory( - pool: &sqlx::PgPool, - node_id: NodeId, - template: &str, - content_hash: Option<&str>, -) -> Result<(), sqlx::Error> { - if let Some(content_hash) = content_hash { - sqlx::query(sql!( - "DELETE FROM node_template_inventory - WHERE node_id = $1 - AND template_name = $2 - AND content_hash = $3" - )) - .bind(node_id) - .bind(template) - .bind(content_hash) - .execute(pool) - .await?; - } else { - sqlx::query(sql!( - "DELETE FROM node_template_inventory - WHERE node_id = $1 - AND template_name = $2" - )) - .bind(node_id) - .bind(template) - .execute(pool) - .await?; - } - - let remaining: i64 = sqlx::query_scalar(sql!( - "SELECT COUNT(*)::BIGINT - FROM node_template_inventory - WHERE node_id = $1 - AND template_name = $2" - )) - .bind(node_id) - .bind(template) - .fetch_one(pool) - .await?; - - if remaining == 0 { - sqlx::query(sql!( - "DELETE FROM node_facet_observed_status - WHERE node_id = $1 - AND facet = $2 - AND element = $3" - )) - .bind(node_id) - .bind(FACET_NETWORK_POLICY) - .bind(format!("template:{template}")) - .execute(pool) - .await?; - } - - Ok(()) -} - -async fn note_node_template_deleted( - pool: &sqlx::PgPool, - node: &NodeRow, - template: &str, - content_hash: Option<&str>, -) { - if let Err(error) = - delete_cached_node_template_inventory(pool, node.id, template, content_hash).await - { - tracing::warn!( - node_id = %node.id, - template, - content_hash = content_hash.unwrap_or(""), - %error, - "failed to delete cached node template inventory after node delete" - ); - } - - if let Err(error) = enqueue_node_reconcile_refs_and_trigger( - pool, - node.id, - &[ReconcileRef::full(ReconcileFacet::TemplateInventory)], - ReconcileReason::ManualNodeReconcile, - ) - .await - { - tracing::warn!( - node_id = %node.id, - template, - content_hash = content_hash.unwrap_or(""), - %error, - "failed to enqueue template inventory refresh after node template delete" - ); - } -} - pub async fn create_node( State(state): State, Extension(AuthUser(actor)): Extension, @@ -1367,554 +702,5 @@ async fn hard_delete_direct_node(pool: &sqlx::PgPool, id: NodeId) -> Result, - Extension(AuthUser(actor)): Extension, - Json(req): Json, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - let cache_ttl_secs = DEFAULT_NODE_STATUS_CACHE_TTL_SECS; - let nodes = match fetch_inventory_nodes(&state, req.role).await { - Ok(nodes) => nodes - .into_iter() - .filter(|node| node_supports_state_reconciliation(node.role)) - .collect::>(), - Err(err) => return err.into_response(), - }; - - if !req.force { - match node_status_refresh_needed(&state.pool, &nodes).await { - Ok(true) => {} - Ok(false) => { - return Json(RefreshNodeStatusResponse { - refreshed: false, - nodes_seen: nodes.len(), - nodes_updated: 0, - nodes_failed: 0, - cache_ttl_secs, - error: None, - }) - .into_response(); - } - Err(err) => return db_error(err).into_response(), - } - } - - let node_ids = nodes.iter().map(|node| node.id).collect::>(); - let enqueued = match enqueue_node_reconcile_refs_for_nodes( - &state.pool, - node_ids, - &default_node_publish_refs(), - ) - .await - { - Ok(enqueued) => enqueued, - Err(error) => { - tracing::warn!(%error, "failed to enqueue node status refresh reconciliation"); - Vec::new() - } - }; - let trigger_nodes = enqueued - .iter() - .map(|enqueued| enqueued.node_id) - .collect::>(); - if let Err(error) = enqueue_triggers( - &state.pool, - trigger_nodes, - ReconcileReason::ManualNodeStatusRefresh, - ) - .await - { - tracing::warn!(%error, "failed to enqueue manual node status refresh triggers"); - } - let nodes_seen = nodes.len(); - let nodes_updated = enqueued.len(); - Json(RefreshNodeStatusResponse { - refreshed: true, - nodes_seen, - nodes_updated, - nodes_failed: nodes_seen.saturating_sub(nodes_updated), - cache_ttl_secs, - error: None, - }) - .into_response() -} - -pub async fn reconcile_nodes( - State(state): State, - Extension(AuthUser(actor)): Extension, - Json(req): Json, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - let refs = match reconcile_refs_from_request(&req) { - Ok(refs) => refs, - Err(err) => return err.into_response(), - }; - - let requested_node_ids = req.node_ids.iter().copied().collect::>(); - let mut nodes = match fetch_inventory_nodes(&state, req.role).await { - Ok(nodes) => nodes - .into_iter() - .filter(|node| node_supports_state_reconciliation(node.role)) - .filter(|node| requested_node_ids.is_empty() || requested_node_ids.contains(&node.id)) - .collect::>(), - Err(err) => return err.into_response(), - }; - nodes.sort_by(|a, b| a.role.cmp(&b.role).then_with(|| a.name.cmp(&b.name))); - - let node_ids = nodes.iter().map(|node| node.id).collect::>(); - let queued_node_ids = - match enqueue_node_reconcile_refs_for_nodes(&state.pool, node_ids, &refs).await { - Ok(enqueued) => enqueued - .into_iter() - .map(|enqueued| enqueued.node_id) - .collect::>(), - Err(error) => { - tracing::warn!(%error, "failed to enqueue manual node reconciliation"); - HashSet::new() - } - }; - let enqueue_error = if queued_node_ids.is_empty() && !nodes.is_empty() { - Some("enqueue reconcile refs failed".to_string()) - } else { - None - }; - let nodes = nodes - .into_iter() - .map(|node| { - let queued = queued_node_ids.contains(&node.id); - ReconcileNodeResponse { - node_id: node.id, - node_name: node.name, - role: node.role.to_string(), - queued, - error: if queued { None } else { enqueue_error.clone() }, - } - }) - .collect::>(); - let nodes_queued = nodes.iter().filter(|node| node.queued).count(); - let trigger_nodes = nodes - .iter() - .filter(|node| node.queued) - .map(|node| node.node_id) - .collect::>(); - if let Err(error) = enqueue_triggers( - &state.pool, - trigger_nodes, - ReconcileReason::ManualNodeReconcile, - ) - .await - { - tracing::warn!(%error, "failed to enqueue manual node reconcile triggers"); - } - let nodes_seen = nodes.len(); - Json(ReconcileNodesResponse { - nodes_seen, - nodes_queued, - nodes_failed: nodes_seen.saturating_sub(nodes_queued), - nodes, - }) - .into_response() -} - -pub async fn node_health( - State(state): State, - Extension(AuthUser(actor)): Extension, - Path(id): Path, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - - let node = match fetch_inventory_node(&state, id).await { - Ok(node) => node, - Err(err) => return err.into_response(), - }; - if !node_supports_state_reconciliation(node.role) { - return api_error( - StatusCode::BAD_REQUEST, - "node health is only available for worker and builder nodes", - ) - .into_response(); - } - - Json(fetch_cached_node_health(&state.pool, &node).await).into_response() -} - -pub async fn reconcile_node_network_policy( - State(state): State, - Extension(AuthUser(actor)): Extension, - Path(id): Path, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - - let node = match fetch_inventory_node(&state, id).await { - Ok(node) => node, - Err(err) => return err.into_response(), - }; - if !node_supports_state_reconciliation(node.role) { - return api_error( - StatusCode::BAD_REQUEST, - "network policy reconciliation is only available for worker and builder nodes", - ) - .into_response(); - } - - let refs = [ReconcileRef::full(ReconcileFacet::NetworkPolicy)]; - match enqueue_node_reconcile_refs_and_trigger( - &state.pool, - node.id, - &refs, - ReconcileReason::ManualNodeNetworkPolicyReconcile, - ) - .await - { - Ok(_) => Json(ReconcileNodeNetworkPolicyResponse { queued: true }).into_response(), - Err(err) => db_error(err).into_response(), - } -} - -pub async fn pull_node_template( - State(state): State, - Extension(AuthUser(actor)): Extension, - Path(target_id): Path, - Json(req): Json, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - if let Err(err) = require_admin_node_registry(&state) { - return err.into_response(); - } - - let template_id = req.template_id.trim().to_string(); - if template_id.is_empty() { - return api_error(StatusCode::BAD_REQUEST, "template_id is required").into_response(); - } - let template_name = match TemplateName::parse(&template_id) { - Ok(template_name) => template_name, - Err(err) => { - return api_error( - StatusCode::BAD_REQUEST, - format!("invalid template_id: {err}"), - ) - .into_response(); - } - }; - - if req.include_snapshot && (req.snapshot_vcpu.is_none() || req.snapshot_memory_mib.is_none()) { - return api_error( - StatusCode::BAD_REQUEST, - "snapshot_vcpu and snapshot_memory_mib are required when include_snapshot is true", - ) - .into_response(); - } - - let target = match fetch_node(&state.pool, target_id).await { - Ok(Some(node)) => node, - Ok(None) => { - return api_error(StatusCode::NOT_FOUND, "target node not found").into_response(); - } - Err(err) => return db_error(err).into_response(), - }; - let target_role = match template_capable_node_role(&target, "target") { - Ok(role) => role, - Err(err) => return err.into_response(), - }; - - let source = match fetch_node(&state.pool, req.source_node_id).await { - Ok(Some(node)) => node, - Ok(None) => { - return api_error(StatusCode::NOT_FOUND, "source node not found").into_response(); - } - Err(err) => return db_error(err).into_response(), - }; - if let Err(err) = ensure_template_capable_node(&source, "source") { - return err.into_response(); - } - - let source_address = match source.node_api_endpoint.clone() { - Some(value) => value, - None => { - return api_error( - StatusCode::INTERNAL_SERVER_ERROR, - "source node has no node_api_endpoint", - ) - .into_response(); - } - }; - let request = PullTemplateRequest { - template_id: template_id.clone(), - content_hash: non_empty_optional(req.content_hash), - tag: None, - sources: vec![TransferSource { - source: Some(Source::Node(NodeSource { - address: source_address, - })), - }], - include_snapshot: req.include_snapshot, - snapshot_vcpu: req.snapshot_vcpu, - snapshot_memory_mib: req.snapshot_memory_mib, - skip_if_exists: req.skip_if_exists, - local_tag: non_empty_optional(req.local_tag), - block_on_snapshot_policy_achieved: req.block_on_snapshot_policy_achieved, - }; - - let mut node_api = match node_api_client(&target, "target", NODE_API_TEMPLATE_RPC_TIMEOUT) { - Ok(client) => client, - Err(err) => return err.into_response(), - }; - let response = match target_role { - NodeRole::Worker => { - let mut client = WorkerApiClient::new(&mut node_api); - client - .pull_template(request, NODE_API_TEMPLATE_RPC_TIMEOUT) - .await - } - NodeRole::Builder => { - let mut client = BuilderApiClient::new(&mut node_api); - client - .pull_template(request, NODE_API_TEMPLATE_RPC_TIMEOUT) - .await - } - NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), - } - .map_err(|status| node_api_rpc_error_response(&target, "PullTemplate", status)); - - match response { - Ok(response) => { - if let Err(err) = enqueue_node_reconcile_refs_and_trigger( - &state.pool, - target.id, - &[ReconcileRef::template_inventory(template_name)], - ReconcileReason::NodeTemplatePulled, - ) - .await - { - tracing::warn!(error = %err, "failed to enqueue pulled template inventory reconcile"); - } - let response = response.into_inner(); - Json(PullNodeTemplateResponse { - content_hash: response.content_hash, - already_existed: response.already_existed, - bytes_transferred: response.bytes_transferred, - }) - .into_response() - } - Err(err) => err.into_response(), - } -} - -pub async fn delete_node_template( - State(state): State, - Extension(AuthUser(actor)): Extension, - Path((node_id, name)): Path<(NodeId, String)>, - Query(query): Query, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - if let Err(err) = require_admin_node_registry(&state) { - return err.into_response(); - } - - let name = name.trim().to_string(); - if name.is_empty() { - return api_error(StatusCode::BAD_REQUEST, "template name is required").into_response(); - } - - let node = match fetch_node(&state.pool, node_id).await { - Ok(Some(node)) => node, - Ok(None) => return api_error(StatusCode::NOT_FOUND, "node not found").into_response(), - Err(err) => return db_error(err).into_response(), - }; - let target_role = match template_capable_node_role(&node, "target") { - Ok(role) => role, - Err(err) => return err.into_response(), - }; - let mut node_api = match node_api_client(&node, "target", NODE_API_FAST_RPC_TIMEOUT) { - Ok(client) => client, - Err(err) => return err.into_response(), - }; - - if let Some(content_hash) = non_empty_optional(query.content_hash) { - let template_name = name.clone(); - let content_hash_for_cache = content_hash.clone(); - let request = DeleteTemplateBuildRequest { - template: Some(TemplateRef { - name, - content_hash: Some(content_hash), - tag: None, - }), - }; - let response = match target_role { - NodeRole::Worker => { - let mut client = WorkerApiClient::new(&mut node_api); - client - .delete_template_build(request, NODE_API_FAST_RPC_TIMEOUT) - .await - } - NodeRole::Builder => { - let mut client = BuilderApiClient::new(&mut node_api); - client - .delete_template_build(request, NODE_API_FAST_RPC_TIMEOUT) - .await - } - NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), - } - .map_err(|status| node_api_rpc_error_response(&node, "DeleteTemplateBuild", status)); - return match response { - Ok(_) => { - note_node_template_deleted( - &state.pool, - &node, - &template_name, - Some(&content_hash_for_cache), - ) - .await; - StatusCode::NO_CONTENT.into_response() - } - Err(err) => err.into_response(), - }; - } - - let template_name = name.clone(); - let request = DeleteTemplateRequest { template: name }; - let response = match target_role { - NodeRole::Worker => { - let mut client = WorkerApiClient::new(&mut node_api); - client - .delete_template(request, NODE_API_FAST_RPC_TIMEOUT) - .await - } - NodeRole::Builder => { - let mut client = BuilderApiClient::new(&mut node_api); - client - .delete_template(request, NODE_API_FAST_RPC_TIMEOUT) - .await - } - NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), - } - .map_err(|status| node_api_rpc_error_response(&node, "DeleteTemplate", status)); - - match response { - Ok(_) => { - note_node_template_deleted(&state.pool, &node, &template_name, None).await; - StatusCode::NO_CONTENT.into_response() - } - Err(err) => err.into_response(), - } -} - -async fn list_templates_for_node( - pool: &sqlx::PgPool, - node: NodeRow, -) -> NodeTemplateInventoryResponse { - let node_response = node_response(node.clone()); - if matches!(node.role, NodeRole::Cluster) { - return NodeTemplateInventoryResponse { - node: node_response, - reachable: true, - error: Some("cluster nodes do not host templates".to_string()), - builds: Vec::new(), - }; - } - - match sqlx::query_as::<_, NodeTemplateInventoryRow>(sql!( - "SELECT template_name AS name, - NULLIF(content_hash, '') AS content_hash, - built_at AS build_date, - artifact_size_bytes, - cache_size_bytes, - disk_size_mb, - runtime, - arch, - snapshots - FROM node_template_inventory - WHERE node_id = $1 - ORDER BY template_name ASC, content_hash ASC" - )) - .bind(node.id) - .fetch_all(pool) - .await - { - Ok(rows) => NodeTemplateInventoryResponse { - node: node_response, - reachable: true, - error: None, - builds: rows - .into_iter() - .map(|build| NodeTemplateBuildResponse { - name: build.name, - content_hash: build.content_hash, - build_date: build.build_date, - artifact_size_bytes: build - .artifact_size_bytes - .and_then(|value| u64::try_from(value).ok()), - cache_size_bytes: build - .cache_size_bytes - .and_then(|value| u64::try_from(value).ok()), - disk_size_mb: build - .disk_size_mb - .and_then(|value| u32::try_from(value).ok()), - runtime: build.runtime, - arch: build.arch, - snapshots: build.snapshots.0, - }) - .collect(), - }, - Err(err) => NodeTemplateInventoryResponse { - node: node_response, - reachable: false, - error: Some(format!( - "query cached node template inventory failed: {err}" - )), - builds: Vec::new(), - }, - } -} - -pub async fn list_node_templates( - State(state): State, - Extension(AuthUser(actor)): Extension, - Query(filters): Query, -) -> impl IntoResponse { - if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { - return err.into_response(); - } - let role_filter = filters.role; - let nodes = match fetch_inventory_nodes(&state, role_filter).await { - Ok(nodes) => nodes - .into_iter() - .filter(|node| role_filter.is_some() || node_supports_state_reconciliation(node.role)) - .collect::>(), - Err(err) => return err.into_response(), - }; - - let pool = state.pool.clone(); - let mut inventories = stream::iter(nodes.into_iter().enumerate()) - .map(|(index, node)| { - let pool = pool.clone(); - async move { (index, list_templates_for_node(&pool, node).await) } - }) - .buffer_unordered(NODE_TEMPLATE_INVENTORY_CONCURRENCY) - .collect::>() - .await; - inventories.sort_by_key(|(index, _)| *index); - Json( - inventories - .into_iter() - .map(|(_, inventory)| inventory) - .collect::>(), - ) - .into_response() -} - #[cfg(test)] mod tests; diff --git a/crates/libs/api/src/routes/admin/node_maintenance/node_api.rs b/crates/libs/api/src/routes/admin/node_maintenance/node_api.rs new file mode 100644 index 000000000..0c36aeb20 --- /dev/null +++ b/crates/libs/api/src/routes/admin/node_maintenance/node_api.rs @@ -0,0 +1,195 @@ +//! Reaching a node's node-api from admin routes: endpoint normalization and verification, the +//! client, and RPC error mapping. + +use super::*; + +pub(super) const NODE_API_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); + +pub(super) const NODE_API_FAST_RPC_TIMEOUT: Duration = Duration::from_secs(10); + +pub(super) fn normalize_node_api_endpoint( + raw: &str, +) -> Result)> { + let mut endpoint = raw.trim().trim_end_matches('/').to_string(); + if let Some(stripped) = endpoint.strip_prefix("http://") { + endpoint = stripped.to_string(); + } + if endpoint.starts_with("https://") { + return Err(api_error( + StatusCode::BAD_REQUEST, + "node_api_endpoint must use plaintext node-api host:port", + )); + } + if endpoint.is_empty() || endpoint.contains('/') { + return Err(api_error( + StatusCode::BAD_REQUEST, + "node_api_endpoint must be a host:port endpoint", + )); + } + let uri: axum::http::Uri = endpoint_uri(&endpoint).parse().map_err(|err| { + api_error( + StatusCode::BAD_REQUEST, + format!("invalid node_api_endpoint: {err}"), + ) + })?; + if uri + .authority() + .and_then(axum::http::uri::Authority::port_u16) + .is_none() + { + return Err(api_error( + StatusCode::BAD_REQUEST, + "node_api_endpoint must include an explicit port", + )); + } + Ok(endpoint) +} + +pub(super) fn endpoint_uri(endpoint: &str) -> String { + pplx_space_common_grpc::internal_transport::qualify_endpoint(endpoint) +} + +pub(super) fn node_endpoint( + endpoint: &str, + request_timeout: Option, +) -> Result> { + // `node_api_endpoint` is scheme-aware: it supplies the internal RPC scheme for bare + // `host:port` and passes `http://`/`https://`-qualified inputs through + // untouched. Trim whitespace only — stripping `http://` here would mangle + // an `https://` input into `http://https://...`. + let endpoint = endpoint.trim(); + Ok(crate::controllers::substrate::node_api_endpoint( + endpoint, + NODE_API_CONNECT_TIMEOUT, + request_timeout.unwrap_or(crate::controllers::substrate::NODE_RPC_TIMEOUT), + )?) +} + +pub(super) fn proto_role_name(role: i32) -> Option<&'static str> { + if role == ProtoNodeRole::Worker as i32 { + Some(NodeRole::Worker.as_str()) + } else if role == ProtoNodeRole::Builder as i32 { + Some(NodeRole::Builder.as_str()) + } else if role == ProtoNodeRole::Cluster as i32 { + Some(NodeRole::Cluster.as_str()) + } else { + None + } +} + +pub(super) fn bearer_kind_for_role(role: NodeRole) -> SpaceBearerToken { + match role { + NodeRole::Cluster => SpaceBearerToken::ClusterApi, + NodeRole::Worker | NodeRole::Builder => SpaceBearerToken::NodeApi, + } +} + +pub(super) async fn verify_node_endpoint( + endpoint: &str, + expected_role: NodeRole, +) -> Result)> { + let endpoint = node_endpoint(endpoint, Some(NODE_API_FAST_RPC_TIMEOUT)).map_err(|err| { + api_error( + StatusCode::BAD_REQUEST, + format!("invalid node_api_endpoint: {err}"), + ) + })?; + let service = match expected_role { + NodeRole::Cluster => pplx_space_common_grpc::internal_transport::Service::ClusterApi, + NodeRole::Worker | NodeRole::Builder => { + pplx_space_common_grpc::internal_transport::Service::NodeApi + } + }; + let endpoint = + pplx_space_common_grpc::internal_transport::client_connect_endpoint(service, endpoint) + .await + .map_err(|error| api_error(StatusCode::BAD_GATEWAY, error.to_string()))?; + let channel = endpoint.connect().await.map_err(|err| { + api_error( + StatusCode::BAD_GATEWAY, + format!("node-api endpoint is not reachable: {err}"), + ) + })?; + let mut client = NodeInfoClient::new(crate::controllers::substrate::node_api_channel( + channel, + bearer_kind_for_role(expected_role), + )); + let response = client + .get_node_info(GetNodeInfoRequest {}) + .await + .map_err(|err| { + api_error( + StatusCode::BAD_GATEWAY, + format!("node-api NodeInfo probe failed: {err}"), + ) + })? + .into_inner(); + + let actual_role = proto_role_name(response.role).ok_or_else(|| { + api_error( + StatusCode::BAD_GATEWAY, + format!("node-api returned unknown role {}", response.role), + ) + })?; + if actual_role != expected_role.as_str() { + return Err(api_error( + StatusCode::BAD_REQUEST, + format!("node-api role mismatch: expected {expected_role}, got {actual_role}"), + )); + } + + response.node_id.parse::().map_err(|err| { + api_error( + StatusCode::BAD_GATEWAY, + format!("node-api returned invalid node_id: {err}"), + ) + }) +} + +pub(super) fn node_api_client( + node: &NodeRow, + label: &str, + request_timeout: Duration, +) -> Result)> { + let raw_endpoint = node.node_api_endpoint.as_deref().ok_or_else(|| { + api_error( + StatusCode::INTERNAL_SERVER_ERROR, + format!("{label} node has no node_api_endpoint"), + ) + })?; + let endpoint = node_endpoint(raw_endpoint, Some(request_timeout)).map_err(|err| { + api_error( + StatusCode::BAD_REQUEST, + format!("invalid {label} node_api_endpoint: {err}"), + ) + })?; + Ok(NodeApiClient::new(endpoint)) +} + +pub(super) fn node_rpc_error( + operation: &str, + err: tonic::Status, +) -> (StatusCode, Json) { + let status = match err.code() { + tonic::Code::InvalidArgument => StatusCode::BAD_REQUEST, + tonic::Code::NotFound => StatusCode::NOT_FOUND, + tonic::Code::FailedPrecondition => StatusCode::CONFLICT, + _ => StatusCode::BAD_GATEWAY, + }; + api_error(status, format!("{operation} failed: {err}")) +} + +pub(super) fn node_api_rpc_error_response( + node: &NodeRow, + command: &'static str, + status: tonic::Status, +) -> (StatusCode, Json) { + tracing::warn!( + node_id = %node.id, + endpoint = %node.node_api_endpoint.as_deref().unwrap_or("-"), + command, + grpc_code = ?status.code(), + "node-api RPC failed" + ); + node_rpc_error(command, status) +} diff --git a/crates/libs/api/src/routes/admin/node_maintenance/node_status.rs b/crates/libs/api/src/routes/admin/node_maintenance/node_status.rs new file mode 100644 index 000000000..bebbfbad1 --- /dev/null +++ b/crates/libs/api/src/routes/admin/node_maintenance/node_status.rs @@ -0,0 +1,504 @@ +//! Admin node status, health and reconcile routes, and the cached node status they read. + +use super::*; + +pub(super) const DEFAULT_NODE_STATUS_CACHE_TTL_SECS: u64 = 30; + +#[derive(Debug, Clone, FromRow)] +pub(super) struct NodeStatusRow { + pub(super) node_id: NodeId, + pub(super) stale_after: OffsetDateTime, + pub(super) reachable: bool, + pub(super) error: Option, + pub(super) health_healthy: Option, + pub(super) services: sqlx::types::Json>, + pub(super) node_ctl: Option>, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct ServiceHealthResponse { + pub(super) name: String, + pub(super) reachable: bool, + pub(super) error: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct ReconciliationStatusResponse { + pub(super) name: String, + pub(super) poll_interval_secs: u64, + pub(super) last_started_at_unix_secs: u64, + pub(super) last_completed_at_unix_secs: u64, + pub(super) last_success_at_unix_secs: u64, + pub(super) next_due_at_unix_secs: u64, + pub(super) running: bool, + pub(super) templates_seen: u64, + pub(super) sandboxes_seen: u64, + pub(super) templates_applied: u64, + pub(super) sandboxes_applied: u64, + pub(super) last_error: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct NodeCtlStatusResponse { + pub(super) reachable: bool, + pub(super) error: Option, + pub(super) reconciliations: Vec, +} + +#[derive(Debug, Clone, Serialize)] +pub(super) struct NodeHealthStatusResponse { + pub(super) reachable: bool, + pub(super) healthy: bool, + pub(super) error: Option, + pub(super) services: Vec, + pub(super) node_ctl: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RefreshNodeStatusRequest { + pub(super) role: Option, + #[serde(default)] + pub(super) force: bool, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ReconcileFacetRequest { + #[serde(deserialize_with = "deserialize_reconcile_facet")] + pub(super) facet: ReconcileFacet, + pub(super) element: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ReconcileNodesRequest { + pub(super) role: Option, + #[serde(default)] + pub(super) node_ids: Vec, + #[serde(default, deserialize_with = "deserialize_reconcile_facets")] + pub(super) facets: Vec, + #[serde(default)] + pub(super) refs: Vec, +} + +#[derive(Debug, Serialize)] +pub(super) struct RefreshNodeStatusResponse { + pub(super) refreshed: bool, + pub(super) nodes_seen: usize, + pub(super) nodes_updated: usize, + pub(super) nodes_failed: usize, + pub(super) cache_ttl_secs: u64, + pub(super) error: Option, +} + +#[derive(Debug, Serialize)] +pub(super) struct ReconcileNodeResponse { + pub(super) node_id: NodeId, + pub(super) node_name: Option, + pub(super) role: String, + pub(super) queued: bool, + pub(super) error: Option, +} + +#[derive(Debug, Serialize)] +pub(super) struct ReconcileNodesResponse { + pub(super) nodes_seen: usize, + pub(super) nodes_queued: usize, + pub(super) nodes_failed: usize, + pub(super) nodes: Vec, +} + +#[derive(Debug, Serialize)] +pub(super) struct ReconcileNodeNetworkPolicyResponse { + pub(super) queued: bool, +} + +pub(super) fn deserialize_reconcile_facet<'de, D>( + deserializer: D, +) -> Result +where + D: Deserializer<'de>, +{ + let value = String::deserialize(deserializer)?; + ReconcileFacet::parse(value.trim()) + .ok_or_else(|| de::Error::custom(format!("unsupported reconcile facet: {value}"))) +} + +pub(super) fn deserialize_reconcile_facets<'de, D>( + deserializer: D, +) -> Result, D::Error> +where + D: Deserializer<'de>, +{ + Vec::::deserialize(deserializer)? + .into_iter() + .map(|value| { + ReconcileFacet::parse(value.trim()) + .ok_or_else(|| de::Error::custom(format!("unsupported reconcile facet: {value}"))) + }) + .collect() +} + +pub(super) fn node_supports_state_reconciliation(role: NodeRole) -> bool { + matches!(role, NodeRole::Worker | NodeRole::Builder) +} + +pub(super) async fn fetch_inventory_node( + state: &AppState, + id: NodeId, +) -> Result)> { + let nodes = fetch_inventory_nodes(state, None).await?; + nodes + .into_iter() + .find(|node| node.id == id && node.deleted_at.is_none()) + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "Node not found")) +} + +pub(super) async fn fetch_node_status_map( + pool: &sqlx::PgPool, + node_ids: &[NodeId], +) -> Result, sqlx::Error> { + if node_ids.is_empty() { + return Ok(Vec::new()); + } + sqlx::query_as::<_, NodeStatusRow>(sql!( + "SELECT node_id, + stale_after, + reachable, + error, + health_healthy, + services, + node_ctl + FROM node_status_snapshots + WHERE node_id = ANY($1)" + )) + .bind(node_ids) + .fetch_all(pool) + .await +} + +pub(super) async fn node_status_refresh_needed( + pool: &sqlx::PgPool, + nodes: &[NodeRow], +) -> Result { + let node_ids = nodes + .iter() + .filter(|node| node_supports_state_reconciliation(node.role)) + .map(|node| node.id) + .collect::>(); + if node_ids.is_empty() { + return Ok(false); + } + let statuses = fetch_node_status_map(pool, &node_ids).await?; + let now = OffsetDateTime::now_utc(); + Ok(node_ids.iter().any(|node_id| { + statuses + .iter() + .find(|status| status.node_id == *node_id) + .map(|status| status.stale_after <= now) + .unwrap_or(true) + })) +} + +pub(super) async fn fetch_cached_node_health( + pool: &sqlx::PgPool, + node: &NodeRow, +) -> NodeHealthStatusResponse { + if !node_supports_state_reconciliation(node.role) { + return NodeHealthStatusResponse { + reachable: false, + healthy: false, + error: Some("node health is only available for worker and builder nodes".to_string()), + services: Vec::new(), + node_ctl: None, + }; + } + + match fetch_node_status_map(pool, &[node.id]).await { + Ok(statuses) => { + let Some(status) = statuses + .into_iter() + .find(|status| status.node_id == node.id) + else { + return NodeHealthStatusResponse { + reachable: false, + healthy: false, + error: Some("cached node health is unavailable".to_string()), + services: Vec::new(), + node_ctl: None, + }; + }; + NodeHealthStatusResponse { + reachable: status.reachable, + healthy: status.health_healthy.unwrap_or(false), + error: status.error, + services: status.services.0, + node_ctl: status.node_ctl.map(|node_ctl| node_ctl.0), + } + } + Err(err) => NodeHealthStatusResponse { + reachable: false, + healthy: false, + error: Some(format!("cached node health query failed: {err}")), + services: Vec::new(), + node_ctl: None, + }, + } +} + +pub(super) fn default_node_publish_refs() -> Vec { + vec![ + ReconcileRef::full(ReconcileFacet::NodeStatus), + ReconcileRef::full(ReconcileFacet::SandboxInventory), + ReconcileRef::full(ReconcileFacet::TemplateInventory), + ] +} + +pub(super) fn reconcile_refs_from_request( + req: &ReconcileNodesRequest, +) -> Result, (StatusCode, Json)> { + if !req.refs.is_empty() { + return req + .refs + .iter() + .map(|reference| { + ReconcileRef::from_storage(reference.facet, reference.element.as_deref()).map_err( + |error| { + api_error( + StatusCode::BAD_REQUEST, + format!("invalid reconcile ref element: {error}"), + ) + }, + ) + }) + .collect(); + } + + let facets = if req.facets.is_empty() { + vec![ + ReconcileFacet::SandboxInventory, + ReconcileFacet::TemplateInventory, + ReconcileFacet::NodeStatus, + ] + } else { + req.facets.clone() + }; + facets + .into_iter() + .map(|facet| Ok(ReconcileRef::full(facet))) + .collect() +} + +pub async fn refresh_node_status( + State(state): State, + Extension(AuthUser(actor)): Extension, + Json(req): Json, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + let cache_ttl_secs = DEFAULT_NODE_STATUS_CACHE_TTL_SECS; + let nodes = match fetch_inventory_nodes(&state, req.role).await { + Ok(nodes) => nodes + .into_iter() + .filter(|node| node_supports_state_reconciliation(node.role)) + .collect::>(), + Err(err) => return err.into_response(), + }; + + if !req.force { + match node_status_refresh_needed(&state.pool, &nodes).await { + Ok(true) => {} + Ok(false) => { + return Json(RefreshNodeStatusResponse { + refreshed: false, + nodes_seen: nodes.len(), + nodes_updated: 0, + nodes_failed: 0, + cache_ttl_secs, + error: None, + }) + .into_response(); + } + Err(err) => return db_error(err).into_response(), + } + } + + let node_ids = nodes.iter().map(|node| node.id).collect::>(); + let enqueued = match enqueue_node_reconcile_refs_for_nodes( + &state.pool, + node_ids, + &default_node_publish_refs(), + ) + .await + { + Ok(enqueued) => enqueued, + Err(error) => { + tracing::warn!(%error, "failed to enqueue node status refresh reconciliation"); + Vec::new() + } + }; + let trigger_nodes = enqueued + .iter() + .map(|enqueued| enqueued.node_id) + .collect::>(); + if let Err(error) = enqueue_triggers( + &state.pool, + trigger_nodes, + ReconcileReason::ManualNodeStatusRefresh, + ) + .await + { + tracing::warn!(%error, "failed to enqueue manual node status refresh triggers"); + } + let nodes_seen = nodes.len(); + let nodes_updated = enqueued.len(); + Json(RefreshNodeStatusResponse { + refreshed: true, + nodes_seen, + nodes_updated, + nodes_failed: nodes_seen.saturating_sub(nodes_updated), + cache_ttl_secs, + error: None, + }) + .into_response() +} + +pub async fn reconcile_nodes( + State(state): State, + Extension(AuthUser(actor)): Extension, + Json(req): Json, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + let refs = match reconcile_refs_from_request(&req) { + Ok(refs) => refs, + Err(err) => return err.into_response(), + }; + + let requested_node_ids = req.node_ids.iter().copied().collect::>(); + let mut nodes = match fetch_inventory_nodes(&state, req.role).await { + Ok(nodes) => nodes + .into_iter() + .filter(|node| node_supports_state_reconciliation(node.role)) + .filter(|node| requested_node_ids.is_empty() || requested_node_ids.contains(&node.id)) + .collect::>(), + Err(err) => return err.into_response(), + }; + nodes.sort_by(|a, b| a.role.cmp(&b.role).then_with(|| a.name.cmp(&b.name))); + + let node_ids = nodes.iter().map(|node| node.id).collect::>(); + let queued_node_ids = + match enqueue_node_reconcile_refs_for_nodes(&state.pool, node_ids, &refs).await { + Ok(enqueued) => enqueued + .into_iter() + .map(|enqueued| enqueued.node_id) + .collect::>(), + Err(error) => { + tracing::warn!(%error, "failed to enqueue manual node reconciliation"); + HashSet::new() + } + }; + let enqueue_error = if queued_node_ids.is_empty() && !nodes.is_empty() { + Some("enqueue reconcile refs failed".to_string()) + } else { + None + }; + let nodes = nodes + .into_iter() + .map(|node| { + let queued = queued_node_ids.contains(&node.id); + ReconcileNodeResponse { + node_id: node.id, + node_name: node.name, + role: node.role.to_string(), + queued, + error: if queued { None } else { enqueue_error.clone() }, + } + }) + .collect::>(); + let nodes_queued = nodes.iter().filter(|node| node.queued).count(); + let trigger_nodes = nodes + .iter() + .filter(|node| node.queued) + .map(|node| node.node_id) + .collect::>(); + if let Err(error) = enqueue_triggers( + &state.pool, + trigger_nodes, + ReconcileReason::ManualNodeReconcile, + ) + .await + { + tracing::warn!(%error, "failed to enqueue manual node reconcile triggers"); + } + let nodes_seen = nodes.len(); + Json(ReconcileNodesResponse { + nodes_seen, + nodes_queued, + nodes_failed: nodes_seen.saturating_sub(nodes_queued), + nodes, + }) + .into_response() +} + +pub async fn node_health( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path(id): Path, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + + let node = match fetch_inventory_node(&state, id).await { + Ok(node) => node, + Err(err) => return err.into_response(), + }; + if !node_supports_state_reconciliation(node.role) { + return api_error( + StatusCode::BAD_REQUEST, + "node health is only available for worker and builder nodes", + ) + .into_response(); + } + + Json(fetch_cached_node_health(&state.pool, &node).await).into_response() +} + +pub async fn reconcile_node_network_policy( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path(id): Path, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + + let node = match fetch_inventory_node(&state, id).await { + Ok(node) => node, + Err(err) => return err.into_response(), + }; + if !node_supports_state_reconciliation(node.role) { + return api_error( + StatusCode::BAD_REQUEST, + "network policy reconciliation is only available for worker and builder nodes", + ) + .into_response(); + } + + let refs = [ReconcileRef::full(ReconcileFacet::NetworkPolicy)]; + match enqueue_node_reconcile_refs_and_trigger( + &state.pool, + node.id, + &refs, + ReconcileReason::ManualNodeNetworkPolicyReconcile, + ) + .await + { + Ok(_) => Json(ReconcileNodeNetworkPolicyResponse { queued: true }).into_response(), + Err(err) => db_error(err).into_response(), + } +} diff --git a/crates/libs/api/src/routes/admin/node_maintenance/node_templates.rs b/crates/libs/api/src/routes/admin/node_maintenance/node_templates.rs new file mode 100644 index 000000000..f16608926 --- /dev/null +++ b/crates/libs/api/src/routes/admin/node_maintenance/node_templates.rs @@ -0,0 +1,550 @@ +//! Admin node template routes: pull and delete a template on a node through its node-api, and list +//! a node's templates from the cached template inventory. + +use super::*; + +pub(super) const NODE_API_TEMPLATE_RPC_TIMEOUT: Duration = Duration::from_secs(600); + +pub(super) const NODE_TEMPLATE_INVENTORY_CONCURRENCY: usize = 16; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(super) struct NodeTemplateSnapshotResponse { + pub(super) vcpu: u32, + pub(super) memory_mib: u32, + pub(super) snapshot_hash: String, +} + +#[derive(Debug, FromRow)] +pub(super) struct NodeTemplateInventoryRow { + pub(super) name: String, + pub(super) content_hash: Option, + pub(super) build_date: Option, + pub(super) artifact_size_bytes: Option, + pub(super) cache_size_bytes: Option, + pub(super) disk_size_mb: Option, + pub(super) runtime: Option, + pub(super) arch: Option, + pub(super) snapshots: sqlx::types::Json>, +} + +#[derive(Debug, Serialize)] +pub(super) struct NodeTemplateBuildResponse { + pub(super) name: String, + pub(super) content_hash: Option, + pub(super) build_date: Option, + pub(super) artifact_size_bytes: Option, + pub(super) cache_size_bytes: Option, + pub(super) disk_size_mb: Option, + pub(super) runtime: Option, + pub(super) arch: Option, + pub(super) snapshots: Vec, +} + +#[derive(Debug, Serialize)] +pub(super) struct NodeTemplateInventoryResponse { + pub(super) node: NodeResponse, + pub(super) reachable: bool, + pub(super) error: Option, + pub(super) builds: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct NodeFilters { + pub(super) role: Option, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct PullNodeTemplateRequest { + pub(super) template_id: String, + pub(super) content_hash: Option, + pub(super) source_node_id: NodeId, + #[serde(default)] + pub(super) include_snapshot: bool, + pub(super) snapshot_vcpu: Option, + pub(super) snapshot_memory_mib: Option, + #[serde(default = "default_true")] + pub(super) skip_if_exists: bool, + pub(super) local_tag: Option, + pub(super) block_on_snapshot_policy_achieved: Option, +} + +#[derive(Debug, Deserialize)] +pub struct DeleteNodeTemplateQuery { + pub(super) content_hash: Option, +} + +#[derive(Debug, Serialize)] +pub(super) struct PullNodeTemplateResponse { + pub(super) content_hash: String, + pub(super) already_existed: bool, + pub(super) bytes_transferred: u64, +} + +pub(super) fn default_true() -> bool { + true +} + +pub(super) fn template_capable_node_role( + node: &NodeRow, + label: &str, +) -> Result)> { + match node.role { + role @ (NodeRole::Worker | NodeRole::Builder) => Ok(role), + NodeRole::Cluster => Err(api_error( + StatusCode::BAD_REQUEST, + format!("{label} node must be a worker or builder"), + )), + } +} + +pub(super) fn ensure_template_capable_node( + node: &NodeRow, + label: &str, +) -> Result<(), (StatusCode, Json)> { + if node.deleted_at.is_some() { + return Err(api_error( + StatusCode::NOT_FOUND, + format!("{label} node not found"), + )); + } + template_capable_node_role(node, label)?; + Ok(()) +} + +pub(super) fn non_empty_optional(value: Option) -> Option { + value.and_then(|value| { + let value = value.trim().to_string(); + if value.is_empty() { None } else { Some(value) } + }) +} + +pub(super) async fn delete_cached_node_template_inventory( + pool: &sqlx::PgPool, + node_id: NodeId, + template: &str, + content_hash: Option<&str>, +) -> Result<(), sqlx::Error> { + if let Some(content_hash) = content_hash { + sqlx::query(sql!( + "DELETE FROM node_template_inventory + WHERE node_id = $1 + AND template_name = $2 + AND content_hash = $3" + )) + .bind(node_id) + .bind(template) + .bind(content_hash) + .execute(pool) + .await?; + } else { + sqlx::query(sql!( + "DELETE FROM node_template_inventory + WHERE node_id = $1 + AND template_name = $2" + )) + .bind(node_id) + .bind(template) + .execute(pool) + .await?; + } + + let remaining: i64 = sqlx::query_scalar(sql!( + "SELECT COUNT(*)::BIGINT + FROM node_template_inventory + WHERE node_id = $1 + AND template_name = $2" + )) + .bind(node_id) + .bind(template) + .fetch_one(pool) + .await?; + + if remaining == 0 { + sqlx::query(sql!( + "DELETE FROM node_facet_observed_status + WHERE node_id = $1 + AND facet = $2 + AND element = $3" + )) + .bind(node_id) + .bind(FACET_NETWORK_POLICY) + .bind(format!("template:{template}")) + .execute(pool) + .await?; + } + + Ok(()) +} + +pub(super) async fn note_node_template_deleted( + pool: &sqlx::PgPool, + node: &NodeRow, + template: &str, + content_hash: Option<&str>, +) { + if let Err(error) = + delete_cached_node_template_inventory(pool, node.id, template, content_hash).await + { + tracing::warn!( + node_id = %node.id, + template, + content_hash = content_hash.unwrap_or(""), + %error, + "failed to delete cached node template inventory after node delete" + ); + } + + if let Err(error) = enqueue_node_reconcile_refs_and_trigger( + pool, + node.id, + &[ReconcileRef::full(ReconcileFacet::TemplateInventory)], + ReconcileReason::ManualNodeReconcile, + ) + .await + { + tracing::warn!( + node_id = %node.id, + template, + content_hash = content_hash.unwrap_or(""), + %error, + "failed to enqueue template inventory refresh after node template delete" + ); + } +} + +pub async fn pull_node_template( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path(target_id): Path, + Json(req): Json, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + if let Err(err) = require_admin_node_registry(&state) { + return err.into_response(); + } + + let template_id = req.template_id.trim().to_string(); + if template_id.is_empty() { + return api_error(StatusCode::BAD_REQUEST, "template_id is required").into_response(); + } + let template_name = match TemplateName::parse(&template_id) { + Ok(template_name) => template_name, + Err(err) => { + return api_error( + StatusCode::BAD_REQUEST, + format!("invalid template_id: {err}"), + ) + .into_response(); + } + }; + + if req.include_snapshot && (req.snapshot_vcpu.is_none() || req.snapshot_memory_mib.is_none()) { + return api_error( + StatusCode::BAD_REQUEST, + "snapshot_vcpu and snapshot_memory_mib are required when include_snapshot is true", + ) + .into_response(); + } + + let target = match fetch_node(&state.pool, target_id).await { + Ok(Some(node)) => node, + Ok(None) => { + return api_error(StatusCode::NOT_FOUND, "target node not found").into_response(); + } + Err(err) => return db_error(err).into_response(), + }; + let target_role = match template_capable_node_role(&target, "target") { + Ok(role) => role, + Err(err) => return err.into_response(), + }; + + let source = match fetch_node(&state.pool, req.source_node_id).await { + Ok(Some(node)) => node, + Ok(None) => { + return api_error(StatusCode::NOT_FOUND, "source node not found").into_response(); + } + Err(err) => return db_error(err).into_response(), + }; + if let Err(err) = ensure_template_capable_node(&source, "source") { + return err.into_response(); + } + + let source_address = match source.node_api_endpoint.clone() { + Some(value) => value, + None => { + return api_error( + StatusCode::INTERNAL_SERVER_ERROR, + "source node has no node_api_endpoint", + ) + .into_response(); + } + }; + let request = PullTemplateRequest { + template_id: template_id.clone(), + content_hash: non_empty_optional(req.content_hash), + tag: None, + sources: vec![TransferSource { + source: Some(Source::Node(NodeSource { + address: source_address, + })), + }], + include_snapshot: req.include_snapshot, + snapshot_vcpu: req.snapshot_vcpu, + snapshot_memory_mib: req.snapshot_memory_mib, + skip_if_exists: req.skip_if_exists, + local_tag: non_empty_optional(req.local_tag), + block_on_snapshot_policy_achieved: req.block_on_snapshot_policy_achieved, + }; + + let mut node_api = match node_api_client(&target, "target", NODE_API_TEMPLATE_RPC_TIMEOUT) { + Ok(client) => client, + Err(err) => return err.into_response(), + }; + let response = match target_role { + NodeRole::Worker => { + let mut client = WorkerApiClient::new(&mut node_api); + client + .pull_template(request, NODE_API_TEMPLATE_RPC_TIMEOUT) + .await + } + NodeRole::Builder => { + let mut client = BuilderApiClient::new(&mut node_api); + client + .pull_template(request, NODE_API_TEMPLATE_RPC_TIMEOUT) + .await + } + NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), + } + .map_err(|status| node_api_rpc_error_response(&target, "PullTemplate", status)); + + match response { + Ok(response) => { + if let Err(err) = enqueue_node_reconcile_refs_and_trigger( + &state.pool, + target.id, + &[ReconcileRef::template_inventory(template_name)], + ReconcileReason::NodeTemplatePulled, + ) + .await + { + tracing::warn!(error = %err, "failed to enqueue pulled template inventory reconcile"); + } + let response = response.into_inner(); + Json(PullNodeTemplateResponse { + content_hash: response.content_hash, + already_existed: response.already_existed, + bytes_transferred: response.bytes_transferred, + }) + .into_response() + } + Err(err) => err.into_response(), + } +} + +pub async fn delete_node_template( + State(state): State, + Extension(AuthUser(actor)): Extension, + Path((node_id, name)): Path<(NodeId, String)>, + Query(query): Query, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + if let Err(err) = require_admin_node_registry(&state) { + return err.into_response(); + } + + let name = name.trim().to_string(); + if name.is_empty() { + return api_error(StatusCode::BAD_REQUEST, "template name is required").into_response(); + } + + let node = match fetch_node(&state.pool, node_id).await { + Ok(Some(node)) => node, + Ok(None) => return api_error(StatusCode::NOT_FOUND, "node not found").into_response(), + Err(err) => return db_error(err).into_response(), + }; + let target_role = match template_capable_node_role(&node, "target") { + Ok(role) => role, + Err(err) => return err.into_response(), + }; + let mut node_api = match node_api_client(&node, "target", NODE_API_FAST_RPC_TIMEOUT) { + Ok(client) => client, + Err(err) => return err.into_response(), + }; + + if let Some(content_hash) = non_empty_optional(query.content_hash) { + let template_name = name.clone(); + let content_hash_for_cache = content_hash.clone(); + let request = DeleteTemplateBuildRequest { + template: Some(TemplateRef { + name, + content_hash: Some(content_hash), + tag: None, + }), + }; + let response = match target_role { + NodeRole::Worker => { + let mut client = WorkerApiClient::new(&mut node_api); + client + .delete_template_build(request, NODE_API_FAST_RPC_TIMEOUT) + .await + } + NodeRole::Builder => { + let mut client = BuilderApiClient::new(&mut node_api); + client + .delete_template_build(request, NODE_API_FAST_RPC_TIMEOUT) + .await + } + NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), + } + .map_err(|status| node_api_rpc_error_response(&node, "DeleteTemplateBuild", status)); + return match response { + Ok(_) => { + note_node_template_deleted( + &state.pool, + &node, + &template_name, + Some(&content_hash_for_cache), + ) + .await; + StatusCode::NO_CONTENT.into_response() + } + Err(err) => err.into_response(), + }; + } + + let template_name = name.clone(); + let request = DeleteTemplateRequest { template: name }; + let response = match target_role { + NodeRole::Worker => { + let mut client = WorkerApiClient::new(&mut node_api); + client + .delete_template(request, NODE_API_FAST_RPC_TIMEOUT) + .await + } + NodeRole::Builder => { + let mut client = BuilderApiClient::new(&mut node_api); + client + .delete_template(request, NODE_API_FAST_RPC_TIMEOUT) + .await + } + NodeRole::Cluster => unreachable!("cluster nodes are rejected above"), + } + .map_err(|status| node_api_rpc_error_response(&node, "DeleteTemplate", status)); + + match response { + Ok(_) => { + note_node_template_deleted(&state.pool, &node, &template_name, None).await; + StatusCode::NO_CONTENT.into_response() + } + Err(err) => err.into_response(), + } +} + +pub(super) async fn list_templates_for_node( + pool: &sqlx::PgPool, + node: NodeRow, +) -> NodeTemplateInventoryResponse { + let node_response = node_response(node.clone()); + if matches!(node.role, NodeRole::Cluster) { + return NodeTemplateInventoryResponse { + node: node_response, + reachable: true, + error: Some("cluster nodes do not host templates".to_string()), + builds: Vec::new(), + }; + } + + match sqlx::query_as::<_, NodeTemplateInventoryRow>(sql!( + "SELECT template_name AS name, + NULLIF(content_hash, '') AS content_hash, + built_at AS build_date, + artifact_size_bytes, + cache_size_bytes, + disk_size_mb, + runtime, + arch, + snapshots + FROM node_template_inventory + WHERE node_id = $1 + ORDER BY template_name ASC, content_hash ASC" + )) + .bind(node.id) + .fetch_all(pool) + .await + { + Ok(rows) => NodeTemplateInventoryResponse { + node: node_response, + reachable: true, + error: None, + builds: rows + .into_iter() + .map(|build| NodeTemplateBuildResponse { + name: build.name, + content_hash: build.content_hash, + build_date: build.build_date, + artifact_size_bytes: build + .artifact_size_bytes + .and_then(|value| u64::try_from(value).ok()), + cache_size_bytes: build + .cache_size_bytes + .and_then(|value| u64::try_from(value).ok()), + disk_size_mb: build + .disk_size_mb + .and_then(|value| u32::try_from(value).ok()), + runtime: build.runtime, + arch: build.arch, + snapshots: build.snapshots.0, + }) + .collect(), + }, + Err(err) => NodeTemplateInventoryResponse { + node: node_response, + reachable: false, + error: Some(format!( + "query cached node template inventory failed: {err}" + )), + builds: Vec::new(), + }, + } +} + +pub async fn list_node_templates( + State(state): State, + Extension(AuthUser(actor)): Extension, + Query(filters): Query, +) -> impl IntoResponse { + if let Err(err) = require_managing_org_admin(&state.pool, &actor).await { + return err.into_response(); + } + let role_filter = filters.role; + let nodes = match fetch_inventory_nodes(&state, role_filter).await { + Ok(nodes) => nodes + .into_iter() + .filter(|node| role_filter.is_some() || node_supports_state_reconciliation(node.role)) + .collect::>(), + Err(err) => return err.into_response(), + }; + + let pool = state.pool.clone(); + let mut inventories = stream::iter(nodes.into_iter().enumerate()) + .map(|(index, node)| { + let pool = pool.clone(); + async move { (index, list_templates_for_node(&pool, node).await) } + }) + .buffer_unordered(NODE_TEMPLATE_INVENTORY_CONCURRENCY) + .collect::>() + .await; + inventories.sort_by_key(|(index, _)| *index); + Json( + inventories + .into_iter() + .map(|(_, inventory)| inventory) + .collect::>(), + ) + .into_response() +} From f90d4e12b395b065d4bb0e49ad24f78c03798593 Mon Sep 17 00:00:00 2001 From: Abhinav Chordia Date: Fri, 16 Oct 2026 06:39:08 +0000 Subject: [PATCH 2/2] docs: point references at the split node_maintenance.rs modules The Kind node-api chart comment cites pull_node_template by file. It now lives in node_maintenance/node_templates.rs; require_admin_node_registry stays in node_maintenance.rs. --- .../kind/charts/space-node-api-local/values.yaml | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/infra/space/kind/charts/space-node-api-local/values.yaml b/infra/space/kind/charts/space-node-api-local/values.yaml index 594af3b67..d6f785595 100644 --- a/infra/space/kind/charts/space-node-api-local/values.yaml +++ b/infra/space/kind/charts/space-node-api-local/values.yaml @@ -75,11 +75,13 @@ auth: # auth on, only these accounts get it (AuthPolicy::authenticated_caller, # auth/k8s.rs), and an empty list refuses every transfer. A pull is started # on the target as the api's account: its only in-tree caller, admin-api's - # pull_node_template (crates/libs/api/src/routes/admin/node_maintenance.rs), - # answers 501 on the Kubernetes substrate (require_admin_node_registry), so - # here a pull is an operator's direct PullTemplate call with a space-api - # token. The target's node-api, of either role, then asks the source for the - # export with its own token (connect_remote, transfer.rs). + # pull_node_template + # (crates/libs/api/src/routes/admin/node_maintenance/node_templates.rs), + # answers 501 on the Kubernetes substrate (require_admin_node_registry, + # node_maintenance.rs), so here a pull is an operator's direct PullTemplate + # call with a space-api token. The target's node-api, of either role, then + # asks the source for the export with its own token (connect_remote, + # transfer.rs). SPACE_NODE_API_AUTH_TEMPLATE_TRANSFER_SERVICE_ACCOUNTS: - space-api - space-node-api