diff --git a/core/src/cache.rs b/core/src/cache.rs new file mode 100644 index 000000000..41fa72446 --- /dev/null +++ b/core/src/cache.rs @@ -0,0 +1,58 @@ +use std::collections::HashMap; + +use monitor_types::busy::Busy; +use tokio::sync::RwLock; + +#[derive(Default)] +pub struct Cache { + cache: RwLock>, +} + +impl Cache { + pub async fn get(&self, key: &str) -> Option { + self.cache.read().await.get(key).cloned() + } + + // pub async fn get_or_default(&self, key: String) -> T { + // let mut cache = self.cache.write().await; + // cache.entry(key).or_default().clone() + // } + + // pub async fn get_list(&self, filter: Option bool>) -> Vec { + // let cache = self.cache.read().await; + // match filter { + // Some(filter) => cache + // .iter() + // .filter(|(k, v)| filter(k, v)) + // .map(|(_, e)| e.clone()) + // .collect(), + // None => cache.iter().map(|(_, e)| e.clone()).collect(), + // } + // } + + pub async fn insert(&self, key: impl Into, val: T) { + self.cache.write().await.insert(key.into(), val); + } + + pub async fn update_entry(&self, key: impl Into, handler: impl Fn(&mut T)) { + let mut cache = self.cache.write().await; + handler(cache.entry(key.into()).or_default()); + } + + // pub async fn clear(&self) { + // self.cache.write().await.clear(); + // } + + pub async fn remove(&self, key: &str) { + self.cache.write().await.remove(key); + } +} + +impl Cache { + pub async fn busy(&self, id: &str) -> bool { + match self.get(id).await { + Some(state) => state.busy(), + None => false, + } + } +} \ No newline at end of file diff --git a/core/src/channel.rs b/core/src/channel.rs new file mode 100644 index 000000000..ce1da5978 --- /dev/null +++ b/core/src/channel.rs @@ -0,0 +1,16 @@ +use tokio::sync::{broadcast, Mutex}; + +pub struct BroadcastChannel { + pub sender: Mutex>, + pub receiver: broadcast::Receiver, +} + +impl BroadcastChannel { + pub fn new(capacity: usize) -> BroadcastChannel { + let (sender, receiver) = broadcast::channel(capacity); + BroadcastChannel { + sender: sender.into(), + receiver, + } + } +} \ No newline at end of file diff --git a/core/src/helpers.rs b/core/src/helpers.rs index af65ab678..0efcd6695 100644 --- a/core/src/helpers.rs +++ b/core/src/helpers.rs @@ -1,8 +1,7 @@ -use std::{collections::HashMap, time::Duration}; +use std::time::Duration; use anyhow::{anyhow, Context}; use monitor_types::{ - busy::Busy, entities::{ alerter::Alerter, build::Build, @@ -19,10 +18,40 @@ use monitor_types::{ }; use periphery_client::{requests, PeripheryClient}; use rand::{thread_rng, Rng}; -use tokio::sync::RwLock; use crate::{auth::RequestUser, state::State}; +pub fn empty_or_only_spaces(word: &str) -> bool { + if word.is_empty() { + return true; + } + for char in word.chars() { + if char != ' ' { + return false; + } + } + true +} + +pub fn random_duration(min_ms: u64, max_ms: u64) -> Duration { + Duration::from_millis(thread_rng().gen_range(min_ms..max_ms)) +} + +pub fn make_update( + target: impl Into, + operation: Operation, + user: &RequestUser, +) -> Update { + Update { + start_ts: monitor_timestamp(), + target: target.into(), + operation, + operator: user.id.clone(), + success: true, + ..Default::default() + } +} + impl State { pub async fn get_user(&self, user_id: &str) -> anyhow::Result { self.db @@ -379,87 +408,3 @@ impl State { PeripheryClient::new(&server.config.address, &self.config.passkey) } } - -#[derive(Default)] -pub struct Cache { - cache: RwLock>, -} - -impl Cache { - pub async fn get(&self, key: &str) -> Option { - self.cache.read().await.get(key).cloned() - } - - // pub async fn get_or_default(&self, key: String) -> T { - // let mut cache = self.cache.write().await; - // cache.entry(key).or_default().clone() - // } - - // pub async fn get_list(&self, filter: Option bool>) -> Vec { - // let cache = self.cache.read().await; - // match filter { - // Some(filter) => cache - // .iter() - // .filter(|(k, v)| filter(k, v)) - // .map(|(_, e)| e.clone()) - // .collect(), - // None => cache.iter().map(|(_, e)| e.clone()).collect(), - // } - // } - - pub async fn insert(&self, key: impl Into, val: T) { - self.cache.write().await.insert(key.into(), val); - } - - pub async fn update_entry(&self, key: impl Into, handler: impl Fn(&mut T)) { - let mut cache = self.cache.write().await; - handler(cache.entry(key.into()).or_default()); - } - - // pub async fn clear(&self) { - // self.cache.write().await.clear(); - // } - - pub async fn remove(&self, key: &str) { - self.cache.write().await.remove(key); - } -} - -impl Cache { - pub async fn busy(&self, id: &str) -> bool { - match self.get(id).await { - Some(state) => state.busy(), - None => false, - } - } -} - -pub fn empty_or_only_spaces(word: &str) -> bool { - if word.is_empty() { - return true; - } - for char in word.chars() { - if char != ' ' { - return false; - } - } - true -} - -pub fn random_duration(min_ms: u64, max_ms: u64) -> Duration { - Duration::from_millis(thread_rng().gen_range(min_ms..max_ms)) -} - -pub fn make_update( - target: impl Into, - operation: Operation, - user: &RequestUser, -) -> Update { - Update { - start_ts: monitor_timestamp(), - target: target.into(), - operation, - operator: user.id.clone(), - ..Default::default() - } -} diff --git a/core/src/main.rs b/core/src/main.rs index be2cde698..0f02f2a00 100644 --- a/core/src/main.rs +++ b/core/src/main.rs @@ -5,6 +5,8 @@ use axum::{Extension, Router}; use termination_signal::tokio::immediate_term_handle; mod auth; +mod cache; +mod channel; mod cloud; mod config; mod db; diff --git a/core/src/requests/execute/build.rs b/core/src/requests/execute/build.rs index a440f1eb3..87143e46d 100644 --- a/core/src/requests/execute/build.rs +++ b/core/src/requests/execute/build.rs @@ -9,11 +9,11 @@ use monitor_types::{ build::{Build, BuildBuilderConfig}, builder::{AwsBuilderConfig, BuilderConfig}, deployment::DockerContainerState, - update::{Log, ResourceTarget, Update, UpdateStatus}, + update::{Log, Update}, Operation, PermissionLevel, }, monitor_timestamp, - requests::execute::{Deploy, RunBuild}, + requests::execute::{CancelBuild, CancelBuildResponse, Deploy, RunBuild}, }; use mungos::mongodb::bson::{doc, to_bson}; use periphery_client::{ @@ -21,10 +21,12 @@ use periphery_client::{ PeripheryClient, }; use resolver_api::Resolve; +use tokio_util::sync::CancellationToken; use crate::{ auth::{InnerRequestUser, RequestUser}, cloud::{aws::Ec2Instance, BuildCleanupData}, + helpers::make_update, state::State, }; @@ -43,18 +45,39 @@ impl Resolve for State { .get_build_check_permissions(&build_id, &user, PermissionLevel::Execute) .await?; - let inner = || async move { - build.config.version.increment(); - let mut update = Update { - target: ResourceTarget::Build(build.id.clone()), - operation: Operation::RunBuild, - start_ts: monitor_timestamp(), - status: UpdateStatus::InProgress, - success: true, - operator: user.id.clone(), - version: build.config.version.clone(), - ..Default::default() + build.config.version.increment(); + + let mut update = make_update(&build, Operation::RunBuild, &user); + update.in_progress(); + update.version = build.config.version.clone(); + + let cancel = CancellationToken::new(); + let cancel_clone = cancel.clone(); + let mut cancel_recv = self.build_cancel.receiver.resubscribe(); + let id_clone = build_id.clone(); + + tokio::spawn(async move { + let poll = async { + loop { + let build_id = tokio::select! { + _ = cancel_clone.cancelled() => return Ok(()), + id = cancel_recv.recv() => id? + }; + if build_id == id_clone { + cancel_clone.cancel(); + return Ok(()); + } + } + #[allow(unreachable_code)] + anyhow::Ok(()) }; + tokio::select! { + _ = cancel_clone.cancelled() => {} + _ = poll => {} + } + }); + + let inner = || async move { update.id = self.add_update(update.clone()).await?; // GET BUILDER PERIPHERY @@ -74,12 +97,22 @@ impl Resolve for State { // CLONE REPO - let clone_success = match periphery - .request(requests::CloneRepo { - args: (&build).into(), - }) - .await - { + let res = tokio::select! { + _ = cancel.cancelled() => { + update.push_error_log("build cancelled", String::from("user cancelled build during repo clone")); + self.cleanup_builder_instance(periphery, cleanup_data, &mut update) + .await; + update.finalize(); + self.update_update(update.clone()).await?; + return Ok(update) + }, + res = periphery + .request(requests::CloneRepo { + args: (&build).into(), + }) => res, + }; + + let clone_success = match res { Ok(clone_logs) => { let success = all_logs_success(&clone_logs); update.logs.extend(clone_logs); @@ -94,19 +127,30 @@ impl Resolve for State { }; if clone_success { - match periphery - .request(requests::Build { - build: build.clone(), - }) - .await - .context("failed at call to periphery to build") - { + let res = tokio::select! { + _ = cancel.cancelled() => { + update.push_error_log("build cancelled", String::from("user cancelled build during docker build")); + self.cleanup_builder_instance(periphery, cleanup_data, &mut update) + .await; + update.finalize(); + self.update_update(update.clone()).await?; + return Ok(update) + }, + res = periphery + .request(requests::Build { + build: build.clone(), + }) => res.context("failed at call to periphery to build"), + }; + + match res { Ok(logs) => update.logs.extend(logs), Err(e) => update.logs.push(Log::error("build", format!("{e:#?}"))), }; } - if all_logs_success(&update.logs) { + update.finalize(); + + if update.success { let _ = self .db .builds @@ -121,18 +165,17 @@ impl Resolve for State { .await; } - // CLEANUP AND FINALIZE UPDATE + cancel.cancel(); self.cleanup_builder_instance(periphery, cleanup_data, &mut update) .await; - self.handle_post_build_redeploy(&build.id, &mut update) - .await; - - update.finalize(); - self.update_update(update.clone()).await?; + if update.success { + self.handle_post_build_redeploy(&build.id).await; + } + Ok(update) }; @@ -156,6 +199,20 @@ impl Resolve for State { } } +#[async_trait] +impl Resolve for State { + async fn resolve( + &self, + CancelBuild { build_id }: CancelBuild, + user: RequestUser, + ) -> anyhow::Result { + self.get_build_check_permissions(&build_id, &user, PermissionLevel::Execute) + .await?; + self.build_cancel.sender.lock().await.send(build_id)?; + Ok(CancelBuildResponse {}) + } +} + const BUILDER_POLL_RATE_SECS: u64 = 2; const BUILDER_POLL_MAX_TRIES: usize = 30; @@ -300,7 +357,7 @@ impl State { } } - async fn handle_post_build_redeploy(&self, build_id: &str, update: &mut Update) { + async fn handle_post_build_redeploy(&self, build_id: &str) { let redeploy_deployments = self .db .deployments @@ -354,25 +411,6 @@ impl State { Err(e) => redeploy_failures.push(format!("{id}: {e:#?}")), } } - - if !redeploys.is_empty() { - update.logs.push(Log::simple( - "redeploy", - format!("redeployed deployments: {}", redeploys.join(", ")), - )) - } - - if !redeploy_failures.is_empty() { - update.logs.push(Log::simple( - "redeploy failures", - redeploy_failures.join("\n"), - )) - } - } else if let Err(e) = redeploy_deployments { - update.logs.push(Log::simple( - "redeploys failed", - format!("failed to get deployments to redeploy: {e:#?}"), - )) } } } diff --git a/core/src/requests/write/alerter.rs b/core/src/requests/write/alerter.rs index fa33b09ba..c0712febb 100644 --- a/core/src/requests/write/alerter.rs +++ b/core/src/requests/write/alerter.rs @@ -5,6 +5,7 @@ use monitor_types::{ monitor_timestamp, requests::write::{CopyAlerter, CreateAlerter, DeleteAlerter, UpdateAlerter}, }; +use mungos::mongodb::bson::{doc, to_bson}; use resolver_api::Resolve; use crate::{auth::RequestUser, helpers::make_update, state::State}; @@ -157,6 +158,29 @@ impl Resolve for State { UpdateAlerter { id, config }: UpdateAlerter, user: RequestUser, ) -> anyhow::Result { - todo!() + let alerter = self + .get_alerter_check_permissions(&id, &user, PermissionLevel::Update) + .await?; + + let mut update = make_update(&alerter, Operation::UpdateAlerter, &user); + + update.push_simple_log("alerter update", serde_json::to_string_pretty(&config)?); + + let config = alerter.config.merge_partial(config); + + self.db + .alerters + .update_one( + &id, + mungos::Update::Set(doc! { "config": to_bson(&config)? }), + ) + .await?; + + let alerter = self.get_alerter(&id).await?; + + update.finalize(); + self.add_update(update).await?; + + Ok(alerter) } } diff --git a/core/src/requests/write/build.rs b/core/src/requests/write/build.rs index bd271aa0c..e6da5ace3 100644 --- a/core/src/requests/write/build.rs +++ b/core/src/requests/write/build.rs @@ -3,7 +3,7 @@ use async_trait::async_trait; use monitor_types::{ entities::{ build::{Build, BuildBuilderConfig}, - update::{Log, ResourceTarget, Update, UpdateStatus}, + update::{Log, UpdateStatus}, Operation, PermissionLevel, }, monitor_timestamp, @@ -147,22 +147,9 @@ impl Resolve for State { "create build", format!("created build\nid: {}\nname: {}", build.id, build.name), ); - let update = Update { - target: ResourceTarget::Build(build_id), - operation: Operation::CreateBuild, - start_ts, - end_ts: Some(monitor_timestamp()), - operator: user.id.clone(), - success: true, - logs: vec![ - Log::simple( - "create build", - format!("created build\nid: {}\nname: {}", build.id, build.name), - ), - Log::simple("config", format!("{:#?}", build.config)), - ], - ..Default::default() - }; + update.push_simple_log("config", serde_json::to_string_pretty(&build)?); + + update.finalize(); self.add_update(update).await?; @@ -185,18 +172,8 @@ impl Resolve for State { .get_build_check_permissions(&id, &user, PermissionLevel::Update) .await?; - let start_ts = monitor_timestamp(); - - let mut update = Update { - target: ResourceTarget::Build(id.clone()), - operation: Operation::DeleteBuild, - start_ts, - operator: user.id.clone(), - success: true, - status: UpdateStatus::InProgress, - ..Default::default() - }; - + let mut update = make_update(&build, Operation::DeleteBuild, &user); + update.status = UpdateStatus::InProgress; update.id = self.add_update(update.clone()).await?; let res = self @@ -235,8 +212,6 @@ impl Resolve for State { .await?; let inner = || async move { - let start_ts = monitor_timestamp(); - if let Some(builder) = &config.builder { match builder { BuildBuilderConfig::Server { server_id } => { @@ -272,38 +247,28 @@ impl Resolve for State { self.db .builds .update_one( - &id, + &build.id, mungos::Update::Set(doc! { "config": to_bson(&config)? }), ) .await .context("failed to update build on database")?; - let update = Update { - operation: Operation::UpdateBuild, - target: ResourceTarget::Build(id.clone()), - start_ts, - end_ts: Some(monitor_timestamp()), - status: UpdateStatus::Complete, - logs: vec![Log::simple( - "build update", - serde_json::to_string_pretty(&config).unwrap(), - )], - operator: user.id.clone(), - success: true, - version: config.version.unwrap_or_default(), - ..Default::default() - }; + let mut update = make_update(&build, Operation::UpdateBuild, &user); + + update.push_simple_log("build update", serde_json::to_string_pretty(&config)?); + + update.finalize(); self.add_update(update).await?; - let build = self.get_build(&id).await?; + let build = self.get_build(&build.id).await?; anyhow::Ok(build) }; self.action_states .build - .update_entry(build.id.clone(), |entry| { + .update_entry(id.clone(), |entry| { entry.updating = true; }) .await; @@ -312,7 +277,7 @@ impl Resolve for State { self.action_states .build - .update_entry(build.id, |entry| { + .update_entry(id, |entry| { entry.updating = false; }) .await; diff --git a/core/src/state.rs b/core/src/state.rs index 9716e74f9..caaded8fb 100644 --- a/core/src/state.rs +++ b/core/src/state.rs @@ -5,7 +5,7 @@ use axum::Extension; use monitor_types::{ entities::{ build::BuildActionState, deployment::DeploymentActionState, repo::RepoActionState, - server::ServerActionState, + server::ServerActionState, update::Update, }, requests::auth::GetLoginOptionsResponse, }; @@ -13,11 +13,11 @@ use simple_logger::SimpleLogger; use crate::{ auth::{GithubOauthClient, GoogleOauthClient, JwtClient}, + cache::Cache, + channel::BroadcastChannel, config::{CoreConfig, Env}, db::DbClient, - helpers::Cache, monitoring::{CachedDeploymentStatus, CachedServerStatus}, - ws::UpdateWsChannel, }; pub type StateExtension = Extension>; @@ -26,7 +26,6 @@ pub struct State { pub env: Env, pub config: CoreConfig, pub db: DbClient, - pub update: UpdateWsChannel, // auth pub jwt: JwtClient, @@ -40,6 +39,10 @@ pub struct State { // cached responses pub login_options_response: String, + + // channels + pub build_cancel: BroadcastChannel, // build id to cancel + pub update: BroadcastChannel, } impl State { @@ -67,7 +70,8 @@ impl State { action_states: Default::default(), deployment_status_cache: Default::default(), server_status_cache: Default::default(), - update: UpdateWsChannel::new(), + update: BroadcastChannel::new(100), + build_cancel: BroadcastChannel::new(10), config, } .into(); diff --git a/core/src/ws.rs b/core/src/ws.rs index 8ae9b9dc8..c5012dfe7 100644 --- a/core/src/ws.rs +++ b/core/src/ws.rs @@ -10,18 +10,12 @@ use axum::{ }; use futures::{SinkExt, StreamExt}; use monitor_types::entities::{ - update::{ResourceTarget, ResourceTargetVariant, Update}, + update::{ResourceTarget, ResourceTargetVariant}, user::User, PermissionLevel, }; use serde_json::json; -use tokio::{ - select, - sync::{ - broadcast::{self, Receiver, Sender}, - Mutex, - }, -}; +use tokio::select; use tokio_util::sync::CancellationToken; use crate::{ @@ -29,30 +23,12 @@ use crate::{ state::{State, StateExtension}, }; -pub type UpdateWsSender = Mutex>; -pub type UpdateWsReciever = Receiver; - pub fn router() -> Router { Router::new().route("/update", get(ws_handler)) } -pub struct UpdateWsChannel { - pub sender: UpdateWsSender, - pub reciever: UpdateWsReciever, -} - -impl UpdateWsChannel { - pub fn new() -> UpdateWsChannel { - let (sender, reciever) = broadcast::channel(16); - UpdateWsChannel { - sender: Mutex::new(sender), - reciever, - } - } -} - async fn ws_handler(state: StateExtension, ws: WebSocketUpgrade) -> impl IntoResponse { - let mut reciever = state.update.reciever.resubscribe(); + let mut receiver = state.update.receiver.resubscribe(); ws.on_upgrade(|socket| async move { let login_res = state.ws_login(socket).await; if login_res.is_none() { @@ -66,7 +42,7 @@ async fn ws_handler(state: StateExtension, ws: WebSocketUpgrade) -> impl IntoRes loop { let update = select! { _ = cancel_clone.cancelled() => break, - update = reciever.recv() => {update.expect("failed to recv update msg")} + update = receiver.recv() => {update.expect("failed to recv update msg")} }; let user = state.db.users.find_one_by_id(&user.id).await; if user.is_err() diff --git a/lib/types/src/entities/update.rs b/lib/types/src/entities/update.rs index a7700eac7..147064015 100644 --- a/lib/types/src/entities/update.rs +++ b/lib/types/src/entities/update.rs @@ -42,6 +42,10 @@ impl Update { self.logs.push(Log::error(stage, msg)); } + pub fn in_progress(&mut self) { + self.status = UpdateStatus::InProgress; + } + pub fn finalize(&mut self) { self.success = all_logs_success(&self.logs); self.end_ts = Some(monitor_timestamp()); diff --git a/lib/types/src/requests/execute/build.rs b/lib/types/src/requests/execute/build.rs index 4edb03b89..83e0a2f4e 100644 --- a/lib/types/src/requests/execute/build.rs +++ b/lib/types/src/requests/execute/build.rs @@ -12,3 +12,16 @@ use crate::entities::update::Update; pub struct RunBuild { pub build_id: String, } + +// + +#[typeshare] +#[derive(Serialize, Deserialize, Debug, Clone, Request)] +#[response(CancelBuildResponse)] +pub struct CancelBuild { + pub build_id: String, +} + +#[typeshare] +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct CancelBuildResponse {}