diff --git a/Cargo.lock b/Cargo.lock index ecfdbae..7d194c2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2172,6 +2172,7 @@ dependencies = [ "hyper 1.2.0", "hyper-util", "include_dir", + "itertools 0.12.0", "lapdev-common", "lapdev-conductor", "lapdev-db", diff --git a/lapdev-api/Cargo.toml b/lapdev-api/Cargo.toml index ae6aa84..9dce0c4 100644 --- a/lapdev-api/Cargo.toml +++ b/lapdev-api/Cargo.toml @@ -7,6 +7,7 @@ license.workspace = true description.workspace = true [dependencies] +itertools.workspace = true rustls-webpki.workspace = true include_dir.workspace = true toml.workspace = true diff --git a/lapdev-api/pages/not_authorised.html b/lapdev-api/pages/not_authorised.html index d61e67c..fe03705 100644 --- a/lapdev-api/pages/not_authorised.html +++ b/lapdev-api/pages/not_authorised.html @@ -5,9 +5,9 @@ -
+
- + diff --git a/lapdev-api/pages/not_forwarded.html b/lapdev-api/pages/not_forwarded.html index 7bdfafe..9e94084 100644 --- a/lapdev-api/pages/not_forwarded.html +++ b/lapdev-api/pages/not_forwarded.html @@ -5,9 +5,9 @@ -
+
- + diff --git a/lapdev-api/pages/not_found.html b/lapdev-api/pages/not_found.html index ae5b08d..cc58110 100644 --- a/lapdev-api/pages/not_found.html +++ b/lapdev-api/pages/not_found.html @@ -5,9 +5,9 @@ -
+
- + diff --git a/lapdev-api/pages/not_running.html b/lapdev-api/pages/not_running.html index c940791..5757809 100644 --- a/lapdev-api/pages/not_running.html +++ b/lapdev-api/pages/not_running.html @@ -5,9 +5,9 @@ -
+
- + diff --git a/lapdev-api/src/router.rs b/lapdev-api/src/router.rs index d90a5bc..6d7cab9 100644 --- a/lapdev-api/src/router.rs +++ b/lapdev-api/src/router.rs @@ -168,6 +168,14 @@ fn v1_api_routes(additional_router: Option>) -> Router, ssh_proxy_port: Option, ssh_proxy_display_port: Option, + force_osuser: Option, } #[derive(Parser)] @@ -87,7 +88,8 @@ async fn run( .ok_or_else(|| anyhow!("can't find database url in your config file"))?; let db = DbApi::new(&db_url, cli.no_migration).await?; - let conductor = Conductor::new(LAPDEV_VERSION, db.clone(), data_folder).await?; + let conductor = + Conductor::new(LAPDEV_VERSION, db.clone(), data_folder, config.force_osuser).await?; let ssh_proxy_port = config.ssh_proxy_port.unwrap_or(2222); let ssh_proxy_display_port = config.ssh_proxy_display_port.unwrap_or(2222); diff --git a/lapdev-api/src/workspace.rs b/lapdev-api/src/workspace.rs index fc3784f..c43df07 100644 --- a/lapdev-api/src/workspace.rs +++ b/lapdev-api/src/workspace.rs @@ -1,4 +1,4 @@ -use std::{collections::HashMap, str::FromStr}; +use std::{cmp::Ordering, collections::HashMap, str::FromStr}; use anyhow::Result; use axum::{ @@ -9,11 +9,12 @@ use axum::{ use axum_extra::{headers::Cookie, TypedHeader}; use chrono::Utc; use hyper::StatusCode; +use itertools::Itertools; use lapdev_common::{ AuditAction, AuditResourceKind, NewWorkspace, RepoBuildResult, UpdateWorkspacePort, - WorkspaceInfo, WorkspacePort, WorkspaceService, WorkspaceStatus, + WorkspaceInfo, WorkspacePort, WorkspaceService, WorkspaceStatus, WorkspaceUpdateEvent, }; -use lapdev_db::entities; +use lapdev_db::{api::LAPDEV_PIN_UNPIN_ERROR, entities}; use lapdev_rpc::error::ApiError; use sea_orm::{ ActiveModelTrait, ActiveValue, ColumnTrait, EntityTrait, QueryFilter, TransactionTrait, @@ -105,8 +106,14 @@ pub async fn all_workspaces( created_at: w.created_at, hostname, build_error: build_result.and_then(|r| r.error), + pinned: w.pinned, }) }) + .sorted_by(|a, b| match (a.pinned, b.pinned) { + (true, true) | (false, false) => b.created_at.cmp(&a.created_at), + (true, false) => Ordering::Less, + (false, true) => Ordering::Greater, + }) .collect(); Ok(Json(workspaces)) } @@ -201,6 +208,7 @@ pub async fn get_workspace( created_at: ws.created_at, hostname, build_error: build_result.and_then(|r| r.error), + pinned: ws.pinned, }; Ok(Json(info)) } @@ -273,6 +281,174 @@ pub async fn stop_workspace( Ok(StatusCode::NO_CONTENT.into_response()) } +pub async fn pin_workspace( + TypedHeader(cookie): TypedHeader, + Path((org_id, workspace_name)): Path<(Uuid, String)>, + State(state): State, + info: RequestInfo, +) -> Result { + let user = state.authenticate(&cookie).await?; + state + .db + .get_organization_member(user.id, org_id) + .await + .map_err(|_| ApiError::Unauthorized)?; + let ws = state + .db + .get_workspace_by_name(&workspace_name) + .await + .map_err(|_| ApiError::InvalidRequest("workspace name doesn't exist".to_string()))?; + if ws.user_id != user.id { + return Err(ApiError::Unauthorized); + } + + let org = state.db.get_organization(ws.organization_id).await?; + if org.running_workspace_limit > 0 { + return Err(ApiError::InvalidRequest( + state + .db + .get_config(LAPDEV_PIN_UNPIN_ERROR) + .await + .unwrap_or_else(|_| "You can't pin/unpin workspaces".to_string()), + )); + } + + if ws.compose_parent.is_some() { + return Err(ApiError::InvalidRequest( + "you can only pin the main workspace".to_string(), + )); + } + + if ws.pinned { + return Err(ApiError::InvalidRequest( + "workspace is already pinned".to_string(), + )); + } + + let now = Utc::now(); + let txn = state.db.conn.begin().await?; + state + .conductor + .enterprise + .insert_audit_log( + &txn, + now.into(), + ws.user_id, + ws.organization_id, + AuditResourceKind::Workspace.to_string(), + ws.id, + format!("{} pin", ws.name), + AuditAction::WorkspaceUpdate.to_string(), + info.ip.clone(), + info.user_agent.clone(), + ) + .await?; + let ws = entities::workspace::ActiveModel { + id: ActiveValue::Set(ws.id), + pinned: ActiveValue::Set(true), + ..Default::default() + } + .update(&txn) + .await?; + txn.commit().await?; + + // send a status update to trigger frontend update + state + .conductor + .add_workspace_update_event( + Some(ws.user_id), + ws.id, + WorkspaceUpdateEvent::Status(WorkspaceStatus::from_str(&ws.status)?), + ) + .await; + + Ok(StatusCode::NO_CONTENT.into_response()) +} + +pub async fn unpin_workspace( + TypedHeader(cookie): TypedHeader, + Path((org_id, workspace_name)): Path<(Uuid, String)>, + State(state): State, + info: RequestInfo, +) -> Result { + let user = state.authenticate(&cookie).await?; + state + .db + .get_organization_member(user.id, org_id) + .await + .map_err(|_| ApiError::Unauthorized)?; + let ws = state + .db + .get_workspace_by_name(&workspace_name) + .await + .map_err(|_| ApiError::InvalidRequest("workspace name doesn't exist".to_string()))?; + if ws.user_id != user.id { + return Err(ApiError::Unauthorized); + } + + let org = state.db.get_organization(ws.organization_id).await?; + if org.running_workspace_limit > 0 { + return Err(ApiError::InvalidRequest( + state + .db + .get_config(LAPDEV_PIN_UNPIN_ERROR) + .await + .unwrap_or_else(|_| "You can't pin/unpin workspaces".to_string()), + )); + } + + if ws.compose_parent.is_some() { + return Err(ApiError::InvalidRequest( + "you can only unpin the main workspace".to_string(), + )); + } + + if !ws.pinned { + return Err(ApiError::InvalidRequest( + "workspace is not pinned".to_string(), + )); + } + + let now = Utc::now(); + let txn = state.db.conn.begin().await?; + state + .conductor + .enterprise + .insert_audit_log( + &txn, + now.into(), + ws.user_id, + ws.organization_id, + AuditResourceKind::Workspace.to_string(), + ws.id, + format!("{} unpin", ws.name), + AuditAction::WorkspaceUpdate.to_string(), + info.ip.clone(), + info.user_agent.clone(), + ) + .await?; + let ws = entities::workspace::ActiveModel { + id: ActiveValue::Set(ws.id), + pinned: ActiveValue::Set(false), + ..Default::default() + } + .update(&txn) + .await?; + txn.commit().await?; + + // send a status update to trigger frontend update + state + .conductor + .add_workspace_update_event( + Some(ws.user_id), + ws.id, + WorkspaceUpdateEvent::Status(WorkspaceStatus::from_str(&ws.status)?), + ) + .await; + + Ok(StatusCode::NO_CONTENT.into_response()) +} + pub async fn rebuild_workspace( TypedHeader(cookie): TypedHeader, Path((org_id, workspace_name)): Path<(Uuid, String)>, diff --git a/lapdev-common/src/lib.rs b/lapdev-common/src/lib.rs index 729a9b6..318d55f 100644 --- a/lapdev-common/src/lib.rs +++ b/lapdev-common/src/lib.rs @@ -332,6 +332,7 @@ pub struct WorkspaceInfo { pub created_at: DateTime, pub hostname: String, pub build_error: Option, + pub pinned: bool, } #[derive(Serialize, Deserialize, Debug, Clone, Hash, Eq, PartialEq)] diff --git a/lapdev-conductor/src/scheduler.rs b/lapdev-conductor/src/scheduler.rs index 38ce62a..b63ada3 100644 --- a/lapdev-conductor/src/scheduler.rs +++ b/lapdev-conductor/src/scheduler.rs @@ -429,6 +429,7 @@ mod tests { deleted_at: ActiveValue::Set(None), env: ActiveValue::Set(None), last_inactivity: ActiveValue::Set(None), + pinned: ActiveValue::Set(false), } .insert(&txn) .await diff --git a/lapdev-conductor/src/server.rs b/lapdev-conductor/src/server.rs index 52fb5f3..72110b8 100644 --- a/lapdev-conductor/src/server.rs +++ b/lapdev-conductor/src/server.rs @@ -115,11 +115,17 @@ pub struct Conductor { Mutex>>>, >, pub enterprise: Arc, + pub force_osuser: Option, pub db: DbApi, } impl Conductor { - pub async fn new(version: &str, db: DbApi, data_folder: PathBuf) -> Result { + pub async fn new( + version: &str, + db: DbApi, + data_folder: PathBuf, + force_osuser: Option, + ) -> Result { tokio::fs::create_dir_all(data_folder.join("projects")) .await .with_context(|| format!("trying to create {:?}", data_folder.join("projects")))?; @@ -147,6 +153,7 @@ impl Conductor { cpu_overcommit: Arc::new(RwLock::new(cpu_overcommit)), all_workspace_updates: Default::default(), enterprise, + force_osuser, db, }; @@ -720,7 +727,7 @@ impl Conductor { .map_err(|_| ApiError::NoAvailableWorkspaceHost)?; txn.commit().await?; - let osuser = org_id.to_string().replace('-', ""); + let osuser = self.get_osuser(org_id); let id = uuid::Uuid::new_v4(); let ws_client = { self.rpcs.lock().await.get(&host.id).cloned() } .ok_or_else(|| anyhow!("can't find the workspace host rpc client"))?; @@ -878,6 +885,14 @@ impl Conductor { Ok(result) } + fn get_osuser(&self, org: Uuid) -> String { + if let Some(osuser) = self.force_osuser.clone() { + osuser + } else { + org.to_string().replace('-', "") + } + } + pub async fn create_project_prebuild( &self, user: &entities::user::Model, @@ -889,7 +904,7 @@ impl Conductor { ) -> Result { let id = uuid::Uuid::new_v4(); - let osuser = project.organization_id.to_string().replace('-', ""); + let osuser = self.get_osuser(project.organization_id); let machine_type_id = ws .map(|ws| ws.machine_type_id) @@ -1101,7 +1116,7 @@ impl Conductor { ) -> Result { let name = format!("{}-{}", repo.name, rand_string(12)); let (id_rsa, public_key) = self.generate_key_pair()?; - let osuser = org.id.to_string().replace('-', ""); + let osuser = self.get_osuser(org.id); self.enterprise .check_organization_limit(org, user.id) @@ -1199,6 +1214,7 @@ impl Conductor { build_output: ActiveValue::Set(None), is_compose: ActiveValue::Set(false), compose_parent: ActiveValue::Set(None), + pinned: ActiveValue::Set(false), }; let ws = ws.insert(&txn).await?; self.enterprise @@ -1919,6 +1935,7 @@ impl Conductor { build_output: ActiveValue::Set(Some(build_output.clone())), is_compose: ActiveValue::Set(is_compose), compose_parent: ActiveValue::Set(Some(ws.id)), + pinned: ActiveValue::Set(false), } .insert(&self.db.conn) .await?; diff --git a/lapdev-dashboard/src/audit_log.rs b/lapdev-dashboard/src/audit_log.rs index 915f088..462b144 100644 --- a/lapdev-dashboard/src/audit_log.rs +++ b/lapdev-dashboard/src/audit_log.rs @@ -144,10 +144,10 @@ pub fn AuditLogView() -> impl IntoView { view! {
-
+
Audit Log
-

{"View your organization's audit logs"}

+

{"View your organization's audit logs"}

diff --git a/lapdev-dashboard/src/cluster.rs b/lapdev-dashboard/src/cluster.rs index 9ffa91a..46cb595 100644 --- a/lapdev-dashboard/src/cluster.rs +++ b/lapdev-dashboard/src/cluster.rs @@ -50,17 +50,17 @@ pub fn WorkspaceHostView() -> impl IntoView { }); view! {
-
+
Workspace Hosts
-

{"Manage the workspace hosts for the cluster"}

+

{"Manage the workspace hosts for the cluster"}

-
@@ -68,7 +68,7 @@ pub fn WorkspaceHostView() -> impl IntoView { prop:value={move || host_filter.get()} on:input=move |ev| { host_filter.set(event_target_value(&ev)); } type="text" - class="bg-white block w-full p-2 pl-10 text-sm text-gray-900 border border-gray-300 rounded-lg bg-gray-50 focus:ring-blue-500 focus:border-blue-500 dark:bg-gray-700 dark:border-gray-600 dark:placeholder-gray-400 dark:text-white dark:focus:ring-blue-500 dark:focus:border-blue-500" + class="bg-white block w-full p-2 pl-10 text-sm text-gray-900 border border-gray-300 rounded-lg bg-gray-50 focus:ring-blue-500 focus:border-blue-500" placeholder="Filter Workspace Hosts" />
@@ -76,7 +76,7 @@ pub fn WorkspaceHostView() -> impl IntoView {
- +
@@ -1182,7 +1182,7 @@ fn MachineTypeControl( on:focusout=on_focusout > -
@@ -296,12 +296,12 @@ fn DatePanel(
- -
-
- { move || workspace.status.to_string() } -
-
- +
+
+ { move || workspace.status.to_string() } +
+
+
+ +
+ +
impl IntoView {
@@ -1222,6 +1341,11 @@ pub fn WorkspaceDetails() -> impl IntoView { .map(|machine_type| format!("{} - {} vCPUs, {}GB memory, {}GB disk",machine_type.name, machine_type.cpu, machine_type.memory, machine_type.disk)) } + + + {"Pinned:"} + { info.pinned } +
- +
{ move || if let Some(error) = error.get() { view! { @@ -1583,15 +1716,15 @@ pub fn WorkspacePortsView(name: String, workspace_hostname: String) -> impl Into { move || if let Some(error) = error.get() { view! { -
- { error } +
+ { error }
}.into_view() } else { view!{}.into_view() }} -
+
Port
shared @@ -1660,7 +1793,7 @@ fn WorkspacePortView( @@ -1669,7 +1802,7 @@ fn WorkspacePortView( @@ -1679,7 +1812,7 @@ fn WorkspacePortView( Open diff --git a/lapdev-db/src/api.rs b/lapdev-db/src/api.rs index 5a5eb77..7bd2492 100644 --- a/lapdev-db/src/api.rs +++ b/lapdev-db/src/api.rs @@ -23,6 +23,7 @@ use crate::{entities, migration::Migrator}; use super::entities::workspace; pub const LAPDEV_CLUSTER_NOT_INITIATED: &str = "lapdev-cluster-not-initiated"; +pub const LAPDEV_PIN_UNPIN_ERROR: &str = "lapdev-pin-unpin-error"; const LAPDEV_API_AUTH_TOKEN_KEY: &str = "lapdev-api-auth-token-key"; const LAPDEV_DEFAULT_USAGE_LIMIT: &str = "lapdev-default-org-usage-limit"; const LAPDEV_DEFAULT_RUNNING_WORKSPACE_LIMIT: &str = "lapdev-default-org-running-workspace-limit"; @@ -237,6 +238,7 @@ impl DbApi { .filter(entities::workspace::Column::HostId.eq(ws_host_id)) .filter(entities::workspace::Column::DeletedAt.is_null()) .filter(entities::workspace::Column::Status.eq(WorkspaceStatus::Stopped.to_string())) + .filter(entities::workspace::Column::Pinned.eq(false)) .filter(entities::workspace::Column::UpdatedAt.lt(last_updated_at)) .all(&self.conn) .await?; diff --git a/lapdev-db/src/entities/workspace.rs b/lapdev-db/src/entities/workspace.rs index 5a46368..bd1269f 100644 --- a/lapdev-db/src/entities/workspace.rs +++ b/lapdev-db/src/entities/workspace.rs @@ -37,6 +37,7 @@ pub struct Model { pub auto_stop: Option, pub is_compose: bool, pub compose_parent: Option, + pub pinned: bool, } #[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] diff --git a/lapdev-db/src/migration/m20231106_100804_create_workspace_table.rs b/lapdev-db/src/migration/m20231106_100804_create_workspace_table.rs index bcb21cd..7e03b68 100644 --- a/lapdev-db/src/migration/m20231106_100804_create_workspace_table.rs +++ b/lapdev-db/src/migration/m20231106_100804_create_workspace_table.rs @@ -56,6 +56,7 @@ impl MigrationTrait for Migration { .col(ColumnDef::new(Workspace::AutoStop).integer()) .col(ColumnDef::new(Workspace::IsCompose).boolean().not_null()) .col(ColumnDef::new(Workspace::ComposeParent).uuid()) + .col(ColumnDef::new(Workspace::Pinned).boolean().not_null()) .foreign_key( ForeignKey::create() .from_tbl(Workspace::Table) @@ -171,4 +172,5 @@ pub enum Workspace { LastInactivity, IsCompose, ComposeParent, + Pinned, } diff --git a/lapdev-enterprise/src/auto_start_stop.rs b/lapdev-enterprise/src/auto_start_stop.rs index 655c82e..68397f3 100644 --- a/lapdev-enterprise/src/auto_start_stop.rs +++ b/lapdev-enterprise/src/auto_start_stop.rs @@ -255,6 +255,7 @@ pub mod tests { build_output: ActiveValue::Set(None), is_compose: ActiveValue::Set(false), compose_parent: ActiveValue::Set(None), + pinned: ActiveValue::Set(false), } .insert(&enterprise.auto_start_stop.db.conn) .await @@ -301,6 +302,7 @@ pub mod tests { build_output: ActiveValue::Set(None), is_compose: ActiveValue::Set(false), compose_parent: ActiveValue::Set(None), + pinned: ActiveValue::Set(false), } .insert(&enterprise.auto_start_stop.db.conn) .await diff --git a/lapdev-enterprise/src/quota.rs b/lapdev-enterprise/src/quota.rs index f51c3f9..b27358a 100644 --- a/lapdev-enterprise/src/quota.rs +++ b/lapdev-enterprise/src/quota.rs @@ -429,6 +429,7 @@ pub mod tests { build_output: ActiveValue::Set(None), is_compose: ActiveValue::Set(false), compose_parent: ActiveValue::Set(None), + pinned: ActiveValue::Set(false), }; let ws = ws.insert(&db.conn).await?; Ok(ws) diff --git a/lapdev-proxy-ssh/src/proxy.rs b/lapdev-proxy-ssh/src/proxy.rs index 0b90600..6ab395a 100644 --- a/lapdev-proxy-ssh/src/proxy.rs +++ b/lapdev-proxy-ssh/src/proxy.rs @@ -10,7 +10,7 @@ use russh::{ server::{Auth, Msg, Session}, Channel, ChannelMsg, }; -use tracing::debug; +use tracing::{debug, info}; use crate::client::handle_client_msg; @@ -187,6 +187,11 @@ impl russh::server::Handler for SshProxyHandler { ) .await?; + info!( + "channel {} open direct tcpip from {originator_address}:{originator_port} to {host_to_connect}:{port_to_connect}", + channel.id() + ); + let server_handle = session.handle(); tokio::spawn(async move { @@ -205,6 +210,8 @@ impl russh::server::Handler for SshProxyHandler { let address = address.to_string(); let port = *port; + info!("channel {} tcpip forward to {address}:{port}", self.id); + if let Some(ws_session) = self.ws_session.as_mut() { ws_session.handle.tcpip_forward(address, port).await?; } @@ -222,6 +229,11 @@ impl russh::server::Handler for SshProxyHandler { ) -> Result { let address = address.to_string(); + info!( + "channel {} cancel tcpip forward to {address}:{port}", + self.id + ); + if let Some(ws_session) = self.ws_session.as_mut() { ws_session .handle @@ -266,31 +278,61 @@ async fn forward_server_client( server_handle: russh::server::Handle, ) -> Result<()> { let channel_id = channel.id(); - debug!("proxy connection started {channel_id}"); + info!("proxy connection started channel {channel_id}"); let ws_channel_id = ws_channel.id(); + let mut channel_eof = false; + let mut ws_channel_eof = false; loop { tokio::select! { msg = channel.wait() => { if let Some(msg) = msg { + if channel_eof && matches!(msg, ChannelMsg::Data {..} | ChannelMsg::ExtendedData {..}) { + // channel received eof, so we ignore data + continue; + } match msg { ChannelMsg::Close => { - debug!("server received close msg"); + info!("channel {channel_id} server received close msg"); + ws_channel.close().await?; break; - }, + } + ChannelMsg::Eof => { + info!("channel {channel_id} server received eof msg"); + ws_channel.eof().await?; + channel_eof = true; + } _ => { handle_server_msg(&ws_channel, msg).await?; } } } else { - debug!("server msg channel closed"); + info!("channel {channel_id} server msg channel closed"); break; } } msg = ws_channel.wait() => { if let Some(msg) = msg { - handle_client_msg(ws_channel_id, &mut channel, &server_handle, msg).await?; + if ws_channel_eof && matches!(msg, ChannelMsg::Data {..} | ChannelMsg::ExtendedData {..}) { + // ws_channel received eof, so we ignore data + continue; + } + match msg { + ChannelMsg::Close => { + info!("channel {channel_id} client received close msg"); + channel.close().await?; + break; + } + ChannelMsg::Eof => { + info!("channel {channel_id} client received eof msg"); + channel.eof().await?; + ws_channel_eof = true; + } + _ => { + handle_client_msg(ws_channel_id, &mut channel, &server_handle, msg).await?; + } + } } else { - debug!("client msg channel closed"); + info!("channel {channel_id} client msg channel closed"); break; } } @@ -298,7 +340,7 @@ async fn forward_server_client( } let _ = channel.close().await; let _ = ws_channel.close().await; - debug!("proxy connection closed {channel_id}"); + info!("proxy connection closed channel {channel_id}"); Ok(()) }