all admin features listing accounts and contracts with access controll and test ban a user by operator tests for all features mock data for test Reviewed-on: #3 Co-authored-by: Noor <noormohammedb@protonmail.com> Co-committed-by: Noor <noormohammedb@protonmail.com> Signed-off-by: Noor <noormohammedb@protonmail.com>
884 lines
29 KiB
Rust
884 lines
29 KiB
Rust
use std::str::FromStr;
|
|
use std::time::Duration;
|
|
|
|
use super::Error;
|
|
use crate::constants::{
|
|
ACCOUNT, ACTIVE_VM, DELETED_VM, NEW_VM_REQ, UPDATE_VM_REQ, VM_NODE, VM_UPDATE_EVENT,
|
|
};
|
|
use crate::db::{Account, ErrorFromTable, Report};
|
|
use crate::old_brain;
|
|
use detee_shared::vm_proto;
|
|
use serde::{Deserialize, Serialize};
|
|
use surrealdb::engine::remote::ws::Client;
|
|
use surrealdb::sql::Datetime;
|
|
use surrealdb::{Notification, RecordId, Surreal};
|
|
use tokio_stream::StreamExt as _;
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct VmNode {
|
|
pub id: RecordId,
|
|
pub operator: RecordId,
|
|
pub country: String,
|
|
pub region: String,
|
|
pub city: String,
|
|
pub ip: String,
|
|
pub avail_mem_mb: u32,
|
|
pub avail_vcpus: u32,
|
|
pub avail_storage_gbs: u32,
|
|
pub avail_ipv4: u32,
|
|
pub avail_ipv6: u32,
|
|
pub avail_ports: u32,
|
|
pub max_ports_per_vm: u32,
|
|
pub price: u64,
|
|
pub connected_at: Datetime,
|
|
pub disconnected_at: Datetime,
|
|
}
|
|
|
|
impl VmNode {
|
|
pub async fn register(self, db: &Surreal<Client>) -> Result<(), Error> {
|
|
Account::get_or_create(db, &self.operator.key().to_string()).await?;
|
|
let _: Option<VmNode> = db.upsert(self.id.clone()).content(self).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn set_online(db: &Surreal<Client>, vm_node_id: &str) -> Result<(), Error> {
|
|
db.query(format!("UPDATE {VM_NODE}:{vm_node_id} SET connected_at = time::now();")).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn set_offline(db: &Surreal<Client>, vm_node_id: &str) -> Result<(), Error> {
|
|
db.query(format!("UPDATE {VM_NODE}:{vm_node_id} SET disconnected_at = time::now();"))
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Serialize)]
|
|
pub struct VmNodeResources {
|
|
pub avail_mem_mb: u32,
|
|
pub avail_vcpus: u32,
|
|
pub avail_storage_gbs: u32,
|
|
pub avail_ipv4: u32,
|
|
pub avail_ipv6: u32,
|
|
pub avail_ports: u32,
|
|
pub max_ports_per_vm: u32,
|
|
}
|
|
|
|
impl VmNodeResources {
|
|
pub async fn merge(self, db: &Surreal<Client>, node_id: &str) -> Result<(), Error> {
|
|
let _: Option<VmNode> = db.update((VM_NODE, node_id)).merge(self).await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct VmNodeWithReports {
|
|
pub id: RecordId,
|
|
pub operator: RecordId,
|
|
pub country: String,
|
|
pub region: String,
|
|
pub city: String,
|
|
pub ip: String,
|
|
pub avail_mem_mb: u32,
|
|
pub avail_vcpus: u32,
|
|
pub avail_storage_gbs: u32,
|
|
pub avail_ipv4: u32,
|
|
pub avail_ipv6: u32,
|
|
pub avail_ports: u32,
|
|
pub max_ports_per_vm: u32,
|
|
pub price: u64,
|
|
pub reports: Vec<Report>,
|
|
}
|
|
|
|
impl VmNodeWithReports {
|
|
pub async fn find_by_filters(
|
|
db: &Surreal<Client>,
|
|
filters: vm_proto::VmNodeFilters,
|
|
) -> Result<Vec<Self>, Error> {
|
|
let mut query = format!(
|
|
"select *, <-report.* as reports from {VM_NODE} where
|
|
avail_ports >= {} &&
|
|
max_ports_per_vm >= {} &&
|
|
avail_ipv4 >= {} &&
|
|
avail_ipv6 >= {} &&
|
|
avail_vcpus >= {} &&
|
|
avail_mem_mb >= {} &&
|
|
avail_storage_gbs >= {}\n",
|
|
filters.free_ports,
|
|
filters.free_ports,
|
|
filters.offers_ipv4 as u32,
|
|
filters.offers_ipv6 as u32,
|
|
filters.vcpus,
|
|
filters.memory_mb,
|
|
filters.storage_gb
|
|
);
|
|
if !filters.city.is_empty() {
|
|
query += &format!("&& city = '{}' ", filters.city);
|
|
}
|
|
if !filters.region.is_empty() {
|
|
query += &format!("&& region = '{}' ", filters.region);
|
|
}
|
|
if !filters.country.is_empty() {
|
|
query += &format!("&& country = '{}' ", filters.country);
|
|
}
|
|
if !filters.ip.is_empty() {
|
|
query += &format!("&& ip = '{}' ", filters.ip);
|
|
}
|
|
query += " && connected_at > disconnected_at;";
|
|
let mut result = db.query(query).await?;
|
|
let vm_nodes: Vec<Self> = result.take(0)?;
|
|
Ok(vm_nodes)
|
|
}
|
|
}
|
|
|
|
pub enum VmDaemonMsg {
|
|
Create(NewVmReq),
|
|
Update(UpdateVmReq),
|
|
Delete(DeletedVm),
|
|
}
|
|
|
|
impl From<NewVmReq> for VmDaemonMsg {
|
|
fn from(value: NewVmReq) -> Self {
|
|
Self::Create(value)
|
|
}
|
|
}
|
|
|
|
impl From<UpdateVmReq> for VmDaemonMsg {
|
|
fn from(value: UpdateVmReq) -> Self {
|
|
Self::Update(value)
|
|
}
|
|
}
|
|
|
|
impl From<DeletedVm> for VmDaemonMsg {
|
|
fn from(value: DeletedVm) -> Self {
|
|
Self::Delete(value)
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct NewVmReq {
|
|
pub id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: RecordId,
|
|
pub hostname: String,
|
|
pub extra_ports: Vec<u32>,
|
|
pub public_ipv4: bool,
|
|
pub public_ipv6: bool,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_url: String,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub kernel_url: String,
|
|
pub created_at: Datetime,
|
|
pub price_per_unit: u64,
|
|
pub locked_nano: u64,
|
|
pub error: String,
|
|
}
|
|
|
|
impl NewVmReq {
|
|
pub async fn get(db: &Surreal<Client>, id: &str) -> Result<Option<Self>, Error> {
|
|
let new_vm_req: Option<Self> = db.select((NEW_VM_REQ, id)).await?;
|
|
Ok(new_vm_req)
|
|
}
|
|
|
|
pub async fn delete(db: &Surreal<Client>, id: &str) -> Result<(), Error> {
|
|
let _: Option<Self> = db.delete((NEW_VM_REQ, id)).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn submit_error(db: &Surreal<Client>, id: &str, error: String) -> Result<(), Error> {
|
|
#[derive(Serialize)]
|
|
struct NewVmError {
|
|
error: String,
|
|
}
|
|
let _: Option<Self> = db.update((NEW_VM_REQ, id)).merge(NewVmError { error }).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn submit(self, db: &Surreal<Client>) -> Result<(), Error> {
|
|
let locked_nano = self.locked_nano;
|
|
let account = self.admin.key().to_string();
|
|
let vm_id = self.id.key().to_string();
|
|
let vm_node = self.vm_node.key().to_string();
|
|
// TODO: check for possible injection and maybe use .bind()
|
|
let query = format!(
|
|
"
|
|
BEGIN TRANSACTION;
|
|
UPDATE account:{account} SET balance -= {locked_nano};
|
|
IF account:{account}.balance < 0 {{
|
|
THROW 'Insufficient funds.'
|
|
}};
|
|
UPDATE account:{account} SET tmp_locked += {locked_nano};
|
|
RELATE
|
|
account:{account}
|
|
->new_vm_req:{vm_id}
|
|
->vm_node:{vm_node}
|
|
CONTENT {{
|
|
created_at: time::now(), hostname: '{}', vcpus: {}, memory_mb: {}, disk_size_gb: {},
|
|
extra_ports: {:?}, public_ipv4: {:?}, public_ipv6: {:?},
|
|
dtrfs_url: '{}', dtrfs_sha: '{}', kernel_url: '{}', kernel_sha: '{}',
|
|
price_per_unit: {}, locked_nano: {locked_nano}, error: ''
|
|
}};
|
|
COMMIT TRANSACTION;
|
|
",
|
|
self.hostname,
|
|
self.vcpus,
|
|
self.memory_mb,
|
|
self.disk_size_gb,
|
|
self.extra_ports,
|
|
self.public_ipv4,
|
|
self.public_ipv6,
|
|
self.dtrfs_url,
|
|
self.dtrfs_sha,
|
|
self.kernel_url,
|
|
self.kernel_sha,
|
|
self.price_per_unit
|
|
);
|
|
//let _: Vec<Self> = db.insert(NEW_VM_REQ).relation(self).await?;
|
|
let mut query_resp = db.query(query).await?;
|
|
let resp_err = query_resp.take_errors();
|
|
|
|
if let Some(surrealdb::Error::Api(surrealdb::error::Api::Query(tx_query_error))) =
|
|
resp_err.get(&1)
|
|
{
|
|
log::error!("Transaction error: {tx_query_error}");
|
|
return Err(Error::InsufficientFunds);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// first string is the vm_id
|
|
pub enum WrappedMeasurement {
|
|
Args(String, vm_proto::MeasurementArgs),
|
|
Error(String, String),
|
|
}
|
|
|
|
impl WrappedMeasurement {
|
|
/// table must be NEW_VM_REQ or UPDATE_VM_REQ
|
|
/// it will however be enforced if you send anything else
|
|
pub async fn listen(
|
|
db: &Surreal<Client>,
|
|
vm_id: &str,
|
|
table: &str,
|
|
) -> Result<WrappedMeasurement, Error> {
|
|
let table = match table {
|
|
UPDATE_VM_REQ => UPDATE_VM_REQ,
|
|
_ => NEW_VM_REQ,
|
|
};
|
|
let mut resp = db
|
|
.query(format!("live select error from {table} where id = {NEW_VM_REQ}:{vm_id};"))
|
|
.query(format!(
|
|
"live select * from measurement_args where id = measurement_args:{vm_id};"
|
|
))
|
|
.await?;
|
|
let mut error_stream = resp.stream::<Notification<ErrorFromTable>>(0)?;
|
|
let mut args_stream = resp.stream::<Notification<vm_proto::MeasurementArgs>>(1)?;
|
|
|
|
let args: Option<vm_proto::MeasurementArgs> =
|
|
db.delete(("measurement_args", vm_id)).await?;
|
|
if let Some(args) = args {
|
|
return Ok(Self::Args(vm_id.to_string(), args));
|
|
}
|
|
log::trace!("listening for table: {table}");
|
|
|
|
tokio::time::timeout(Duration::from_secs(10), async {
|
|
loop {
|
|
tokio::select! {
|
|
error_notification = error_stream.next() => {
|
|
if let Some(err_notif) = error_notification {
|
|
match err_notif {
|
|
Ok(err_notif) => {
|
|
if err_notif.action == surrealdb::Action::Update
|
|
&& !err_notif.data.error.is_empty() {
|
|
return Ok::<WrappedMeasurement, Error>(
|
|
Self::Error(vm_id.to_string(),
|
|
err_notif.data.error)
|
|
);
|
|
};
|
|
},
|
|
Err(e) => return Err(e.into()),
|
|
}
|
|
}
|
|
}
|
|
args_notif = args_stream.next() => {
|
|
if let Some(args_notif) = args_notif {
|
|
match args_notif {
|
|
Ok(args_notif) => {
|
|
if args_notif.action == surrealdb::Action::Create {
|
|
let _: Option<vm_proto::MeasurementArgs> = db.delete(("measurement_args", vm_id)).await?;
|
|
return Ok(Self::Args(vm_id.to_string(), args_notif.data));
|
|
};
|
|
},
|
|
Err(e) => return Err(e.into()),
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
})
|
|
.await?
|
|
}
|
|
}
|
|
|
|
#[derive(Clone, Debug, Serialize, Deserialize)]
|
|
pub struct ActiveVm {
|
|
pub id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: RecordId,
|
|
pub hostname: String,
|
|
pub mapped_ports: Vec<(u32, u32)>,
|
|
pub public_ipv4: String,
|
|
pub public_ipv6: String,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub created_at: Datetime,
|
|
pub price_per_unit: u64,
|
|
pub locked_nano: u64,
|
|
pub collected_at: Datetime,
|
|
}
|
|
|
|
impl ActiveVm {
|
|
/// total hardware units of this VM
|
|
fn total_units(&self) -> u64 {
|
|
// TODO: Optimize this based on price of hardware.
|
|
// I tried, but this can be done better.
|
|
// Storage cost should also be based on tier
|
|
(self.vcpus as u64 * 10)
|
|
+ ((self.memory_mb + 256) as u64 / 200)
|
|
+ (self.disk_size_gb as u64 / 10)
|
|
+ (!self.public_ipv4.is_empty() as u64 * 10)
|
|
}
|
|
|
|
/// Returns price per minute in nanoLP
|
|
pub fn price_per_minute(&self) -> u64 {
|
|
self.total_units() * self.price_per_unit
|
|
}
|
|
|
|
pub async fn get_by_uuid(db: &Surreal<Client>, uuid: &str) -> Result<Option<Self>, Error> {
|
|
let contract: Option<Self> =
|
|
db.query(format!("select * from {ACTIVE_VM}:{uuid};")).await?.take(0)?;
|
|
Ok(contract)
|
|
}
|
|
|
|
pub async fn activate(
|
|
db: &Surreal<Client>,
|
|
id: &str,
|
|
args: vm_proto::MeasurementArgs,
|
|
) -> Result<(), Error> {
|
|
let new_vm_req = match NewVmReq::get(db, id).await? {
|
|
Some(r) => r,
|
|
None => return Ok(()),
|
|
};
|
|
|
|
let mut public_ipv4 = String::new();
|
|
let mut public_ipv6 = String::new();
|
|
|
|
for ip in args.ips.iter() {
|
|
if let Ok(ipv4_addr) = std::net::Ipv4Addr::from_str(&ip.address) {
|
|
if !ipv4_addr.is_private() && !ipv4_addr.is_link_local() {
|
|
public_ipv4 = ipv4_addr.to_string();
|
|
}
|
|
continue;
|
|
}
|
|
if let Ok(ipv6_addr) = std::net::Ipv6Addr::from_str(&ip.address) {
|
|
public_ipv6 = ipv6_addr.to_string();
|
|
}
|
|
}
|
|
|
|
let mut mapped_ports = Vec::new();
|
|
let mut guest_ports = vec![22];
|
|
guest_ports.append(&mut args.exposed_ports.clone());
|
|
let mut i = 0;
|
|
while i < args.exposed_ports.len() && i < guest_ports.len() {
|
|
mapped_ports.push((args.exposed_ports[i], guest_ports[i]));
|
|
i += 1;
|
|
}
|
|
|
|
let active_vm = ActiveVm {
|
|
id: RecordId::from((ACTIVE_VM, id)),
|
|
admin: new_vm_req.admin,
|
|
vm_node: new_vm_req.vm_node,
|
|
hostname: new_vm_req.hostname,
|
|
mapped_ports,
|
|
public_ipv4,
|
|
public_ipv6,
|
|
disk_size_gb: new_vm_req.disk_size_gb,
|
|
vcpus: new_vm_req.vcpus,
|
|
memory_mb: new_vm_req.memory_mb,
|
|
dtrfs_sha: new_vm_req.dtrfs_sha,
|
|
kernel_sha: new_vm_req.kernel_sha,
|
|
created_at: new_vm_req.created_at.clone(),
|
|
price_per_unit: new_vm_req.price_per_unit,
|
|
locked_nano: new_vm_req.locked_nano,
|
|
collected_at: new_vm_req.created_at,
|
|
};
|
|
|
|
let admin_account = active_vm.admin.key().to_string();
|
|
let locked_nano = active_vm.locked_nano;
|
|
let _: Vec<ActiveVm> = db.insert(()).relation(active_vm).await?;
|
|
NewVmReq::delete(db, id).await?;
|
|
db.query(format!("UPDATE {ACCOUNT}:{admin_account} SET tmp_locked -= {locked_nano};"))
|
|
.await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn update(db: &Surreal<Client>, id: &str) -> Result<(), Error> {
|
|
let update_vm_req = match UpdateVmReq::get(db, id).await? {
|
|
Some(r) => r,
|
|
None => return Ok(()),
|
|
};
|
|
|
|
let mut active_vm = match Self::get_by_uuid(db, id).await? {
|
|
Some(vm) => vm,
|
|
None => return Ok(()),
|
|
};
|
|
|
|
if update_vm_req.vcpus > 0 {
|
|
active_vm.vcpus = update_vm_req.vcpus;
|
|
}
|
|
if update_vm_req.memory_mb > 0 {
|
|
active_vm.memory_mb = update_vm_req.memory_mb;
|
|
}
|
|
if update_vm_req.disk_size_gb > 0 {
|
|
active_vm.disk_size_gb = update_vm_req.disk_size_gb;
|
|
}
|
|
if !update_vm_req.dtrfs_sha.is_empty() && !update_vm_req.kernel_sha.is_empty() {
|
|
active_vm.dtrfs_sha = update_vm_req.dtrfs_sha;
|
|
active_vm.kernel_sha = update_vm_req.kernel_sha;
|
|
}
|
|
|
|
let _: Option<ActiveVm> = db.update(active_vm.id.clone()).content(active_vm).await?;
|
|
UpdateVmReq::delete(db, id).await?;
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn change_hostname(
|
|
db: &Surreal<Client>,
|
|
id: &str,
|
|
new_hostname: &str,
|
|
) -> Result<bool, Error> {
|
|
let contract: Option<Self> = db
|
|
.query(format!(
|
|
"UPDATE {ACTIVE_VM}:{id} SET hostname = '{new_hostname}' RETURN BEFORE;"
|
|
))
|
|
.await?
|
|
.take(0)?;
|
|
if let Some(contract) = contract {
|
|
if contract.hostname != new_hostname {
|
|
return Ok(true);
|
|
}
|
|
}
|
|
Ok(false)
|
|
}
|
|
|
|
pub async fn delete(db: &Surreal<Client>, id: &str) -> Result<bool, Error> {
|
|
let deleted_vm: Option<Self> = db.delete((ACTIVE_VM, id)).await?;
|
|
if let Some(deleted_vm) = deleted_vm {
|
|
let deleted_vm: DeletedVm = deleted_vm.into();
|
|
let _: Vec<DeletedVm> = db.insert(DELETED_VM).relation(deleted_vm).await?;
|
|
Ok(true)
|
|
} else {
|
|
Ok(false)
|
|
}
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct UpdateVmReq {
|
|
pub id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: RecordId,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_url: String,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub kernel_url: String,
|
|
pub created_at: Datetime,
|
|
pub error: String,
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct UpdateVmEvent {
|
|
pub vm_id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: RecordId,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_url: String,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub kernel_url: String,
|
|
pub executed_at: Datetime,
|
|
}
|
|
|
|
impl From<UpdateVmReq> for UpdateVmEvent {
|
|
fn from(update_vm_req: UpdateVmReq) -> Self {
|
|
Self {
|
|
vm_id: RecordId::from((VM_UPDATE_EVENT, update_vm_req.id.key().to_string())),
|
|
admin: update_vm_req.admin,
|
|
vm_node: update_vm_req.vm_node,
|
|
disk_size_gb: update_vm_req.disk_size_gb,
|
|
vcpus: update_vm_req.vcpus,
|
|
memory_mb: update_vm_req.memory_mb,
|
|
dtrfs_url: update_vm_req.dtrfs_url,
|
|
dtrfs_sha: update_vm_req.dtrfs_sha,
|
|
kernel_sha: update_vm_req.kernel_sha,
|
|
kernel_url: update_vm_req.kernel_url,
|
|
executed_at: Datetime::default(),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl UpdateVmReq {
|
|
pub async fn get(db: &Surreal<Client>, id: &str) -> Result<Option<Self>, Error> {
|
|
let update_vm_req: Option<Self> = db.select((UPDATE_VM_REQ, id)).await?;
|
|
Ok(update_vm_req)
|
|
}
|
|
|
|
pub async fn delete(db: &Surreal<Client>, id: &str) -> Result<(), Error> {
|
|
let update_vm_req: Option<Self> = db.delete((UPDATE_VM_REQ, id)).await?;
|
|
if let Some(update_vm_req) = update_vm_req {
|
|
let update_vm_event: UpdateVmEvent = update_vm_req.into();
|
|
let _: Option<UpdateVmEvent> =
|
|
db.create(VM_UPDATE_EVENT).content(update_vm_event).await?;
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
/// returns None if VM does not exist
|
|
/// returns Some(false) if hw update is not needed
|
|
/// returns Some(true) if hw update is needed and got submitted
|
|
/// returns error if something happened with the DB
|
|
pub async fn request_hw_update(mut self, db: &Surreal<Client>) -> Result<Option<bool>, Error> {
|
|
let contract = ActiveVm::get_by_uuid(db, &self.id.key().to_string()).await?;
|
|
|
|
if contract.is_none() {
|
|
return Ok(None);
|
|
}
|
|
let contract = contract.unwrap();
|
|
// this is needed cause TryFrom does not support await
|
|
self.vm_node = contract.vm_node;
|
|
|
|
if !((self.vcpus != 0 && contract.vcpus != self.vcpus)
|
|
|| (self.memory_mb != 0 && contract.memory_mb != self.memory_mb)
|
|
|| (!self.dtrfs_sha.is_empty() && contract.dtrfs_sha != self.dtrfs_sha)
|
|
|| (self.disk_size_gb != 0 && contract.disk_size_gb != self.disk_size_gb))
|
|
{
|
|
return Ok(Some(false));
|
|
}
|
|
|
|
let _: Vec<Self> = db.insert(UPDATE_VM_REQ).relation(self).await?;
|
|
Ok(Some(true))
|
|
}
|
|
|
|
pub async fn submit_error(db: &Surreal<Client>, id: &str, error: String) -> Result<(), Error> {
|
|
#[derive(Serialize)]
|
|
struct UpdateVmError {
|
|
error: String,
|
|
}
|
|
let _: Option<Self> = db.update((UPDATE_VM_REQ, id)).merge(UpdateVmError { error }).await?;
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct DeletedVm {
|
|
pub id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: RecordId,
|
|
pub hostname: String,
|
|
pub mapped_ports: Vec<(u32, u32)>,
|
|
pub public_ipv4: String,
|
|
pub public_ipv6: String,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub created_at: Datetime,
|
|
pub deleted_at: Datetime,
|
|
pub price_per_unit: u64,
|
|
}
|
|
|
|
impl From<ActiveVm> for DeletedVm {
|
|
fn from(active_vm: ActiveVm) -> Self {
|
|
Self {
|
|
id: RecordId::from((DELETED_VM, active_vm.id.key().to_string())),
|
|
admin: active_vm.admin,
|
|
vm_node: active_vm.vm_node,
|
|
hostname: active_vm.hostname,
|
|
mapped_ports: active_vm.mapped_ports,
|
|
public_ipv4: active_vm.public_ipv4,
|
|
public_ipv6: active_vm.public_ipv6,
|
|
disk_size_gb: active_vm.disk_size_gb,
|
|
vcpus: active_vm.vcpus,
|
|
memory_mb: active_vm.memory_mb,
|
|
dtrfs_sha: active_vm.dtrfs_sha,
|
|
kernel_sha: active_vm.kernel_sha,
|
|
created_at: active_vm.created_at,
|
|
deleted_at: Datetime::default(),
|
|
price_per_unit: active_vm.price_per_unit,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl DeletedVm {
|
|
pub async fn get_by_uuid(db: &Surreal<Client>, uuid: &str) -> Result<Option<Self>, Error> {
|
|
let contract: Option<Self> =
|
|
db.query(format!("select * from {DELETED_VM}:{uuid};")).await?.take(0)?;
|
|
Ok(contract)
|
|
}
|
|
|
|
pub async fn list_by_admin(db: &Surreal<Client>, admin: &str) -> Result<Vec<Self>, Error> {
|
|
let mut result =
|
|
db.query(format!("select * from {DELETED_VM} where in = {ACCOUNT}:{admin};")).await?;
|
|
let contracts: Vec<Self> = result.take(0)?;
|
|
Ok(contracts)
|
|
}
|
|
|
|
pub async fn list_by_node(db: &Surreal<Client>, admin: &str) -> Result<Vec<Self>, Error> {
|
|
let mut result =
|
|
db.query(format!("select * from {DELETED_VM} where out = {VM_NODE}:{admin};")).await?;
|
|
let contracts: Vec<Self> = result.take(0)?;
|
|
Ok(contracts)
|
|
}
|
|
|
|
pub async fn list_by_operator(
|
|
db: &Surreal<Client>,
|
|
operator: &str,
|
|
) -> Result<Vec<Self>, Error> {
|
|
let mut result = db
|
|
.query(format!(
|
|
"select
|
|
(select * from ->operator->vm_node<-{DELETED_VM}) as contracts
|
|
from {ACCOUNT}:{operator};"
|
|
))
|
|
.await?;
|
|
|
|
#[derive(Deserialize)]
|
|
struct Wrapper {
|
|
contracts: Vec<DeletedVm>,
|
|
}
|
|
|
|
let c: Option<Wrapper> = result.take(0)?;
|
|
match c {
|
|
Some(c) => Ok(c.contracts),
|
|
None => Ok(Vec::new()),
|
|
}
|
|
}
|
|
|
|
/// total hardware units of this VM
|
|
fn total_units(&self) -> u64 {
|
|
// TODO: Optimize this based on price of hardware.
|
|
// I tried, but this can be done better.
|
|
// Storage cost should also be based on tier
|
|
(self.vcpus as u64 * 10)
|
|
+ ((self.memory_mb + 256) as u64 / 200)
|
|
+ (self.disk_size_gb as u64 / 10)
|
|
+ (!self.public_ipv4.is_empty() as u64 * 10)
|
|
}
|
|
|
|
/// Returns price per minute in nanoLP
|
|
pub fn price_per_minute(&self) -> u64 {
|
|
self.total_units() * self.price_per_unit
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Serialize, Deserialize)]
|
|
pub struct ActiveVmWithNode {
|
|
pub id: RecordId,
|
|
#[serde(rename = "in")]
|
|
pub admin: RecordId,
|
|
#[serde(rename = "out")]
|
|
pub vm_node: VmNode,
|
|
pub hostname: String,
|
|
pub mapped_ports: Vec<(u32, u32)>,
|
|
pub public_ipv4: String,
|
|
pub public_ipv6: String,
|
|
pub disk_size_gb: u32,
|
|
pub vcpus: u32,
|
|
pub memory_mb: u32,
|
|
pub dtrfs_sha: String,
|
|
pub kernel_sha: String,
|
|
pub created_at: Datetime,
|
|
pub price_per_unit: u64,
|
|
pub locked_nano: u64,
|
|
pub collected_at: Datetime,
|
|
}
|
|
|
|
impl From<ActiveVmWithNode> for ActiveVm {
|
|
fn from(val: ActiveVmWithNode) -> Self {
|
|
Self {
|
|
id: val.id,
|
|
admin: val.admin,
|
|
vm_node: val.vm_node.id,
|
|
hostname: val.hostname,
|
|
mapped_ports: val.mapped_ports,
|
|
public_ipv4: val.public_ipv4,
|
|
public_ipv6: val.public_ipv6,
|
|
disk_size_gb: val.disk_size_gb,
|
|
vcpus: val.vcpus,
|
|
memory_mb: val.memory_mb,
|
|
dtrfs_sha: val.dtrfs_sha,
|
|
kernel_sha: val.kernel_sha,
|
|
created_at: val.created_at,
|
|
price_per_unit: val.price_per_unit,
|
|
locked_nano: val.locked_nano,
|
|
collected_at: val.collected_at,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl ActiveVmWithNode {
|
|
pub async fn get_by_uuid(db: &Surreal<Client>, uuid: &str) -> Result<Option<Self>, Error> {
|
|
let contract: Option<Self> = db
|
|
.query(format!("select * from {ACTIVE_VM} where id = $uuid_input fetch out;"))
|
|
.bind(("uuid_input", RecordId::from((ACTIVE_VM, uuid))))
|
|
.await?
|
|
.take(0)?;
|
|
Ok(contract)
|
|
}
|
|
|
|
pub async fn list_by_admin(db: &Surreal<Client>, admin: &str) -> Result<Vec<Self>, Error> {
|
|
let mut result = db
|
|
.query(format!("select * from {ACTIVE_VM} where in = {ACCOUNT}:{admin} fetch out;"))
|
|
.await?;
|
|
let contracts: Vec<Self> = result.take(0)?;
|
|
Ok(contracts)
|
|
}
|
|
|
|
pub async fn list_by_node(db: &Surreal<Client>, admin: &str) -> Result<Vec<Self>, Error> {
|
|
let mut result = db
|
|
.query(format!("select * from {ACTIVE_VM} where out = {VM_NODE}:{admin} fetch out;"))
|
|
.await?;
|
|
let contracts: Vec<Self> = result.take(0)?;
|
|
Ok(contracts)
|
|
}
|
|
|
|
pub async fn list_by_operator(
|
|
db: &Surreal<Client>,
|
|
operator: &str,
|
|
) -> Result<Vec<Self>, Error> {
|
|
let mut result = db
|
|
.query(format!(
|
|
"select
|
|
(select * from ->operator->vm_node<-{ACTIVE_VM} fetch out) as contracts
|
|
from {ACCOUNT}:{operator};"
|
|
))
|
|
.await?;
|
|
|
|
#[derive(Deserialize)]
|
|
struct Wrapper {
|
|
contracts: Vec<ActiveVmWithNode>,
|
|
}
|
|
|
|
let c: Option<Wrapper> = result.take(0)?;
|
|
match c {
|
|
Some(c) => Ok(c.contracts),
|
|
None => Ok(Vec::new()),
|
|
}
|
|
}
|
|
|
|
/// total hardware units of this VM
|
|
fn total_units(&self) -> u64 {
|
|
// TODO: Optimize this based on price of hardware.
|
|
// I tried, but this can be done better.
|
|
// Storage cost should also be based on tier
|
|
(self.vcpus as u64 * 10)
|
|
+ ((self.memory_mb + 256) as u64 / 200)
|
|
+ (self.disk_size_gb as u64 / 10)
|
|
+ (!self.public_ipv4.is_empty() as u64 * 10)
|
|
}
|
|
|
|
/// Returns price per minute in nanoLP
|
|
pub fn price_per_minute(&self) -> u64 {
|
|
self.total_units() * self.price_per_unit
|
|
}
|
|
|
|
pub async fn list_all(db: &Surreal<Client>) -> Result<Vec<Self>, Error> {
|
|
let mut query_response = db.query(format!("SELECT * FROM {ACTIVE_VM} FETCH out;")).await?;
|
|
let active_vms: Vec<Self> = query_response.take(0)?;
|
|
Ok(active_vms)
|
|
}
|
|
}
|
|
|
|
// TODO: delete all of these From implementation after migration 0 gets executed
|
|
|
|
impl From<&old_brain::BrainData> for Vec<VmNode> {
|
|
fn from(old_data: &old_brain::BrainData) -> Self {
|
|
let mut nodes = Vec::new();
|
|
for old_node in old_data.vm_nodes.iter() {
|
|
nodes.push(VmNode {
|
|
id: RecordId::from((VM_NODE, old_node.public_key.clone())),
|
|
operator: RecordId::from((ACCOUNT, old_node.operator_wallet.clone())),
|
|
country: old_node.country.clone(),
|
|
region: old_node.region.clone(),
|
|
city: old_node.city.clone(),
|
|
ip: old_node.ip.clone(),
|
|
avail_mem_mb: old_node.avail_mem_mb,
|
|
avail_vcpus: old_node.avail_vcpus,
|
|
avail_storage_gbs: old_node.avail_storage_gbs,
|
|
avail_ipv4: old_node.avail_ipv4,
|
|
avail_ipv6: old_node.avail_ipv6,
|
|
avail_ports: old_node.avail_ports,
|
|
max_ports_per_vm: old_node.max_ports_per_vm,
|
|
price: old_node.price,
|
|
disconnected_at: Datetime::default(),
|
|
connected_at: Datetime::default(),
|
|
});
|
|
}
|
|
nodes
|
|
}
|
|
}
|
|
|
|
impl From<&old_brain::BrainData> for Vec<ActiveVm> {
|
|
fn from(old_data: &old_brain::BrainData) -> Self {
|
|
let mut contracts = Vec::new();
|
|
for old_c in old_data.vm_contracts.iter() {
|
|
let mut mapped_ports = Vec::new();
|
|
for port in old_c.exposed_ports.iter() {
|
|
mapped_ports.push((*port, 8080u32));
|
|
}
|
|
contracts.push(ActiveVm {
|
|
id: RecordId::from((ACTIVE_VM, old_c.uuid.replace("-", ""))),
|
|
admin: RecordId::from((ACCOUNT, old_c.admin_pubkey.clone())),
|
|
vm_node: RecordId::from((VM_NODE, old_c.node_pubkey.clone())),
|
|
hostname: old_c.hostname.clone(),
|
|
mapped_ports,
|
|
public_ipv4: old_c.public_ipv4.clone(),
|
|
public_ipv6: old_c.public_ipv6.clone(),
|
|
disk_size_gb: old_c.disk_size_gb,
|
|
vcpus: old_c.vcpus,
|
|
memory_mb: old_c.memory_mb,
|
|
dtrfs_sha: old_c.dtrfs_sha.clone(),
|
|
kernel_sha: old_c.kernel_sha.clone(),
|
|
price_per_unit: old_c.price_per_unit,
|
|
locked_nano: old_c.locked_nano,
|
|
created_at: old_c.created_at.into(),
|
|
collected_at: old_c.collected_at.into(),
|
|
});
|
|
}
|
|
contracts
|
|
}
|
|
}
|