diff --git a/API_CHANGELOG.md b/API_CHANGELOG.md index 927d1aa1..8a5c2af7 100644 --- a/API_CHANGELOG.md +++ b/API_CHANGELOG.md @@ -8,6 +8,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Added +- **Import existing host VMs (admin)** — new admin endpoints to adopt VMs that exist on a host but aren't tracked in the database (issue #166). + - `GET /api/admin/v1/hosts/{id}/vms/unmanaged` lists VMs present on the host with no matching database record (returns host vmid, mapped database id, name, CPU/memory/disk specs, storage, MAC, running state). The admin service dispatches a discovery job to the worker and reads the reply over a temporary Redis pub/sub channel. + - `POST /api/admin/v1/hosts/{id}/vms/import` (`{ host_vm_id, user_id, reason? }`) imports one VM: it is assigned to `user_id` and billed via the region's **custom pricing** (required — import fails if the region has none), capturing the VM's current CPU/memory/disk specs into a custom template. Currently supports Proxmox hosts. Returns a `job_id`. + - **Per-request OAuth return URL** — `GET /api/v1/oauth/{provider}/login` now accepts an optional `?redirect=` to override the configured `success-redirect` for that login only (e.g. `http://localhost:3000/oauth/complete` in dev). The URL is validated against a new `allowed-redirects` allowlist (`localhost` host always allowed; `success-redirect` always implicitly allowed) and rejected with `400` otherwise, then round-tripped through the signed `state`. See the OAuth login flow in API_DOCUMENTATION.md for details and a `config.yaml` example. - **2026-07-18** - Passwordless WebAuthn / passkey login diff --git a/Cargo.lock b/Cargo.lock index 7777eabc..b790056c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2930,6 +2930,7 @@ dependencies = [ "serde_json", "tokio", "tower-http", + "uuid", ] [[package]] diff --git a/lnvps_api/src/host/dummy_host.rs b/lnvps_api/src/host/dummy_host.rs index 47011d32..04a92099 100644 --- a/lnvps_api/src/host/dummy_host.rs +++ b/lnvps_api/src/host/dummy_host.rs @@ -5,7 +5,7 @@ use crate::host::{ use async_trait::async_trait; use chrono::Utc; use lnvps_api_common::retry::OpResult; -use lnvps_api_common::{GB, PB, TB, VmRunningState, VmRunningStates, op_fatal}; +use lnvps_api_common::{GB, HostVmSpec, PB, TB, VmRunningState, VmRunningStates, op_fatal}; use lnvps_db::{Vm, VmOsImage}; use serde::{Deserialize, Serialize}; use std::collections::HashMap; @@ -152,10 +152,27 @@ impl DummyVmHost { let _ = std::fs::write(STATE_FILE, json); } } + + /// Set the list of host VMs reported by [`list_host_vms`]. + /// + /// Backed by a process-wide registry so the value survives the fresh + /// [`DummyVmHost`] instances that `get_host_client` constructs. Intended for + /// exercising VM import/discovery flows. + pub async fn set_host_vms(vms: Vec) { + *DUMMY_HOST_VMS.lock().await = vms; + } } +/// Process-wide registry of "host" VMs reported by [`DummyVmHost::list_host_vms`]. +static DUMMY_HOST_VMS: LazyLock>>> = + LazyLock::new(|| Arc::new(Mutex::new(Vec::new()))); + #[async_trait] impl VmHostClient for DummyVmHost { + async fn list_host_vms(&self) -> OpResult> { + Ok(DUMMY_HOST_VMS.lock().await.clone()) + } + async fn get_info(&self) -> OpResult { Ok(VmHostInfo { cpu: 100, diff --git a/lnvps_api/src/host/mod.rs b/lnvps_api/src/host/mod.rs index e4469329..2fd4760c 100644 --- a/lnvps_api/src/host/mod.rs +++ b/lnvps_api/src/host/mod.rs @@ -2,6 +2,7 @@ use crate::settings::ProvisionerConfig; use anyhow::{Result, anyhow, bail}; use async_trait::async_trait; use futures::future::join_all; +use lnvps_api_common::HostVmSpec; use lnvps_api_common::VmRunningState; use lnvps_api_common::retry::OpResult; use lnvps_db::{ @@ -30,6 +31,18 @@ pub struct TerminalStream { pub trait VmHostClient: Send + Sync { async fn get_info(&self) -> OpResult; + /// List all VMs present on the host, described in host-native terms. + /// + /// Used for discovering/importing VMs that exist on the host but aren't + /// tracked in the database. Defaults to unsupported for hosts that don't + /// implement discovery. + async fn list_host_vms(&self) -> OpResult> { + use lnvps_api_common::retry::OpError; + Err(OpError::Fatal(anyhow!( + "VM discovery is not supported on this host type" + ))) + } + /// Download OS image to the host async fn download_os_image(&self, image: &VmOsImage) -> OpResult<()>; diff --git a/lnvps_api/src/host/proxmox.rs b/lnvps_api/src/host/proxmox.rs index da15dde2..81409f24 100644 --- a/lnvps_api/src/host/proxmox.rs +++ b/lnvps_api/src/host/proxmox.rs @@ -8,6 +8,7 @@ use anyhow::Result; use async_trait::async_trait; use chrono::Utc; use ipnetwork::IpNetwork; +use lnvps_api_common::HostVmSpec; use lnvps_api_common::JsonApi; use lnvps_api_common::retry::{OpError, OpResult, Pipeline, RetryPolicy}; use lnvps_api_common::{VmRunningState, VmRunningStates, op_fatal, parse_gateway}; @@ -1279,6 +1280,51 @@ impl VmHostClient for ProxmoxClient { } } + async fn list_host_vms(&self) -> OpResult> { + let vms = self.list_vms(&self.node).await?; + let mut out = Vec::with_capacity(vms.len()); + for vm in vms { + // Map to the LNVPS db id (vmid = db_id + 100). VMs with vmid < 100 + // fall outside the managed range and can't be imported. + let mapped_vm_id = if vm.vm_id >= 100 { + let id: ProxmoxVmId = vm.vm_id.into(); + Some(id.inner()) + } else { + None + }; + + // Pull the live config for MAC + backing storage; tolerate failures + // so a single unreadable VM doesn't abort discovery. + let (mac_address, disk_storage) = + match self.get_vm_config(&self.node, vm.vm_id.into()).await { + Ok(cfg) => ( + cfg.config.net.as_deref().and_then(parse_mac_from_net), + cfg.config + .scsi_0 + .as_deref() + .and_then(parse_storage_from_disk), + ), + Err(e) => { + warn!("Failed to read config for vm {}: {}", vm.vm_id, e); + (None, None) + } + }; + + out.push(HostVmSpec { + host_vm_id: vm.vm_id as i64, + mapped_vm_id, + name: vm.name.clone(), + cpu: vm.cpus.unwrap_or(0), + memory: vm.max_mem.unwrap_or(0), + disk_size: vm.max_disk.unwrap_or(0), + disk_storage, + mac_address, + running: matches!(vm.status, VmStatus::Running), + }); + } + Ok(out) + } + async fn download_os_image(&self, image: &VmOsImage) -> OpResult<()> { let iso_storage = self.get_iso_storage(&self.node).await?; let files = self.list_storage_files(&self.node, &iso_storage).await?; @@ -1973,6 +2019,39 @@ impl ProxmoxClient { #[derive(Debug, Copy, Clone, Default)] pub struct ProxmoxVmId(u64); +impl ProxmoxVmId { + /// The underlying LNVPS database id (host vmid minus the +100 offset). + pub fn inner(&self) -> u64 { + self.0 + } +} + +/// Extract the MAC address from a Proxmox `netN` config string, e.g. +/// `virtio=BC:24:11:00:11:22,bridge=vmbr0,firewall=1` -> `BC:24:11:00:11:22`. +fn parse_mac_from_net(net: &str) -> Option { + net.split(',').find_map(|kv| { + let (k, v) = kv.split_once('=')?; + // The NIC model key (virtio/e1000/...) holds the MAC as its value. + if v.split(':').count() == 6 && k != "bridge" { + Some(v.to_string()) + } else { + None + } + }) +} + +/// Extract the storage pool from a Proxmox disk config string, e.g. +/// `local-lvm:vm-1566-disk-0,size=32G` -> `local-lvm`. +fn parse_storage_from_disk(disk: &str) -> Option { + let first = disk.split(',').next()?; + let (storage, _) = first.split_once(':')?; + if storage.is_empty() { + None + } else { + Some(storage.to_string()) + } +} + impl From for i32 { fn from(val: ProxmoxVmId) -> Self { val.0 as i32 + 100 @@ -2865,6 +2944,46 @@ mod tests { assert!(!json.contains("proto"), "json: {json}"); } + #[test] + fn test_parse_mac_from_net() { + assert_eq!( + parse_mac_from_net("virtio=BC:24:11:00:11:22,bridge=vmbr0,firewall=1").as_deref(), + Some("BC:24:11:00:11:22") + ); + assert_eq!( + parse_mac_from_net("e1000=00:15:5D:01:02:03,bridge=vmbr1").as_deref(), + Some("00:15:5D:01:02:03") + ); + // No MAC present + assert_eq!(parse_mac_from_net("bridge=vmbr0,firewall=1"), None); + assert_eq!(parse_mac_from_net(""), None); + } + + #[test] + fn test_parse_storage_from_disk() { + assert_eq!( + parse_storage_from_disk("local-lvm:vm-1566-disk-0,size=32G").as_deref(), + Some("local-lvm") + ); + assert_eq!( + parse_storage_from_disk("ceph:vm-100-disk-0").as_deref(), + Some("ceph") + ); + // No storage separator + assert_eq!(parse_storage_from_disk("none"), None); + assert_eq!(parse_storage_from_disk(""), None); + } + + #[test] + fn test_proxmox_vm_id_inner_maps_to_db_id() { + // Host vmid 1566 -> db id 1466 + let id: ProxmoxVmId = 1566i32.into(); + assert_eq!(id.inner(), 1466); + // Round trip back to host vmid + let host_id: i32 = id.into(); + assert_eq!(host_id, 1566); + } + #[test] fn test_to_pve_firewall_rule_reject_action() { let rule = lnvps_db::VmFirewallRule { diff --git a/lnvps_api/src/provisioner/vm.rs b/lnvps_api/src/provisioner/vm.rs index cc638333..aeba9fe2 100644 --- a/lnvps_api/src/provisioner/vm.rs +++ b/lnvps_api/src/provisioner/vm.rs @@ -2,7 +2,7 @@ use crate::host::{FullVmInfo, VmHostClient, get_host_client}; use crate::provisioner::VmNetworkProvisioner; use crate::router::{ArpEntry, Router, get_router}; use crate::settings::{ProvisionerConfig, Settings}; -use anyhow::{Context, Result, bail, ensure}; +use anyhow::{Context, Result, anyhow, bail, ensure}; use chrono::Utc; use ipnetwork::IpNetwork; use isocountry::CountryCode; @@ -328,6 +328,180 @@ impl VmProvisioner { Ok(new_vm) } + /// Import a VM that already exists on a host but isn't tracked in the + /// database (issue #166). + /// + /// The VM is assigned to `user_id` and billed via the region's custom + /// pricing (which is required — import fails if the region has none). The + /// VM's current specs (CPU/memory/disk) are captured into a custom template. + /// No changes are made on the host; this only creates the database records. + /// + /// The database VM id is fixed by the host's id mapping (e.g. Proxmox + /// `vmid = db_id + 100`) so that all subsequent lifecycle operations target + /// the correct host VM. + pub async fn import_vm(&self, host_id: u64, host_vm_id: i64, user_id: u64) -> Result { + let user = self.db.get_user(user_id).await?; + let host = self.db.get_host(host_id).await?; + + // Discover the VM on the host + let client = get_host_client(&host, &self.provisioner_config)?; + let spec = client + .list_host_vms() + .await? + .into_iter() + .find(|v| v.host_vm_id == host_vm_id) + .ok_or_else(|| anyhow!("VM {} not found on host {}", host_vm_id, host_id))?; + + // The host VM must map to a managed database id + let vm_id = spec.mapped_vm_id.ok_or_else(|| { + anyhow!( + "Host VM {} is outside the managed id range and cannot be imported", + host_vm_id + ) + })?; + + // Reject if a live VM already occupies this id + if let Ok(existing) = self.db.get_vm(vm_id).await + && !existing.deleted + { + bail!("VM {} is already tracked in the database", vm_id); + } + + // Region custom pricing is required for import + let pricing = self + .db + .list_custom_pricing(host.region_id) + .await? + .into_iter() + .find(|p| p.enabled && p.expires.map(|e| e > Utc::now()).unwrap_or(true)) + .ok_or_else(|| { + anyhow!( + "No enabled custom pricing for region {}, cannot import VM", + host.region_id + ) + })?; + + // Pick the host disk backing this VM (match on the storage pool name, + // else fall back to the first enabled/available disk). + let host_disks = self.db.list_host_disks(host.id).await?; + let disk = spec + .disk_storage + .as_ref() + .and_then(|store| host_disks.iter().find(|d| &d.name == store)) + .or_else(|| host_disks.iter().find(|d| d.enabled)) + .or_else(|| host_disks.first()) + .ok_or_else(|| anyhow!("No host disk found for host {}", host.id))? + .clone(); + + // Placeholder OS image (image_id is a required FK). The imported VM + // keeps its real disk; the image is only cosmetic / used on reinstall. + let image = self + .db + .list_os_image() + .await? + .into_iter() + .find(|i| i.enabled) + .ok_or_else(|| anyhow!("No OS image available to use as import placeholder"))?; + + // Capture the current specs into a custom template + let template = VmCustomTemplate { + id: 0, + cpu: spec.cpu, + memory: spec.memory, + disk_size: spec.disk_size, + disk_type: disk.kind, + disk_interface: disk.interface, + pricing_id: pricing.id, + cpu_mfg: pricing.cpu_mfg, + cpu_arch: pricing.cpu_arch, + cpu_features: pricing.cpu_features.clone(), + disk_iops_read: pricing.disk_iops_read, + disk_iops_write: pricing.disk_iops_write, + disk_mbps_read: pricing.disk_mbps_read, + disk_mbps_write: pricing.disk_mbps_write, + network_mbps: pricing.network_mbps, + cpu_limit: pricing.cpu_limit, + firewall_rule_limit: None, + }; + let template_id = self.db.insert_custom_vm_template(&template).await?; + let region = self.db.get_host_region(host.region_id).await?; + + // Create a subscription for this imported VM (mirrors provision_custom) + let subscription = Subscription { + id: 0, + user_id: user.id, + company_id: region.company_id, + name: "Imported VM subscription".to_string(), + description: None, + created: Utc::now(), + expires: None, + is_active: false, + is_setup: false, + currency: pricing.currency.clone(), + interval_amount: 1, + interval_type: IntervalType::Month, + setup_fee: 0, + auto_renewal_enabled: true, + external_id: None, + }; + let line_item = SubscriptionLineItem { + id: 0, + subscription_id: 0, + subscription_type: SubscriptionType::Vps, + name: pricing.name.clone(), + description: None, + amount: 0, + setup_amount: 0, + configuration: None, + }; + let (subscription_id, line_item_ids) = self + .db + .insert_subscription_with_line_items(&subscription, vec![line_item]) + .await?; + let subscription_line_item_id = line_item_ids[0]; + + let new_vm = Vm { + id: vm_id, + host_id: host.id, + user_id: user.id, + image_id: image.id, + template_id: None, + custom_template_id: Some(template_id), + subscription_line_item_id, + ssh_key_id: None, + disk_id: disk.id, + mac_address: spec + .mac_address + .clone() + .unwrap_or_else(|| "ff:ff:ff:ff:ff:ff".to_string()), + deleted: false, + ref_code: None, + disabled: false, + fw_policy_in: None, + fw_policy_out: None, + }; + + // Insert with the explicit (mapped) id so lifecycle ops target the right host VM + self.db.insert_vm_with_id(&new_vm).await?; + + // Name the subscription/line item and record the base monthly amount + let mut sub = self.db.get_subscription(subscription_id).await?; + sub.name = format!("VM{} subscription", new_vm.id); + self.db.update_subscription(&sub).await?; + + let mut li = self + .db + .get_subscription_line_item(subscription_line_item_id) + .await?; + li.name = format!("VM{} - {}", new_vm.id, pricing.name); + let price = + PricingEngine::get_custom_vm_cost_amount(&self.db, new_vm.id, &template).await?; + li.amount = price.total(); + self.db.update_subscription_line_item(&li).await?; + + Ok(new_vm) + } + /// Apply vm config to host pub async fn apply_vm_config_to_host(&self, vm_id: u64) -> OpResult<()> { let info = FullVmInfo::load(vm_id, self.db.clone()).await?; @@ -2427,4 +2601,93 @@ mod tests { VmStateCache::new(), )?) } + + /// Covers the import flow scenarios. Run as a single test because + /// [`DummyVmHost`] discovery is backed by a process-wide registry, so + /// splitting these into parallel tests would race on that global state. + #[tokio::test] + async fn test_import_vm() -> Result<()> { + use lnvps_api_common::HostVmSpec; + + fn spec(host_vm_id: i64, mapped: u64) -> HostVmSpec { + HostVmSpec { + host_vm_id, + mapped_vm_id: Some(mapped), + name: Some("legacy-vm".to_string()), + cpu: 2, + memory: 2 * GB, + disk_size: 20 * GB, + disk_storage: Some("mock-disk".to_string()), + mac_address: Some("bc:24:11:aa:bb:cc".to_string()), + running: true, + } + } + + // --- Scenario 1: successful import builds a custom template from specs + let db = Arc::new(MockDb::default()); + insert_custom_pricing(&db, DiskType::SSD, DiskInterface::PCIe).await?; + let (user, _ssh_key) = add_user(&db).await?; + crate::host::dummy_host::DummyVmHost::set_host_vms(vec![spec(200, 100)]).await; + + let provisioner = make_provisioner(db.clone()); + let vm = provisioner.import_vm(1, 200, user.id).await?; + + assert_eq!(vm.id, 100); + assert_eq!(vm.user_id, user.id); + assert_eq!(vm.host_id, 1); + assert_eq!(vm.disk_id, 1); + assert_eq!(vm.mac_address, "bc:24:11:aa:bb:cc"); + assert!(vm.template_id.is_none()); + let custom_template_id = vm.custom_template_id.expect("custom template set"); + + let tmpl = db.get_custom_vm_template(custom_template_id).await?; + assert_eq!(tmpl.cpu, 2); + assert_eq!(tmpl.memory, 2 * GB); + assert_eq!(tmpl.disk_size, 20 * GB); + assert_eq!(tmpl.disk_type, DiskType::SSD); + assert_eq!(tmpl.disk_interface, DiskInterface::PCIe); + + let li = db + .get_subscription_line_item(vm.subscription_line_item_id) + .await?; + assert!(li.amount > 0, "line item should have a priced amount"); + + let fetched = db.get_vm(100).await?; + assert_eq!(fetched.custom_template_id, Some(custom_template_id)); + + // --- Scenario 2: re-importing an already-tracked VM fails + let err = provisioner + .import_vm(1, 200, user.id) + .await + .expect_err("re-import must fail"); + assert!( + err.to_string().contains("already tracked"), + "unexpected error: {}", + err + ); + + // --- Scenario 3: VM not present on host + let err = provisioner + .import_vm(1, 9999, user.id) + .await + .expect_err("import must fail when VM not present on host"); + assert!(err.to_string().contains("not found"), "unexpected: {}", err); + + // --- Scenario 4: region has no custom pricing + let db2 = Arc::new(MockDb::default()); + let (user2, _k2) = add_user(&db2).await?; + crate::host::dummy_host::DummyVmHost::set_host_vms(vec![spec(205, 105)]).await; + let provisioner2 = make_provisioner(db2.clone()); + let err = provisioner2 + .import_vm(1, 205, user2.id) + .await + .expect_err("import must fail without custom pricing"); + assert!( + err.to_string().contains("custom pricing"), + "unexpected error: {}", + err + ); + + Ok(()) + } } diff --git a/lnvps_api/src/worker.rs b/lnvps_api/src/worker.rs index a2ae0051..057b5349 100644 --- a/lnvps_api/src/worker.rs +++ b/lnvps_api/src/worker.rs @@ -810,6 +810,31 @@ impl Worker { .expires } + /// Discover VMs present on a host that aren't tracked in the database. + /// + /// A host VM is "unmanaged" when it maps to a database id (i.e. is within + /// the managed id range) that has no live (non-deleted) VM record. Host VMs + /// outside the managed range (e.g. Proxmox vmid < 100) can't be imported and + /// are omitted. + async fn list_unmanaged_vms(&self, host_id: u64) -> Result> { + let host = self.db.get_host(host_id).await?; + let client = get_host_client(&host, &self.settings.provisioner_config)?; + let all = client.list_host_vms().await?; + + let mut unmanaged = Vec::new(); + for spec in all { + let Some(mapped) = spec.mapped_vm_id else { + // Outside the managed id range, not importable + continue; + }; + match self.db.get_vm(mapped).await { + Ok(vm) if !vm.deleted => continue, // already tracked + _ => unmanaged.push(spec), + } + } + Ok(unmanaged) + } + /// Check VM state from hypervisor and update cache /// Lifecycle enforcement (stop/delete) is handled by subscription lifecycle handlers. async fn check_vm(&self, vm: &Vm) -> Result<()> { @@ -2281,6 +2306,76 @@ impl Worker { vm.id, user_id ))); } + WorkJob::ListUnmanagedVms { + host_id, + reply_channel, + } => { + // Discover VMs on the host that aren't tracked in the database + // and reply with the JSON list on the requested temp channel. + let result = self.list_unmanaged_vms(*host_id).await; + let feedback = match &result { + Ok(list) => match serde_json::to_string(list) { + Ok(json) => JobFeedback::create_job_completed_feedback( + reply_channel.clone(), + "ListUnmanagedVms".to_string(), + Some(json), + ), + Err(e) => JobFeedback::create_job_failed_feedback( + reply_channel.clone(), + "ListUnmanagedVms".to_string(), + e.to_string(), + ), + }, + Err(e) => JobFeedback::create_job_failed_feedback( + reply_channel.clone(), + "ListUnmanagedVms".to_string(), + e.to_string(), + ), + }; + if let Err(e) = self.feedback.publish(feedback).await { + warn!("Failed to publish ListUnmanagedVms reply: {}", e); + } + return Ok(Some(match &result { + Ok(list) => format!("Found {} unmanaged VM(s) on host {}", list.len(), host_id), + Err(e) => format!("Discovery failed for host {}: {}", host_id, e), + })); + } + WorkJob::ImportVm { + host_id, + host_vm_id, + user_id, + admin_user_id, + reason, + } => { + info!( + "Admin {} importing host VM {} on host {} for user {}", + admin_user_id, host_vm_id, host_id, user_id + ); + let provisioner = self.subscription_handler.vm_provisioner(); + let vm = provisioner + .import_vm(*host_id, *host_vm_id, *user_id) + .await?; + + let metadata = Some(serde_json::json!({ + "admin_user_id": admin_user_id, + "admin_action": true, + "imported": true, + "host_vm_id": host_vm_id, + "reason": reason + })); + if let Err(e) = self + .vm_history_logger + .log_vm_created(&vm, Some(*user_id), metadata) + .await + { + error!("Failed to log VM {} import: {}", vm.id, e); + } + + return Ok(Some(format!( + "VM {} imported successfully for user {}", + vm.id, user_id + ))); + } WorkJob::SendEmailVerification { user_id, verify_url, diff --git a/lnvps_api_admin/Cargo.toml b/lnvps_api_admin/Cargo.toml index 636b2439..aad09df3 100644 --- a/lnvps_api_admin/Cargo.toml +++ b/lnvps_api_admin/Cargo.toml @@ -33,6 +33,7 @@ tokio.workspace = true payments-rs.workspace = true futures.workspace = true async-trait.workspace = true +uuid.workspace = true chrono = { version = "0.4.38", features = ["serde"] } clap = { version = "4.5.21", features = ["derive"] } diff --git a/lnvps_api_admin/src/admin/hosts.rs b/lnvps_api_admin/src/admin/hosts.rs index 740a1a0b..c672cc35 100644 --- a/lnvps_api_admin/src/admin/hosts.rs +++ b/lnvps_api_admin/src/admin/hosts.rs @@ -1,15 +1,21 @@ use crate::admin::RouterState; use crate::admin::auth::AdminAuth; -use crate::admin::model::{AdminHostDisk, AdminHostInfo, AdminVmHostKind}; +use crate::admin::model::{ + AdminHostDisk, AdminHostInfo, AdminImportVmRequest, AdminUnmanagedVmInfo, AdminVmHostKind, + JobResponse, +}; use axum::extract::{Path, Query, State}; -use axum::routing::get; +use axum::routing::{get, post}; use axum::{Json, Router}; +use futures::StreamExt; use lnvps_api_common::{ ApiData, ApiDiskInterface, ApiDiskType, ApiError, ApiPaginatedData, ApiPaginatedResult, - ApiResult, PageQuery, + ApiResult, JobFeedback, JobFeedbackStatus, PageQuery, WorkFeedback, WorkJob, }; use lnvps_db::{AdminAction, AdminResource}; +use log::info; use serde::Deserialize; +use std::time::Duration; pub fn router() -> Router { Router::new() @@ -30,6 +36,12 @@ pub fn router() -> Router { "/api/admin/v1/hosts/{id}/disks/{disk_id}", get(admin_get_host_disk).patch(admin_update_host_disk), ) + // VM import (issue #166) + .route( + "/api/admin/v1/hosts/{id}/vms/unmanaged", + get(admin_list_unmanaged_vms), + ) + .route("/api/admin/v1/hosts/{id}/vms/import", post(admin_import_vm)) } /// List all VM hosts with pagination @@ -460,6 +472,100 @@ pub struct AdminHostDiskUpdateRequest { pub enabled: Option, } +/// Discover VMs present on a host that aren't tracked in the database. +/// +/// The admin service has no direct connection to the host, so this dispatches a +/// discovery job to the worker and waits for the reply on a temporary Redis +/// pub/sub channel (see issue #166). +async fn admin_list_unmanaged_vms( + auth: AdminAuth, + State(this): State, + Path(id): Path, +) -> ApiResult> { + auth.require_permission(AdminResource::Hosts, AdminAction::View)?; + + // Ensure the host exists before dispatching work + let _host = this.db.get_host(id).await?; + + let feedback = this + .feedback + .as_ref() + .ok_or_else(|| ApiError::new("Job feedback service is not available"))?; + + // Temporary reply channel the worker publishes the result on + let reply_channel = uuid::Uuid::new_v4().to_string(); + let channel_name = JobFeedback::channel_name(&reply_channel); + + // Subscribe BEFORE dispatching so we can't miss the reply + let mut stream = feedback + .subscribe(&channel_name) + .await + .map_err(|e| ApiError::new(&format!("Failed to subscribe to reply channel: {}", e)))?; + + this.work_commander + .send(WorkJob::ListUnmanagedVms { + host_id: id, + reply_channel: reply_channel.clone(), + }) + .await + .map_err(|e| ApiError::new(&format!("Failed to dispatch discovery job: {}", e)))?; + + // Wait for the worker reply (bounded) + let msg = tokio::time::timeout(Duration::from_secs(30), stream.next()) + .await + .map_err(|_| ApiError::new("Timed out waiting for host discovery"))?; + + match msg { + Some(Ok(fb)) => match fb.status { + JobFeedbackStatus::Completed { result: Some(json) } => { + let specs: Vec = serde_json::from_str(&json) + .map_err(|e| ApiError::new(&format!("Invalid discovery result: {}", e)))?; + ApiData::ok(specs.into_iter().map(AdminUnmanagedVmInfo::from).collect()) + } + JobFeedbackStatus::Failed { error } => { + ApiData::err(&format!("Host discovery failed: {}", error)) + } + _ => ApiData::err("Unexpected discovery reply"), + }, + Some(Err(e)) => ApiData::err(&format!("Discovery reply error: {}", e)), + None => ApiData::err("Discovery reply channel closed"), + } +} + +/// Import an existing host VM into the database, assigning it to a user and +/// billing via the region's custom pricing (issue #166). +async fn admin_import_vm( + auth: AdminAuth, + State(this): State, + Path(id): Path, + Json(req): Json, +) -> ApiResult { + auth.require_permission(AdminResource::VirtualMachines, AdminAction::Create)?; + + // Validate host and user up front + let _host = this.db.get_host(id).await?; + let _user = this.db.get_user(req.user_id).await?; + + let job = WorkJob::ImportVm { + host_id: id, + host_vm_id: req.host_vm_id, + user_id: req.user_id, + admin_user_id: auth.user_id, + reason: req.reason, + }; + + match this.work_commander.send(job).await { + Ok(job_id) => { + info!( + "VM import job queued (host {}, vmid {}) with stream ID: {}", + id, req.host_vm_id, job_id + ); + ApiData::ok(JobResponse { job_id }) + } + Err(e) => ApiData::err(&format!("Failed to queue VM import job: {}", e)), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/lnvps_api_admin/src/admin/model.rs b/lnvps_api_admin/src/admin/model.rs index abb2b432..03b37cd4 100644 --- a/lnvps_api_admin/src/admin/model.rs +++ b/lnvps_api_admin/src/admin/model.rs @@ -2844,6 +2844,48 @@ pub struct AdminCreateVmRequest { pub reason: Option, } +/// Request to import an existing host VM into the database (issue #166) +#[derive(Deserialize)] +pub struct AdminImportVmRequest { + /// Raw host VM id (e.g. Proxmox vmid) + pub host_vm_id: i64, + /// User the imported VM is assigned to + pub user_id: u64, + pub reason: Option, +} + +/// A VM present on a host that is not tracked in the database +#[derive(Serialize)] +pub struct AdminUnmanagedVmInfo { + /// Raw host VM id (e.g. Proxmox vmid) + pub host_vm_id: i64, + /// Database id this VM would map to on import + pub mapped_vm_id: Option, + pub name: Option, + pub cpu: u16, + pub memory: u64, + pub disk_size: u64, + pub disk_storage: Option, + pub mac_address: Option, + pub running: bool, +} + +impl From for AdminUnmanagedVmInfo { + fn from(s: lnvps_api_common::HostVmSpec) -> Self { + Self { + host_vm_id: s.host_vm_id, + mapped_vm_id: s.mapped_vm_id, + name: s.name, + cpu: s.cpu, + memory: s.memory, + disk_size: s.disk_size, + disk_storage: s.disk_storage, + mac_address: s.mac_address, + running: s.running, + } + } +} + // ============================================================================ // Subscription Models // ============================================================================ diff --git a/lnvps_api_common/src/mock.rs b/lnvps_api_common/src/mock.rs index 2549a040..8f550505 100644 --- a/lnvps_api_common/src/mock.rs +++ b/lnvps_api_common/src/mock.rs @@ -984,6 +984,36 @@ impl LNVpsDbBase for MockDb { Ok(max_id + 1) } + async fn insert_vm_with_id(&self, vm: &Vm) -> DbResult { + let mut vms = self.vms.lock().await; + if vm.id == 0 { + return Err(DbError::from(anyhow!( + "insert_vm_with_id requires a non-zero id" + ))); + } + if vms.contains_key(&vm.id) { + return Err(DbError::from(anyhow!("VM id {} already exists", vm.id))); + } + + // lazy test FK + self.get_host(vm.host_id).await?; + self.get_user(vm.user_id).await?; + self.get_os_image(vm.image_id).await?; + if let Some(t) = vm.template_id { + self.get_vm_template(t).await?; + } + if let Some(t) = vm.custom_template_id { + self.get_custom_vm_template(t).await?; + } + if let Some(k) = vm.ssh_key_id { + self.get_user_ssh_key(k).await?; + } + self.get_host_disk(vm.disk_id).await?; + + vms.insert(vm.id, vm.clone()); + Ok(vm.id) + } + async fn delete_vm(&self, vm_id: u64) -> DbResult<()> { let mut vms = self.vms.lock().await; if let Some(vm) = vms.get_mut(&vm_id) { diff --git a/lnvps_api_common/src/model.rs b/lnvps_api_common/src/model.rs index e29f932d..4a8d3388 100644 --- a/lnvps_api_common/src/model.rs +++ b/lnvps_api_common/src/model.rs @@ -797,3 +797,31 @@ mod tests { assert!(s.contains(r#""ip_range_subscription_id":7"#)); } } + +/// A VM discovered directly on a host, described in host-native terms. +/// +/// Used to import VMs that exist on a host but are not tracked in the database +/// (see issue #166). `mapped_vm_id` is the LNVPS database id this host VM would +/// map to (e.g. Proxmox `vmid - 100`), or `None` when the host VM falls outside +/// the managed id range and therefore can't be imported. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct HostVmSpec { + /// Raw host VM id (e.g. Proxmox vmid) + pub host_vm_id: i64, + /// LNVPS database id this VM maps to, if within the managed range + pub mapped_vm_id: Option, + /// Host-reported VM name + pub name: Option, + /// Allocated CPU cores + pub cpu: u16, + /// Allocated memory in bytes + pub memory: u64, + /// Primary disk size in bytes + pub disk_size: u64, + /// Storage pool backing the primary disk + pub disk_storage: Option, + /// Primary NIC MAC address + pub mac_address: Option, + /// Whether the VM is currently running + pub running: bool, +} diff --git a/lnvps_api_common/src/work/mod.rs b/lnvps_api_common/src/work/mod.rs index c789532c..5e7af1a6 100644 --- a/lnvps_api_common/src/work/mod.rs +++ b/lnvps_api_common/src/work/mod.rs @@ -111,6 +111,24 @@ pub enum WorkJob { payment_method: String, // "lightning", "revolut", "paypal" lightning_invoice: Option, // Required when payment_method is "lightning" }, + /// Discover VMs present on a host that are not tracked in the database and + /// publish the resulting list (JSON `Vec`) to a temporary Redis + /// pub/sub channel so the admin API can read it synchronously. + ListUnmanagedVms { + host_id: u64, + /// Temporary channel id the worker replies on (via job feedback). + reply_channel: String, + }, + /// Import a VM that exists on a host but isn't tracked in the database, + /// assigning it to a user and billing it via the region's custom pricing. + ImportVm { + host_id: u64, + /// Raw host VM id (e.g. Proxmox vmid) + host_vm_id: i64, + user_id: u64, + admin_user_id: u64, + reason: Option, + }, /// Create a VM for a specific user (admin action) CreateVm { user_id: u64, @@ -172,6 +190,9 @@ impl WorkJob { Self::CheckVm { .. } => true, Self::CheckVms => true, Self::CheckSubscriptions => true, + // A discovery request is a one-shot read tied to a waiting admin + // request; never retry it if it fails. + Self::ListUnmanagedVms { .. } => true, _ => false, } } @@ -197,6 +218,8 @@ impl fmt::Display for WorkJob { WorkJob::UnassignVmIp { .. } => write!(f, "UnassignVmIp"), WorkJob::UpdateVmIp { .. } => write!(f, "UpdateVmIp"), WorkJob::ProcessVmRefund { .. } => write!(f, "ProcessVmRefund"), + WorkJob::ListUnmanagedVms { .. } => write!(f, "ListUnmanagedVms"), + WorkJob::ImportVm { .. } => write!(f, "ImportVm"), WorkJob::CreateVm { .. } => write!(f, "CreateVm"), WorkJob::SendEmailVerification { .. } => write!(f, "SendEmailVerification"), WorkJob::DownloadOsImages { .. } => write!(f, "DownloadOsImages"), diff --git a/lnvps_db/src/lib.rs b/lnvps_db/src/lib.rs index 5ff4cb72..1e7c4d92 100644 --- a/lnvps_db/src/lib.rs +++ b/lnvps_db/src/lib.rs @@ -367,6 +367,13 @@ pub trait LNVpsDbBase: Send + Sync { /// Insert a new VM record async fn insert_vm(&self, vm: &Vm) -> DbResult; + /// Insert a VM record with an explicit id. + /// + /// Used when importing a pre-existing host VM whose id is fixed by the + /// host's VM id mapping (e.g. Proxmox `vmid = db_id + 100`). Fails if the + /// id is already taken. + async fn insert_vm_with_id(&self, vm: &Vm) -> DbResult; + /// Delete a VM by id async fn delete_vm(&self, vm_id: u64) -> DbResult<()>; diff --git a/lnvps_db/src/mysql.rs b/lnvps_db/src/mysql.rs index 7f18486c..2142a62a 100644 --- a/lnvps_db/src/mysql.rs +++ b/lnvps_db/src/mysql.rs @@ -888,6 +888,24 @@ impl LNVpsDbBase for LNVpsDbMysql { .try_get(0)?) } + async fn insert_vm_with_id(&self, vm: &Vm) -> DbResult { + sqlx::query("insert into vm(id,host_id,user_id,image_id,template_id,custom_template_id,subscription_line_item_id,ssh_key_id,disk_id,mac_address,ref_code) values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)") + .bind(vm.id) + .bind(vm.host_id) + .bind(vm.user_id) + .bind(vm.image_id) + .bind(vm.template_id) + .bind(vm.custom_template_id) + .bind(vm.subscription_line_item_id) + .bind(vm.ssh_key_id) + .bind(vm.disk_id) + .bind(&vm.mac_address) + .bind(&vm.ref_code) + .execute(&self.db) + .await?; + Ok(vm.id) + } + async fn delete_vm(&self, vm_id: u64) -> DbResult<()> { sqlx::query("update vm set deleted = 1, ssh_key_id = null where id = ?") .bind(vm_id)