diff --git a/Cargo.lock b/Cargo.lock index ec987f870f..bcf64b0eef 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2504,6 +2504,7 @@ dependencies = [ "carbide-utils", "carbide-uuid", "chrono", + "component-manager", "config-version", "duration-str", "eyre", diff --git a/crates/api-core/src/handlers/component_manager.rs b/crates/api-core/src/handlers/component_manager.rs index 2c0fe34c00..ee82416289 100644 --- a/crates/api-core/src/handlers/component_manager.rs +++ b/crates/api-core/src/handlers/component_manager.rs @@ -28,7 +28,7 @@ use carbide_uuid::machine::MachineId; use carbide_uuid::power_shelf::PowerShelfId; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; -use component_manager::component_manager::ComponentManager; +use component_manager::component_manager::{ComponentManager, SwitchMaintenanceRequestResult}; use component_manager::compute_tray_manager::{ComputeTrayEndpoint, ComputeTrayVendor}; use component_manager::error::ComponentManagerError; use component_manager::nv_switch_manager::SwitchEndpoint; @@ -47,7 +47,6 @@ use model::rack::{FirmwareUpgradeJob, MaintenanceActivity}; use model::switch::SwitchMaintenanceOperation; use tonic::{Code, Request, Response, Status}; -use crate::CarbideError; use crate::api::{Api, log_request_data, log_request_data_redacted}; const MACHINE_POWER_OVERRIDE_SOURCE: &str = "component_power_control"; @@ -65,7 +64,7 @@ fn unsupported_from_json_firmware_versions(target: &str) -> Status { )) } -fn component_manager_error_to_status(err: ComponentManagerError) -> Status { +pub(crate) fn component_manager_error_to_status(err: ComponentManagerError) -> Status { match err { ComponentManagerError::Unavailable(msg) => Status::unavailable(msg), ComponentManagerError::NotFound(msg) => Status::not_found(msg), @@ -303,61 +302,43 @@ fn map_switch_maintenance_operation(action: PowerAction) -> SwitchMaintenanceOpe async fn queue_switch_power_control_via_state_controller( api: &Api, + cm: &ComponentManager, switch_ids: &[SwitchId], action: PowerAction, ) -> Result, Status> { let operation = map_switch_maintenance_operation(action); - queue_switch_maintenance_via_state_controller(api, switch_ids, operation).await + queue_switch_maintenance_via_state_controller(api, cm, switch_ids, operation).await } async fn queue_switch_maintenance_via_state_controller( api: &Api, + cm: &ComponentManager, switch_ids: &[SwitchId], operation: SwitchMaintenanceOperation, ) -> Result, Status> { - let mut txn = api.txn_begin().await?; - let existing = db::switch::find_by( - &mut txn, - db::ObjectColumnFilter::List(db::switch::IdColumn, switch_ids), - ) - .await - .map_err(CarbideError::from)?; - - let by_id: HashMap = - existing.into_iter().map(|sw| (sw.id, sw)).collect(); - let mut results = Vec::with_capacity(switch_ids.len()); - - for switch_id in switch_ids { - let Some(switch) = by_id.get(switch_id) else { - results.push(error_result( - &switch_id.to_string(), - format!("switch {switch_id} not found"), - )); - continue; - }; - - if switch.is_marked_as_deleted() { - results.push(error_result( - &switch_id.to_string(), - format!("switch {switch_id} is marked for deletion"), - )); - continue; - } - - db::switch::set_switch_maintenance_requested( - &mut txn, - *switch_id, - "component-manager", + let results = cm + .request_switch_maintenance_via_state_controller( + &api.database_connection, + switch_ids, operation, + "component-manager", ) .await - .map_err(CarbideError::from)?; + .map_err(component_manager_error_to_status)?; - results.push(success_result(&switch_id.to_string())); - } + Ok(results + .iter() + .map(switch_maintenance_request_result_to_component_result) + .collect()) +} - txn.commit().await?; - Ok(results) +fn switch_maintenance_request_result_to_component_result( + result: &SwitchMaintenanceRequestResult, +) -> rpc::ComponentResult { + match &result.error { + Some(error) => error_result(&result.switch_id.to_string(), error.clone()), + None => success_result(&result.switch_id.to_string()), + } } /// Maps raw proto `ComputeTrayComponent` values to display-name strings. @@ -1241,7 +1222,8 @@ pub(crate) async fn component_power_control( rpc::component_power_control_request::Target::SwitchIds(list) => { if cm.nv_switch_use_state_controller && !bypass_state_controller { let results = - queue_switch_power_control_via_state_controller(api, &list.ids, action).await?; + queue_switch_power_control_via_state_controller(api, cm, &list.ids, action) + .await?; (results, Vec::new()) } else { let endpoints = resolve_switch_endpoints(api, &list.ids).await?; @@ -1458,6 +1440,7 @@ pub(crate) async fn component_configure_switch_certificate( if cm.nv_switch_use_state_controller && !bypass_state_controller { let results = queue_switch_maintenance_via_state_controller( api, + cm, &switch_ids.ids, SwitchMaintenanceOperation::ReconfigureCertificate, ) diff --git a/crates/api-core/src/handlers/rack.rs b/crates/api-core/src/handlers/rack.rs index da3bacbe1d..e631e827e2 100644 --- a/crates/api-core/src/handlers/rack.rs +++ b/crates/api-core/src/handlers/rack.rs @@ -22,11 +22,14 @@ use ::rpc::forge::{self as rpc, HealthReportEntry}; use carbide_rack::firmware_object::{ rack_maintenance_access_token_key, rms_access_token_or_noauth, }; -use carbide_secrets::credentials::Credentials; use carbide_uuid::machine::MachineId; use carbide_uuid::power_shelf::PowerShelfId; use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; +use component_manager::component_manager::{ + RackMaintenanceAccessToken, RackMaintenanceEligibility, RackMaintenanceRequestOutcome, + request_rack_maintenance_via_state_controller, +}; use db::{ ObjectColumnFilter, WithTransaction, machine as db_machine, power_shelf as db_power_shelf, rack as db_rack, switch as db_switch, @@ -41,6 +44,7 @@ use tonic::{Request, Response, Status}; use crate::CarbideError; use crate::api::{Api, log_request_data, log_request_data_redacted}; use crate::auth::AuthContext; +use crate::handlers::component_manager::component_manager_error_to_status; pub async fn get_rack( api: &Api, @@ -766,58 +770,44 @@ pub(crate) async fn on_demand_rack_maintenance( } } - let access_token_stored = maintenance_access_token.is_some(); - if let Some(token) = maintenance_access_token { - api.credential_manager - .set_credentials( - &rack_maintenance_access_token_key(&rack_id), - &Credentials::UsernamePassword { - username: "access_token".into(), - password: token, - }, - ) - .await - .map_err(|error| CarbideError::Internal { - message: format!("failed to store rack maintenance access token: {error}"), - })?; - } - - let mut updated_config = rack.config.clone(); - updated_config.maintenance_requested = Some(scope); + let maintenance_access_token = + maintenance_access_token.map(|token| RackMaintenanceAccessToken { + credential_manager: api.credential_manager.as_ref(), + token, + }); - let db_result: Result<(), Status> = async { - let mut txn = api.txn_begin().await?; - db_rack::update(&mut txn, &rack_id, &updated_config).await?; - if updated_config - .maintenance_requested - .as_ref() - .is_some_and(|scope| { - scope.should_run(&MaintenanceActivity::FirmwareUpgrade { - firmware_version: None, - components: vec![], - force_update: false, - }) - }) - { - db_rack::update_firmware_upgrade_job(txn.as_mut(), &rack_id, None).await?; - } - txn.commit().await?; - Ok(()) - } + let schedule_result = request_rack_maintenance_via_state_controller( + &api.database_connection, + &rack_id, + scope, + RackMaintenanceEligibility::AllowErrorRecovery, + maintenance_access_token, + ) .await; - if let Err(status) = db_result { - if access_token_stored - && let Err(error) = api - .credential_manager - .delete_credentials(&rack_maintenance_access_token_key(&rack_id)) - .await - { - tracing::warn!( - rack_id = %rack_id, - error = %error, - "failed to delete rack maintenance access token after DB error", - ); - } + + let scheduling_error = match schedule_result { + Ok( + RackMaintenanceRequestOutcome::Scheduled + | RackMaintenanceRequestOutcome::AlreadyPending, + ) => None, + Ok(RackMaintenanceRequestOutcome::Busy) => Some( + CarbideError::InvalidArgument(format!( + "On-demand maintenance for rack {} is already scheduled.", + rack_id, + )) + .into(), + ), + Ok(RackMaintenanceRequestOutcome::Deferred { state }) => Some( + CarbideError::InvalidArgument(format!( + "Rack {} is not in Ready or Error state (current: {:?}). Maintenance can only be requested when the rack is Ready or in Error.", + rack_id, state, + )) + .into(), + ), + Err(error) => Some(component_manager_error_to_status(error)), + }; + + if let Some(status) = scheduling_error { return Err(status); } diff --git a/crates/api-core/src/setup.rs b/crates/api-core/src/setup.rs index 1716f4626f..d4353c6033 100644 --- a/crates/api-core/src/setup.rs +++ b/crates/api-core/src/setup.rs @@ -1493,9 +1493,7 @@ async fn initialize_and_start_controllers<'a>( meter: meter.clone(), config: carbide_config.nvlink_config.clone().unwrap_or_default(), host_health: carbide_config.host_health, - rms_client: api_service.rms_client.clone(), - credential_manager: api_service.credential_manager.clone(), - rack_profiles: carbide_config.rack_profiles.clone(), + component_manager: api_service.component_manager.clone().map(Arc::new), work_lock_manager_handle: work_lock_manager_handle.clone(), }) .start(join_set, cancel_token.clone())?; diff --git a/crates/api-core/src/tests/common/api_fixtures/mod.rs b/crates/api-core/src/tests/common/api_fixtures/mod.rs index a171ef29bd..075d9c6046 100644 --- a/crates/api-core/src/tests/common/api_fixtures/mod.rs +++ b/crates/api-core/src/tests/common/api_fixtures/mod.rs @@ -1455,9 +1455,7 @@ pub async fn create_test_env_with_overrides( db_pool.clone(), test_meter.meter(), config.nvlink_config.clone().unwrap(), - rms_sim.as_rms_client(), - composite_manager.clone(), - config.rack_profiles.clone(), + test_component_manager.clone(), api.work_lock_manager_handle.clone(), ); diff --git a/crates/api-core/src/tests/nvl_instance.rs b/crates/api-core/src/tests/nvl_instance.rs index 7221ffcc86..a0390330c0 100644 --- a/crates/api-core/src/tests/nvl_instance.rs +++ b/crates/api-core/src/tests/nvl_instance.rs @@ -35,13 +35,13 @@ use common::api_fixtures::{ TestEnv, TestManagedHost, create_managed_host_with_hardware_info_template, insert_nvlink_nmxc_endpoint_from_managed_host, }; -use db::switch as db_switch; +use db::{ObjectColumnFilter, rack as db_rack, switch as db_switch}; use ipnetwork::IpNetwork; use libnmxc::nmxc_model::{GetGpuInfoListRequest, GetPartitionInfoListRequest, GpuAttr}; -use librms::protos::rack_manager as rms; use model::expected_switch::ExpectedSwitch; use model::instance::config::nvlink::InstanceNvLinkConfig; use model::metadata::Metadata; +use model::rack::{MaintenanceActivity, RackState}; use model::switch::{ CONTROL_PLANE_STATE_CONFIGURED, FabricManagerState, FabricManagerStatus, NewSwitch, SwitchConfig, SwitchControllerState, @@ -2640,25 +2640,28 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( .persist(&mut txn) .await .expect("create rack"); + let rack = db_rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db_rack::IdColumn, &rack_id), + ) + .await + .expect("load rack") + .pop() + .expect("rack"); + let updated = db_rack::try_update_controller_state( + txn.as_mut(), + &rack_id, + rack.controller_state.version, + rack.controller_state.version.increment(), + &RackState::Ready, + ) + .await + .expect("set rack ready"); + assert!(updated, "rack should transition to Ready for rotation"); txn.commit().await.expect("commit rack"); let switch_id = create_rack_switch_for_nmxc_simulator(&env, &rack_id).await; - if expected_fingerprint_mismatches > 0 { - env.rms_sim - .queue_configure_switch_certificate_response(Ok( - rms::ConfigureSwitchCertificateResponse { - response: Some(rms::NodeBatchResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - ..Default::default() - }), - ..Default::default() - }, - )) - .await; - } - let result = env.run_switch_cert_monitor_iteration().await; assert_eq!(result.observed_endpoints, 1); assert_eq!(result.successful_probes, 1); @@ -2672,92 +2675,56 @@ async fn assert_switch_cert_monitor_nmxc_simulator_probe( assert_eq!(result.pending_updates, expected_fingerprint_mismatches); assert_eq!(result.apply_errors, 0); - let rms_requests = env - .rms_sim - .submitted_configure_switch_certificate_requests() - .await; - assert_eq!(rms_requests.len(), expected_fingerprint_mismatches); + let mut txn = pool.begin().await.expect("begin txn"); + let switch = db_switch::find_by_id(txn.as_mut(), &switch_id) + .await + .expect("load switch") + .expect("switch"); + let rack = db_rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db_rack::IdColumn, &rack_id), + ) + .await + .expect("load rack") + .pop() + .expect("rack"); + txn.commit().await.expect("commit txn"); + if expected_fingerprint_mismatches > 0 { - let request = rms_requests.first().expect("RMS request"); - assert_eq!( - request.services, - vec![rms::SwitchService::ScaleUpFabricManager as i32] + assert!( + switch.switch_maintenance_requested.is_none(), + "certificate rotation should not request per-switch maintenance" ); - assert!(request.test_hello); - assert_eq!(request.domain.as_deref(), Some(rack_id.as_ref())); - let node = request - .nodes + let scope = rack + .config + .maintenance_requested .as_ref() - .expect("RMS request node set") - .nodes - .first() - .expect("RMS request node"); - assert_eq!(node.node_id, switch_id.to_string()); - assert_eq!(node.rack_id, rack_id.to_string()); - - env.rms_sim - .queue_get_configure_switch_certificate_job_status_response(Ok( - rms::GetConfigureSwitchCertificateJobStatusResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - state: "running".to_string(), - ..Default::default() - }, - )) - .await; - let result = env.run_switch_cert_monitor_iteration().await; - assert_eq!(result.observed_endpoints, 1); - assert_eq!(result.successful_probes, 1); + .expect("rack NMX cluster maintenance request"); + assert!(scope.is_full_rack()); assert_eq!( - result.fingerprint_mismatches, - expected_fingerprint_mismatches + scope.activities, + vec![MaintenanceActivity::ConfigureNmxCluster] ); - assert_eq!(result.applied_updates, 0); - assert_eq!(result.pending_updates, expected_fingerprint_mismatches); - assert_eq!(result.apply_errors, 0); - assert_eq!( - env.rms_sim - .submitted_configure_switch_certificate_requests() - .await - .len(), - 1 - ); - - let job_status_requests = env - .rms_sim - .submitted_get_configure_switch_certificate_job_status_requests() - .await; - assert_eq!(job_status_requests.len(), 1); - assert_eq!(job_status_requests[0].job_id, "test-switch-cert-job"); - - env.rms_sim - .queue_get_configure_switch_certificate_job_status_response(Ok( - rms::GetConfigureSwitchCertificateJobStatusResponse { - status: rms::ReturnCode::Success as i32, - job_id: "test-switch-cert-job".to_string(), - state: "completed".to_string(), - ..Default::default() - }, - )) - .await; - let result = env.run_switch_cert_monitor_iteration().await; - assert_eq!(result.observed_endpoints, 1); - assert_eq!(result.successful_probes, 1); - assert_eq!( - result.fingerprint_mismatches, - expected_fingerprint_mismatches + } else { + assert!( + switch.switch_maintenance_requested.is_none(), + "matching certificate should not request switch certificate maintenance" ); - assert_eq!(result.applied_updates, expected_fingerprint_mismatches); - assert_eq!(result.pending_updates, 0); - assert_eq!(result.apply_errors, 0); - assert_eq!( - env.rms_sim - .submitted_configure_switch_certificate_requests() - .await - .len(), - 1 + assert!( + rack.config.maintenance_requested.is_none(), + "matching certificate should not request rack maintenance" ); } + + let rms_requests = env + .rms_sim + .submitted_configure_switch_certificate_requests() + .await; + assert_eq!( + rms_requests.len(), + 0, + "switch cert monitor should queue the rack state machine instead of calling RMS directly" + ); } #[crate::sqlx_test] diff --git a/crates/api-db/src/rack.rs b/crates/api-db/src/rack.rs index c2b3cb043f..8c06a056f1 100644 --- a/crates/api-db/src/rack.rs +++ b/crates/api-db/src/rack.rs @@ -58,6 +58,17 @@ where .map_err(|e| DatabaseError::new(query.sql(), e)) } +/// Lock one active rack until the caller's transaction completes. +pub async fn lock_for_update(txn: &mut PgConnection, rack_id: &RackId) -> DatabaseResult { + let query = "SELECT id FROM racks WHERE id = $1 AND deleted IS NULL FOR UPDATE"; + sqlx::query_as::<_, (RackId,)>(query) + .bind(rack_id) + .fetch_optional(txn) + .await + .map_err(|e| DatabaseError::new(query, e)) + .map(|row| row.is_some()) +} + pub async fn find_ids( txn: impl DbReader<'_>, filter: model::rack::RackSearchFilter, diff --git a/crates/api-model/src/rack.rs b/crates/api-model/src/rack.rs index 55ef4a187b..63825a5880 100644 --- a/crates/api-model/src/rack.rs +++ b/crates/api-model/src/rack.rs @@ -745,7 +745,7 @@ impl std::fmt::Display for MaintenanceActivity { /// Specifies which devices in the rack should be included in an on-demand /// maintenance cycle. When all three device-id lists are empty, the full rack /// is maintained. -#[derive(Debug, Clone, Default, Deserialize, Serialize)] +#[derive(Debug, Clone, Default, PartialEq, Eq, Deserialize, Serialize)] pub struct MaintenanceScope { #[serde(default)] pub machine_ids: Vec, diff --git a/crates/component-manager/Cargo.toml b/crates/component-manager/Cargo.toml index 22bbd6af43..a075dc628f 100644 --- a/crates/component-manager/Cargo.toml +++ b/crates/component-manager/Cargo.toml @@ -47,6 +47,7 @@ tracing = { workspace = true } [dev-dependencies] carbide-api-test-helper = { path = "../api-test-helper" } carbide-macros = { path = "../macros" } +carbide-secrets = { path = "../secrets", features = ["test-support"] } carbide-sqlx-testing = { path = "../sqlx-testing" } carbide-test-support = { path = "../test-support" } toml = { workspace = true } diff --git a/crates/component-manager/src/component_manager.rs b/crates/component-manager/src/component_manager.rs index e8bb0f7a91..2f23280ef1 100644 --- a/crates/component-manager/src/component_manager.rs +++ b/crates/component-manager/src/component_manager.rs @@ -1,11 +1,19 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +use std::collections::HashMap; use std::sync::Arc; +use carbide_rack::firmware_object::rack_maintenance_access_token_key; use carbide_redfish::libredfish::RedfishClientPool; +use carbide_secrets::credentials::{CredentialManager, Credentials}; +use carbide_uuid::rack::RackId; +use carbide_uuid::switch::SwitchId; +use db::{ObjectColumnFilter, WithTransaction}; use librms::RmsApi; +use model::rack::{MaintenanceActivity, MaintenanceScope, RackState}; use model::rack_type::RackProfileConfig; +use model::switch::SwitchMaintenanceOperation; use sqlx::PgPool; use crate::compute_tray_manager::{Backend as ComputeBackend, ComputeTrayManager}; @@ -41,7 +49,309 @@ pub struct ComponentManager { pub compute_tray_use_state_controller: bool, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct SwitchMaintenanceRequestResult { + pub switch_id: SwitchId, + pub error: Option, +} + +/// Which rack states a maintenance caller is willing to schedule from. +/// +/// Automatic maintenance uses [`RequireReady`](Self::RequireReady), while the +/// operator-facing API retains its existing ability to recover racks in +/// `Error` via [`AllowErrorRecovery`](Self::AllowErrorRecovery). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RackMaintenanceEligibility { + RequireReady, + AllowErrorRecovery, +} + +/// The complete, non-error result of attempting to schedule rack maintenance. +/// +/// Keeping contention and eligibility as outcomes (rather than string errors) +/// lets periodic callers retry without marking certificate rotation failed, +/// while operator-facing callers can map the same result to their public API. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RackMaintenanceRequestOutcome { + /// This call persisted the requested scope. + Scheduled, + /// The exact same scope is already pending. This is an idempotent retry. + AlreadyPending, + /// A different maintenance request is pending and was left untouched. + Busy, + /// No request is pending, but the rack's current state is not eligible. + Deferred { state: RackState }, +} + +/// Credential to persist if this call wins the rack-maintenance scheduling +/// race. It is written only after the scheduling transaction commits, so no +/// database lock is held during the external credential-store operation. +pub struct RackMaintenanceAccessToken<'a> { + pub credential_manager: &'a dyn CredentialManager, + pub token: String, +} + +/// Atomically request rack maintenance through the rack state controller. +/// +/// The rack row is locked before inspecting its state and existing scope, and +/// remains locked through the config write. Exact retries are idempotent; +/// unrelated pending work is never overwritten. Existing work is checked +/// before state eligibility so a retry can still observe `AlreadyPending` +/// after the rack controller has started transitioning the rack. +/// +/// This is a free function because the API supports rack maintenance even when +/// no `ComponentManager` backend is configured. +pub async fn request_rack_maintenance_via_state_controller( + db_pool: &PgPool, + rack_id: &RackId, + scope: MaintenanceScope, + eligibility: RackMaintenanceEligibility, + maintenance_access_token: Option>, +) -> Result { + let rack_id = rack_id.clone(); + let scheduled_scope = scope.clone(); + let transaction_rack_id = rack_id.clone(); + + let result = db_pool + .with_txn(|txn| { + Box::pin(async move { + if !db::rack::lock_for_update(txn.as_mut(), &transaction_rack_id) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + { + return Err(ComponentManagerError::NotFound(format!( + "rack {transaction_rack_id} not found" + ))); + } + let rack = db::rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, &transaction_rack_id), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + .into_iter() + .next() + .ok_or_else(|| { + ComponentManagerError::NotFound(format!("rack {transaction_rack_id} not found")) + })?; + + if let Some(existing_scope) = rack.config.maintenance_requested.as_ref() { + return Ok(if existing_scope == &scope { + RackMaintenanceRequestOutcome::AlreadyPending + } else { + RackMaintenanceRequestOutcome::Busy + }); + } + + let state = rack.controller_state.value.clone(); + let eligible = match eligibility { + RackMaintenanceEligibility::RequireReady => state == RackState::Ready, + RackMaintenanceEligibility::AllowErrorRecovery => { + matches!(&state, RackState::Ready | RackState::Error { .. }) + } + }; + if !eligible { + return Ok(RackMaintenanceRequestOutcome::Deferred { state }); + } + + let reset_firmware_upgrade_job = + scope.should_run(&MaintenanceActivity::FirmwareUpgrade { + firmware_version: None, + components: vec![], + force_update: false, + }); + let mut config = rack.config; + config.maintenance_requested = Some(scope); + db::rack::update(txn.as_mut(), &transaction_rack_id, &config) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + // Preserve the operator API's existing behavior, and keep this + // reset atomic with accepting the maintenance request. + if reset_firmware_upgrade_job { + db::rack::update_firmware_upgrade_job(txn.as_mut(), &transaction_rack_id, None) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + } + + Ok(RackMaintenanceRequestOutcome::Scheduled) + }) + }) + .await; + + let outcome = match result { + Ok(Ok(outcome)) => outcome, + Ok(Err(error)) => return Err(error), + Err(error) => return Err(ComponentManagerError::Internal(error.to_string())), + }; + + if outcome == RackMaintenanceRequestOutcome::Scheduled + && let Some(RackMaintenanceAccessToken { + credential_manager, + token, + }) = maintenance_access_token + && let Err(error) = credential_manager + .set_credentials( + &rack_maintenance_access_token_key(&rack_id), + &Credentials::UsernamePassword { + username: "access_token".into(), + password: token, + }, + ) + .await + { + let recovery = + recover_rack_maintenance_after_credential_failure(db_pool, &rack_id, &scheduled_scope) + .await; + return Err(ComponentManagerError::Internal(match recovery { + Ok(recovery) => format!( + "failed to store rack maintenance access token: {error}; recovery: {recovery:?}" + ), + Err(recovery_error) => format!( + "failed to store rack maintenance access token: {error}; failed to recover maintenance request: {recovery_error}" + ), + })); + } + + Ok(outcome) +} + +#[derive(Debug)] +enum RackMaintenanceCredentialRecovery { + Cleared, + RequestChanged, + ExecutionStarted, + RackNotFound, +} + +async fn recover_rack_maintenance_after_credential_failure( + db_pool: &PgPool, + rack_id: &RackId, + scheduled_scope: &MaintenanceScope, +) -> Result { + let rack_id = rack_id.clone(); + let scheduled_scope = scheduled_scope.clone(); + let result = db_pool + .with_txn(|txn| { + Box::pin(async move { + if !db::rack::lock_for_update(txn.as_mut(), &rack_id) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + { + return Ok(RackMaintenanceCredentialRecovery::RackNotFound); + } + let Some(mut rack) = db::rack::find_by( + txn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, &rack_id), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + .into_iter() + .next() + else { + return Ok(RackMaintenanceCredentialRecovery::RackNotFound); + }; + + if rack.config.maintenance_requested.as_ref() != Some(&scheduled_scope) { + return Ok(RackMaintenanceCredentialRecovery::RequestChanged); + } + + let state = rack.controller_state.value.clone(); + if !matches!(&state, RackState::Ready | RackState::Error { .. }) { + // Once execution starts, clearing the scope would make the rack controller + // interpret the default scope as "run all activities". Leave recovery to + // its established missing-credential error path instead. + tracing::warn!( + rack_id = %rack_id, + ?state, + "rack maintenance started before credential storage failed; leaving cleanup to the rack state controller", + ); + return Ok(RackMaintenanceCredentialRecovery::ExecutionStarted); + } + + rack.config.maintenance_requested = None; + db::rack::update(txn.as_mut(), &rack_id, &rack.config) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + Ok(RackMaintenanceCredentialRecovery::Cleared) + }) + }) + .await; + + match result { + Ok(Ok(recovery)) => Ok(recovery), + Ok(Err(error)) => Err(error), + Err(error) => Err(ComponentManagerError::Internal(error.to_string())), + } +} + impl ComponentManager { + pub async fn request_switch_maintenance_via_state_controller( + &self, + db_pool: &PgPool, + switch_ids: &[SwitchId], + operation: SwitchMaintenanceOperation, + initiator: &str, + ) -> Result, ComponentManagerError> { + if !self.nv_switch_use_state_controller { + return Err(ComponentManagerError::InvalidArgument( + "nv_switch_use_state_controller is disabled; switch maintenance through the state controller is unavailable" + .to_string(), + )); + } + + let switch_ids = switch_ids.to_vec(); + let initiator = initiator.to_string(); + db_pool + .with_txn(|txn| { + Box::pin(async move { + let existing = db::switch::find_by( + txn, + db::ObjectColumnFilter::List(db::switch::IdColumn, &switch_ids), + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + let by_id: HashMap = + existing.into_iter().map(|sw| (sw.id, sw)).collect(); + let mut results = Vec::with_capacity(switch_ids.len()); + + for switch_id in &switch_ids { + let Some(switch) = by_id.get(switch_id) else { + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: Some(format!("switch {switch_id} not found")), + }); + continue; + }; + + if switch.is_marked_as_deleted() { + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: Some(format!("switch {switch_id} is marked for deletion")), + }); + continue; + } + + db::switch::set_switch_maintenance_requested( + txn, *switch_id, &initiator, operation, + ) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))?; + + results.push(SwitchMaintenanceRequestResult { + switch_id: *switch_id, + error: None, + }); + } + + Ok(results) + }) + }) + .await + .map_err(|error| ComponentManagerError::Internal(error.to_string()))? + } + pub async fn configure_switch_certificate( &self, endpoint: &SwitchEndpoint, @@ -213,6 +523,15 @@ pub async fn build_component_manager( #[cfg(test)] mod tests { + use async_trait::async_trait; + use carbide_secrets::SecretsError; + use carbide_secrets::credentials::{CredentialKey, CredentialReader, CredentialWriter}; + use carbide_secrets::test_support::credentials::TestCredentialManager; + use carbide_uuid::rack::RackId; + use db::ObjectColumnFilter; + use model::rack::{ + FirmwareUpgradeJob, MaintenanceActivity, MaintenanceScope, RackConfig, RackState, + }; use model::rack_type::{ RackCapabilitiesSet, RackCapabilityCompute, RackCapabilityPowerShelf, RackCapabilitySwitch, RackHardwareTopology, RackProductFamily, RackProfile, @@ -221,6 +540,361 @@ mod tests { use super::*; use crate::config::ComponentManagerConfig; + struct FailingCredentialManager; + + #[async_trait] + impl CredentialReader for FailingCredentialManager { + async fn get_credentials( + &self, + _key: &CredentialKey, + ) -> Result, SecretsError> { + Ok(None) + } + } + + #[async_trait] + impl CredentialWriter for FailingCredentialManager { + async fn set_credentials( + &self, + _key: &CredentialKey, + _credentials: &Credentials, + ) -> Result<(), SecretsError> { + Err(SecretsError::GenericError( + std::io::Error::other("test credential write failure").into(), + )) + } + + async fn create_credentials( + &self, + _key: &CredentialKey, + _credentials: &Credentials, + ) -> Result<(), SecretsError> { + unreachable!("test only exercises set_credentials") + } + + async fn delete_credentials(&self, _key: &CredentialKey) -> Result<(), SecretsError> { + Ok(()) + } + } + + impl CredentialManager for FailingCredentialManager {} + + async fn create_rack_in_state(pool: &PgPool, state: RackState) -> RackId { + let rack_id = RackId::new(uuid::Uuid::new_v4().to_string()); + let mut txn = pool.begin().await.unwrap(); + let rack = db::rack::create(txn.as_mut(), &rack_id, None, &RackConfig::default(), None) + .await + .unwrap(); + if state != RackState::Created { + assert!( + db::rack::try_update_controller_state( + txn.as_mut(), + &rack_id, + rack.controller_state.version, + rack.controller_state.version.increment(), + &state, + ) + .await + .unwrap() + ); + } + txn.commit().await.unwrap(); + rack_id + } + + async fn load_rack(pool: &PgPool, rack_id: &RackId) -> model::rack::Rack { + let mut conn = pool.acquire().await.unwrap(); + db::rack::find_by( + conn.as_mut(), + ObjectColumnFilter::One(db::rack::IdColumn, rack_id), + ) + .await + .unwrap() + .pop() + .unwrap() + } + + fn nmx_scope() -> MaintenanceScope { + MaintenanceScope { + activities: vec![MaintenanceActivity::ConfigureNmxCluster], + ..Default::default() + } + } + + #[carbide_macros::sqlx_test] + async fn rack_maintenance_scheduler_is_atomic_and_idempotent(pool: PgPool) { + let ready_rack = create_rack_in_state(&pool, RackState::Ready).await; + let scope = nmx_scope(); + + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + scope.clone(), + RackMaintenanceEligibility::RequireReady, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + assert_eq!( + load_rack(&pool, &ready_rack) + .await + .config + .maintenance_requested, + Some(scope.clone()), + ); + + // Exact retries are idempotent, including after the rack state has + // moved on. A different request is busy and cannot replace it. + let rack = load_rack(&pool, &ready_rack).await; + let mut txn = pool.begin().await.unwrap(); + assert!( + db::rack::try_update_controller_state( + txn.as_mut(), + &ready_rack, + rack.controller_state.version, + rack.controller_state.version.increment(), + &RackState::Discovering, + ) + .await + .unwrap() + ); + txn.commit().await.unwrap(); + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + scope.clone(), + RackMaintenanceEligibility::RequireReady, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::AlreadyPending, + ); + + let unrelated_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::PowerSequence], + ..Default::default() + }; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &ready_rack, + unrelated_scope, + RackMaintenanceEligibility::RequireReady, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Busy, + ); + assert_eq!( + load_rack(&pool, &ready_rack) + .await + .config + .maintenance_requested, + Some(scope), + ); + + // Automatic callers defer non-Ready racks; operator callers may + // explicitly opt into recovering a rack from Error. + let error_state = RackState::Error { + cause: "test".into(), + }; + let error_rack = create_rack_in_state(&pool, error_state.clone()).await; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &error_rack, + nmx_scope(), + RackMaintenanceEligibility::RequireReady, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Deferred { state: error_state }, + ); + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &error_rack, + nmx_scope(), + RackMaintenanceEligibility::AllowErrorRecovery, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + + // Two different requests racing for an empty slot serialize on the + // rack row: one wins and the other observes Busy. + let race_rack = create_rack_in_state(&pool, RackState::Ready).await; + let first_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"first"}"#.into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let second_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some(r#"{"Id":"second"}"#.into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let credential_manager = TestCredentialManager::default(); + let (first, second) = tokio::join!( + request_rack_maintenance_via_state_controller( + &pool, + &race_rack, + first_scope, + RackMaintenanceEligibility::RequireReady, + Some(RackMaintenanceAccessToken { + credential_manager: &credential_manager, + token: "first-token".into(), + }), + ), + request_rack_maintenance_via_state_controller( + &pool, + &race_rack, + second_scope, + RackMaintenanceEligibility::RequireReady, + Some(RackMaintenanceAccessToken { + credential_manager: &credential_manager, + token: "second-token".into(), + }), + ), + ); + let first = first.unwrap(); + let second = second.unwrap(); + let winning_token = match (&first, &second) { + (RackMaintenanceRequestOutcome::Scheduled, RackMaintenanceRequestOutcome::Busy) => { + "first-token" + } + (RackMaintenanceRequestOutcome::Busy, RackMaintenanceRequestOutcome::Scheduled) => { + "second-token" + } + unexpected => panic!("expected one Scheduled and one Busy outcome, got {unexpected:?}"), + }; + let stored_credentials = credential_manager + .get_credentials(&CredentialKey::RackMaintenanceAccessToken { + rack_id: race_rack.clone(), + }) + .await + .unwrap() + .expect("the winning request should persist its token"); + assert_eq!( + stored_credentials, + Credentials::UsernamePassword { + username: "access_token".into(), + password: winning_token.into(), + } + ); + + // Firmware bookkeeping is cleared in the same transaction that + // accepts a firmware maintenance request. + let firmware_rack = create_rack_in_state(&pool, RackState::Ready).await; + let mut txn = pool.begin().await.unwrap(); + db::rack::update_firmware_upgrade_job( + txn.as_mut(), + &firmware_rack, + Some(&FirmwareUpgradeJob { + job_id: Some("stale-job".into()), + ..Default::default() + }), + ) + .await + .unwrap(); + txn.commit().await.unwrap(); + let firmware_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some("{}".into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + assert_eq!( + request_rack_maintenance_via_state_controller( + &pool, + &firmware_rack, + firmware_scope, + RackMaintenanceEligibility::RequireReady, + None, + ) + .await + .unwrap(), + RackMaintenanceRequestOutcome::Scheduled, + ); + assert!( + load_rack(&pool, &firmware_rack) + .await + .firmware_upgrade_job + .is_none() + ); + + // Soft-deleted racks cannot accept new maintenance work. + let deleted_rack = create_rack_in_state(&pool, RackState::Ready).await; + let mut txn = pool.begin().await.unwrap(); + db::rack::mark_as_deleted(&deleted_rack, txn.as_mut()) + .await + .unwrap(); + txn.commit().await.unwrap(); + assert!(matches!( + request_rack_maintenance_via_state_controller( + &pool, + &deleted_rack, + nmx_scope(), + RackMaintenanceEligibility::RequireReady, + None, + ) + .await, + Err(ComponentManagerError::NotFound(_)), + )); + + // Credential storage happens after commit. If it fails before the rack controller + // starts the request, the compensating transaction removes that exact request. + let credential_failure_rack = create_rack_in_state(&pool, RackState::Ready).await; + let credential_failure_scope = MaintenanceScope { + activities: vec![MaintenanceActivity::FirmwareUpgrade { + firmware_version: Some("{}".into()), + components: vec![], + force_update: false, + }], + ..Default::default() + }; + let error = request_rack_maintenance_via_state_controller( + &pool, + &credential_failure_rack, + credential_failure_scope, + RackMaintenanceEligibility::RequireReady, + Some(RackMaintenanceAccessToken { + credential_manager: &FailingCredentialManager, + token: "test-token".into(), + }), + ) + .await + .unwrap_err(); + assert!( + error.to_string().contains("recovery: Cleared"), + "unexpected error: {error}" + ); + assert!( + load_rack(&pool, &credential_failure_rack) + .await + .config + .maintenance_requested + .is_none() + ); + } + fn rms_rack_profiles(profile: RackProfile) -> RackProfileConfig { RackProfileConfig { rack_profiles: [("NVL72".to_string(), profile)].into_iter().collect(), diff --git a/crates/nvlink-manager/Cargo.toml b/crates/nvlink-manager/Cargo.toml index 98cb8ed3e1..8ab69ef1df 100644 --- a/crates/nvlink-manager/Cargo.toml +++ b/crates/nvlink-manager/Cargo.toml @@ -21,6 +21,7 @@ carbide-secrets = { path = "../secrets" } carbide-utils = { path = "../utils" } carbide-uuid = { path = "../uuid" } config-version = { path = "../config-version" } +component-manager = { path = "../component-manager" } libnmxc = { path = "../libnmxc" } libnmxm = { path = "../libnmxm" } diff --git a/crates/nvlink-manager/src/lib.rs b/crates/nvlink-manager/src/lib.rs index 5e677c6bcd..80800d6be1 100644 --- a/crates/nvlink-manager/src/lib.rs +++ b/crates/nvlink-manager/src/lib.rs @@ -27,12 +27,12 @@ use std::io; use std::sync::Arc; use std::time::Duration; -use carbide_secrets::credentials::CredentialManager; use carbide_utils::periodic_timer::PeriodicTimer; use carbide_uuid::machine::MachineId; use carbide_uuid::nvlink::{NvLinkDomainId, NvLinkLogicalPartitionId, NvLinkPartitionId}; use carbide_uuid::rack::RackId; use chrono::Utc; +use component_manager::component_manager::ComponentManager; use config::NvLinkConfig; use config_version::Versioned; use db::machine::find_machine_ids; @@ -59,7 +59,6 @@ use model::machine::nvlink::{MachineNvLinkGpuStatusObservation, MachineNvLinkSta use model::machine::{HostHealthConfig, LoadSnapshotOptions, ManagedHostStateSnapshot}; use model::nvl_logical_partition::LogicalPartition; use model::nvl_partition::{NvlPartition, NvlPartitionName}; -use model::rack_type::RackProfileConfig; use sqlx::PgPool; #[cfg(feature = "test-support")] pub use switch_cert_monitor::{SwitchCertificateMonitor, SwitchCertificateMonitorIterationResult}; @@ -900,9 +899,7 @@ pub struct NvLinkManager { meter: opentelemetry::metrics::Meter, config: NvLinkConfig, host_health: HostHealthConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, + component_manager: Option>, work_lock_manager_handle: WorkLockManagerHandle, } @@ -912,9 +909,7 @@ pub struct NvLinkManagerArgs { pub meter: opentelemetry::metrics::Meter, pub config: NvLinkConfig, pub host_health: HostHealthConfig, - pub rms_client: Option>, - pub credential_manager: Arc, - pub rack_profiles: RackProfileConfig, + pub component_manager: Option>, pub work_lock_manager_handle: WorkLockManagerHandle, } @@ -926,9 +921,7 @@ impl NvLinkManager { meter: args.meter, config: args.config, host_health: args.host_health, - rms_client: args.rms_client, - credential_manager: args.credential_manager, - rack_profiles: args.rack_profiles, + component_manager: args.component_manager, work_lock_manager_handle: args.work_lock_manager_handle, } } @@ -953,9 +946,7 @@ impl NvLinkManager { self.db_pool, self.meter, self.config, - self.rms_client, - self.credential_manager, - self.rack_profiles, + self.component_manager, self.work_lock_manager_handle, ); join_set diff --git a/crates/nvlink-manager/src/switch_cert_monitor.rs b/crates/nvlink-manager/src/switch_cert_monitor.rs index 02f6a4c8b9..b14ec5ff47 100644 --- a/crates/nvlink-manager/src/switch_cert_monitor.rs +++ b/crates/nvlink-manager/src/switch_cert_monitor.rs @@ -20,18 +20,18 @@ use std::sync::Arc; use std::time::Duration; use std::{fmt, io}; -use carbide_rack::firmware_update::{build_new_node_info, load_switch_firmware_device_info}; -use carbide_rack::rms_node_type::switch_node_type_for_profile; -use carbide_secrets::credentials::CredentialManager; use carbide_utils::metrics::SharedMetricsHolder; use carbide_utils::periodic_timer::PeriodicTimer; -use carbide_uuid::rack::{RackId, RackProfileId}; +use carbide_uuid::rack::RackId; use carbide_uuid::switch::SwitchId; use chrono::Utc; +use component_manager::component_manager::{ + ComponentManager, RackMaintenanceEligibility, RackMaintenanceRequestOutcome, + request_rack_maintenance_via_state_controller, +}; use db::db_read::PgPoolReader; use db::work_lock_manager::WorkLockManagerHandle; -use librms::protos::rack_manager as rms; -use model::rack_type::RackProfileConfig; +use model::rack::{MaintenanceActivity, MaintenanceScope}; use opentelemetry::KeyValue; use opentelemetry::metrics::{Histogram, Meter}; use rustls::{ClientConfig, RootCertStore}; @@ -39,7 +39,6 @@ use rustls_pki_types::{CertificateDer, PrivateKeyDer, ServerName}; use sha2::{Digest, Sha256}; use sqlx::PgPool; use tokio::net::TcpStream; -use tokio::sync::Mutex; use tokio_rustls::TlsConnector; use tokio_util::sync::CancellationToken; use tracing::Instrument; @@ -59,7 +58,6 @@ struct CertificateInfo { struct SwitchCertificateMonitorTarget { switch_id: SwitchId, rack_id: RackId, - rack_profile_id: Option, endpoint_url: String, } @@ -67,7 +65,6 @@ struct SwitchCertificateMonitorTarget { enum SwitchCertApplyStatus { NotNeeded, Pending, - Applied, Error, Skipped, } @@ -77,7 +74,6 @@ impl SwitchCertApplyStatus { match self { Self::NotNeeded => "not_needed", Self::Pending => "pending", - Self::Applied => "applied", Self::Error => "error", Self::Skipped => "skipped", } @@ -160,11 +156,6 @@ impl fmt::Display for SwitchCertMonitorMetrics { .iter() .filter(|cert| !cert.desired_cert_error.is_empty()) .count(); - let applied_updates = self - .observed_certs - .iter() - .filter(|cert| cert.apply_status == SwitchCertApplyStatus::Applied) - .count(); let pending_updates = self .observed_certs .iter() @@ -172,12 +163,11 @@ impl fmt::Display for SwitchCertMonitorMetrics { .count(); write!( f, - "{{ observed_endpoints: {}, desired_cert_errors: {}, successful_probes: {}, matching_fingerprints: {}, applied_updates: {}, pending_updates: {}, duration: {} }}", + "{{ observed_endpoints: {}, desired_cert_errors: {}, successful_probes: {}, matching_fingerprints: {}, pending_updates: {}, duration: {} }}", self.observed_certs.len(), desired_cert_errors, successful_probes, matching_fingerprints, - applied_updates, pending_updates, self.recording_started_at.elapsed().as_millis(), ) @@ -463,39 +453,9 @@ impl MetricHolder { pub struct SwitchCertificateMonitor { db_pool: PgPool, config: NvLinkConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, + component_manager: Option>, metric_holder: Arc, work_lock_manager_handle: WorkLockManagerHandle, - in_flight_certificate_jobs: Mutex>, -} - -#[derive(Clone, Debug)] -struct InFlightSwitchCertJob { - job_id: String, -} - -#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)] -struct SwitchCertJobKey { - switch_id: String, - rack_id: String, -} - -impl SwitchCertJobKey { - fn from_target(target: &SwitchCertificateMonitorTarget) -> Self { - Self { - switch_id: target.switch_id.to_string(), - rack_id: target.rack_id.to_string(), - } - } -} - -#[derive(Clone, Debug, PartialEq, Eq)] -enum RmsSwitchCertJobState { - Pending(String), - Completed, - Failed(String), } #[derive(Debug, Clone, PartialEq, Eq)] @@ -534,11 +494,7 @@ impl SwitchCertificateMonitorIterationResult { .iter() .filter(|cert| !cert.error.is_empty()) .count(), - applied_updates: metrics - .observed_certs - .iter() - .filter(|cert| cert.apply_status == SwitchCertApplyStatus::Applied) - .count(), + applied_updates: 0, pending_updates: metrics .observed_certs .iter() @@ -562,9 +518,7 @@ impl SwitchCertificateMonitor { db_pool: PgPool, meter: Meter, config: NvLinkConfig, - rms_client: Option>, - credential_manager: Arc, - rack_profiles: RackProfileConfig, + component_manager: Option>, work_lock_manager_handle: WorkLockManagerHandle, ) -> Self { let hold_period = config @@ -575,12 +529,9 @@ impl SwitchCertificateMonitor { Self { db_pool, config, - rms_client, - credential_manager, - rack_profiles, + component_manager, metric_holder, work_lock_manager_handle, - in_flight_certificate_jobs: Mutex::new(BTreeMap::new()), } } @@ -769,7 +720,7 @@ impl SwitchCertificateMonitor { rack_id = %rack_id_label, endpoint = %target.endpoint_url, error = %error, - "Failed to request RMS NMX-C switch certificate configuration" + "Failed to request NMX-C cluster configuration via rack state machine" ); (SwitchCertApplyStatus::Error, error) } @@ -845,59 +796,8 @@ impl SwitchCertificateMonitor { target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, ) -> Result { - let key = SwitchCertJobKey::from_target(target); - let in_flight_job = self - .in_flight_certificate_jobs - .lock() + self.request_nmx_cluster_configuration_with_rack_state_controller(target, cancel_token) .await - .get(&key) - .cloned(); - - if let Some(in_flight_job) = in_flight_job { - let job_state = self - .get_in_flight_certificate_job_state(&in_flight_job.job_id, cancel_token) - .await?; - match job_state { - RmsSwitchCertJobState::Pending(state) => { - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - endpoint = %target.endpoint_url, - job_id = %in_flight_job.job_id, - state = %state, - "RMS NMX-C switch certificate configuration job is still in progress" - ); - Ok(SwitchCertApplyStatus::Pending) - } - RmsSwitchCertJobState::Completed => { - self.in_flight_certificate_jobs.lock().await.remove(&key); - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - endpoint = %target.endpoint_url, - job_id = %in_flight_job.job_id, - "RMS NMX-C switch certificate configuration job completed" - ); - Ok(SwitchCertApplyStatus::Applied) - } - RmsSwitchCertJobState::Failed(error) => { - self.in_flight_certificate_jobs.lock().await.remove(&key); - Err(format!( - "RMS NMX-C switch certificate configuration job {} failed: {}", - in_flight_job.job_id, error - )) - } - } - } else { - let job_id = self - .apply_desired_certificate_with_rms(target, cancel_token) - .await?; - self.in_flight_certificate_jobs - .lock() - .await - .insert(key, InFlightSwitchCertJob { job_id }); - Ok(SwitchCertApplyStatus::Pending) - } } async fn load_switch_certificate_monitor_targets( @@ -914,7 +814,6 @@ impl SwitchCertificateMonitor { .map(|row| SwitchCertificateMonitorTarget { switch_id: row.switch_id, rack_id: row.rack_id, - rack_profile_id: row.rack_profile_id, endpoint_url: nmx_c_endpoint::nmx_c_endpoint_url_from_nvos_ip( &row.nvos_ip, None, @@ -924,171 +823,79 @@ impl SwitchCertificateMonitor { .collect()) } - async fn apply_desired_certificate_with_rms( + async fn request_nmx_cluster_configuration_with_rack_state_controller( &self, target: &SwitchCertificateMonitorTarget, cancel_token: &CancellationToken, - ) -> Result { - let rms_client = self.rms_client.as_ref().ok_or_else(|| { - "RMS client is not configured, so NMX-C switch certificate cannot be applied" - .to_string() - })?; - - let rack_profile_id = target.rack_profile_id.as_ref().ok_or_else(|| { - format!( - "rack {} has no rack_profile_id, so RMS switch node type and topology cannot be resolved", - target.rack_id - ) - })?; - let profile = self - .rack_profiles - .get(rack_profile_id.as_ref()) - .ok_or_else(|| { - format!( - "rack profile {} is not configured, so RMS switch node type and topology cannot be resolved", - rack_profile_id - ) - })?; - let switch_node_type = switch_node_type_for_profile(profile) - .map_err(|error| format!("failed to resolve RMS switch node type: {error}"))?; + ) -> Result { + if self.component_manager.is_none() { + return Err( + "component manager is not configured; cannot request NMX-C cluster configuration" + .to_string(), + ); + } - let switch = tokio::select! { - _ = cancel_token.cancelled() => { - return Err(Self::APPLY_CANCELLED_ERROR.to_string()); - } - switch = load_switch_firmware_device_info( - &self.db_pool, - self.credential_manager.as_ref(), - &target.switch_id, - ) => switch - .map_err(|error| { - format!( - "failed to load switch endpoint info for RMS certificate apply: {error}" - ) - })?, + // Empty device lists intentionally select the full rack. ConfigureNmxCluster runs the + // same certificate and fabric-manager workflow as `rack maintenance start + // --activities configure-nmx-cluster`. + let scope = MaintenanceScope { + activities: vec![MaintenanceActivity::ConfigureNmxCluster], + ..Default::default() }; - - validate_switch_for_rms_certificate_apply(&switch)?; - - let response = tokio::select! { + let outcome = tokio::select! { _ = cancel_token.cancelled() => { return Err(Self::APPLY_CANCELLED_ERROR.to_string()); } - response = rms_client.configure_switch_certificate( - rms::ConfigureSwitchCertificateRequest { - nodes: Some(rms::NodeSet { - nodes: vec![build_new_node_info( - &target.rack_id, - &switch, - switch_node_type, - )], - }), - services: vec![rms::SwitchService::ScaleUpFabricManager as i32], - test_hello: true, - domain: Some(target.rack_id.to_string()), - } - ) => response.map_err(|error| { - format!("RMS ConfigureSwitchCertificate failed: {error}") + outcome = request_rack_maintenance_via_state_controller( + &self.db_pool, + &target.rack_id, + scope, + RackMaintenanceEligibility::RequireReady, + None, + ) => outcome.map_err(|error| { + format!( + "component manager failed to request NMX-C cluster configuration: {error}" + ) })?, }; - let batch_response = response.response.ok_or_else(|| { - "RMS ConfigureSwitchCertificate response did not include a batch response".to_string() - })?; - if batch_response.status != rms::ReturnCode::Success as i32 { - let message = if batch_response.message.trim().is_empty() { - "no error details provided".to_string() - } else { - batch_response.message - }; - return Err(format!( - "RMS ConfigureSwitchCertificate returned status {}: {}", - batch_response.status, message - )); - } - - let switch_id = target.switch_id.to_string(); - let child_job_id = response - .jobs - .iter() - .find(|job| job.node_id == switch_id) - .map(|job| job.job_id.trim()) - .filter(|job_id| !job_id.is_empty()); - let job_id = child_job_id - .or_else(|| { - let parent_job_id = batch_response.job_id.trim(); - if parent_job_id.is_empty() { - None - } else { - Some(parent_job_id) - } - }) - .ok_or_else(|| { - "RMS ConfigureSwitchCertificate response did not include a job id".to_string() - })? - .to_string(); - - tracing::info!( - switch_id = %target.switch_id, - rack_id = %target.rack_id, - job_id = %job_id, - "Submitted RMS switch certificate configuration" - ); - - Ok(job_id) - } - - async fn get_in_flight_certificate_job_state( - &self, - job_id: &str, - cancel_token: &CancellationToken, - ) -> Result { - let rms_client = self.rms_client.as_ref().ok_or_else(|| { - "RMS client is not configured, so NMX-C switch certificate job status cannot be checked" - .to_string() - })?; - - let response = tokio::select! { - _ = cancel_token.cancelled() => { - return Err(Self::APPLY_CANCELLED_ERROR.to_string()); + match outcome { + RackMaintenanceRequestOutcome::Scheduled => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Requested full-rack NMX-C cluster configuration via component manager" + ); + Ok(SwitchCertApplyStatus::Pending) } - response = rms_client.get_configure_switch_certificate_job_status( - rms::GetConfigureSwitchCertificateJobStatusRequest { - job_id: job_id.to_string(), - }, - ) => response.map_err(|error| { - format!("RMS GetConfigureSwitchCertificateJobStatus failed: {error}") - })?, - }; - - if response.status != rms::ReturnCode::Success as i32 { - return Ok(RmsSwitchCertJobState::Failed(format!( - "RMS GetConfigureSwitchCertificateJobStatus returned status {}: {}", - response.status, - non_empty_or(response.error_message.as_str(), response.message.as_str()) - ))); - } - - let state = response.state.trim().to_ascii_lowercase(); - match state.as_str() { - "completed" | "complete" | "succeeded" | "success" => { - Ok(RmsSwitchCertJobState::Completed) + RackMaintenanceRequestOutcome::AlreadyPending => { + tracing::debug!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Full-rack NMX-C cluster configuration is already pending" + ); + Ok(SwitchCertApplyStatus::Pending) } - "failed" | "failure" | "error" => Ok(RmsSwitchCertJobState::Failed(non_empty_or( - response.error_message.as_str(), - response.message.as_str(), - ))), - "queued" | "running" | "pending" | "active" | "in_progress" => { - Ok(RmsSwitchCertJobState::Pending(state)) + RackMaintenanceRequestOutcome::Busy => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + "Deferring NMX-C certificate rotation because different rack maintenance is pending" + ); + Ok(SwitchCertApplyStatus::Skipped) } - "" => Ok(RmsSwitchCertJobState::Pending("unknown".to_string())), - _ => { - tracing::warn!( - job_id = %job_id, - state = %response.state, - "RMS returned unknown NMX-C switch certificate job state; treating as pending" + RackMaintenanceRequestOutcome::Deferred { state } => { + tracing::info!( + switch_id = %target.switch_id, + rack_id = %target.rack_id, + endpoint = %target.endpoint_url, + ?state, + "Deferring NMX-C certificate rotation until the rack is ready" ); - Ok(RmsSwitchCertJobState::Pending(response.state)) + Ok(SwitchCertApplyStatus::Skipped) } } } @@ -1183,27 +990,6 @@ fn desired_server_cert_path(config: &NvLinkConfig, rack_id: &RackId) -> Result Result<(), String> { - if switch.os_ip.as_deref().unwrap_or_default().is_empty() { - return Err(format!( - "switch {} is missing an NVOS IP address for RMS certificate apply", - switch.node_id - )); - } - if switch.os_username.as_deref().unwrap_or_default().is_empty() - || switch.os_password.as_deref().unwrap_or_default().is_empty() - { - return Err(format!( - "switch {} is missing NVOS credentials for RMS certificate apply", - switch.node_id - )); - } - - Ok(()) -} - async fn build_tls_client_config(config: &NvLinkConfig) -> Result { let mut roots = RootCertStore::empty(); let ca_cert_path = config @@ -1295,14 +1081,6 @@ fn cert_expires_within(cert: &CertificateInfo, window: Duration) -> bool { not_after <= warning_threshold } -fn non_empty_or(primary: &str, fallback: &str) -> String { - if primary.trim().is_empty() { - fallback.trim().to_string() - } else { - primary.trim().to_string() - } -} - fn metric_attrs(base_attrs: &[KeyValue], extra_attrs: &[KeyValue]) -> Vec { [base_attrs, extra_attrs].concat() }