From e33f9a59f875edf1240ca80c1014235296ff3cbf Mon Sep 17 00:00:00 2001 From: yuzu Date: Sat, 12 Jul 2025 22:17:26 +0000 Subject: add additional endpoints; change out mutexes for rwlocks git-svn-id: svn+ssh://diminuette.aengel.lesbianunix.dev/salaryman/trunk@15 b9215c17-b818-4693-b096-d1e41a411fef --- src/server/context.rs | 47 +++++++ src/server/endpoints.rs | 359 ++++++++++++++++++++++++++++++++++++++++++++++++ src/server/main.rs | 162 ++++++++++++++++++++++ 3 files changed, 568 insertions(+) create mode 100644 src/server/context.rs create mode 100644 src/server/endpoints.rs create mode 100644 src/server/main.rs (limited to 'src/server') diff --git a/src/server/context.rs b/src/server/context.rs new file mode 100644 index 0000000..132e2dc --- /dev/null +++ b/src/server/context.rs @@ -0,0 +1,47 @@ +use super::Config; +use salaryman::service::{Service, ServiceConf}; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::RwLock; + +pub struct SalarymanService { + pub config: ServiceConf, + pub service: Arc>, +} +impl SalarymanService { + pub fn new() -> Self { + Self { + config: ServiceConf::new(), + service: Arc::new(RwLock::new(Service::new())), + } + } + pub fn from_parts(config: ServiceConf, service: Arc>) -> Self { + Self { config, service } + } +} + +pub struct SalarymanDContext { + pub services: RwLock>>, + pub save_file: PathBuf, + pub config: Arc>, +} +impl SalarymanDContext { + pub fn new() -> Self { + Self { + services: RwLock::new(Vec::new()), + save_file: PathBuf::from(""), + config: Arc::new(RwLock::new(Config::new())), + } + } + pub fn from_parts( + services: RwLock>>, + save_file: PathBuf, + config: Arc>, + ) -> Self { + Self { + services, + save_file, + config, + } + } +} diff --git a/src/server/endpoints.rs b/src/server/endpoints.rs new file mode 100644 index 0000000..8ebc42e --- /dev/null +++ b/src/server/endpoints.rs @@ -0,0 +1,359 @@ +use super::Config; +use super::context::{SalarymanDContext, SalarymanService}; +use dropshot::{HttpError, HttpResponseOk, Path, RequestContext, TypedBody, endpoint}; +use salaryman::model::{NewService, ServicePath, StdinBuffer, UpdateConf}; +use salaryman::service::{Service, ServiceConf}; +use std::sync::Arc; +use tokio::fs::File; +use tokio::io::AsyncWriteExt; +use tokio::sync::RwLock; + +#[endpoint { + method = GET, + path = "/config", +}] +pub async fn endpoint_get_config( + rqctx: RequestContext>, +) -> Result, HttpError> { + let ctx = rqctx.context(); + let lock = ctx.config.read().await; + let conf = lock.clone(); + drop(lock); + Ok(HttpResponseOk(conf)) +} +#[endpoint { + method = PUT, + path = "/config", +}] +pub async fn endpoint_put_config( + rqctx: RequestContext>, + body: TypedBody, +) -> Result, HttpError> { + let ctx = rqctx.context(); + let update = body.into_inner(); + if let Some(addr) = update.address { + let mut lock = ctx.config.write().await; + lock.address = Some(addr); + drop(lock); + } + if let Some(port) = update.port { + let mut lock = ctx.config.write().await; + lock.port = Some(port); + drop(lock); + } + Ok(HttpResponseOk(())) +} +#[endpoint { + method = GET, + path = "/config/save" +}] +pub async fn endpoint_get_config_save( + rqctx: RequestContext>, +) -> Result, HttpError> { + let ctx = rqctx.context(); + let mut v: Vec = Vec::new(); + let rlock = ctx.services.read().await; + for i in 0..rlock.len() { + v.push(rlock[i].config.clone()); + } + drop(rlock); + let rlock = ctx.config.read().await; + let save = Config { + address: rlock.address, + port: rlock.port, + user: rlock.user.clone(), + service: v, + }; + drop(rlock); + let mut f = match File::open(&ctx.save_file).await { + Ok(f) => f, + Err(_) => { + return Err(HttpError::for_internal_error(String::from( + "cannot open desired file", + ))); + } + }; + let save = match toml::to_string(&save) { + Ok(s) => s, + Err(_) => { + return Err(HttpError::for_internal_error(String::from( + "cannot serialize config!", + ))); + } + }; + match f.write_all(save.as_str().as_bytes()).await { + Ok(_) => (), + Err(_) => { + return Err(HttpError::for_internal_error(String::from( + "could not write to file!", + ))); + } + } + Ok(HttpResponseOk(())) +} +#[endpoint { + method = GET, + path = "/service", +}] +pub async fn endpoint_get_services( + rqctx: RequestContext>, +) -> Result>, HttpError> { + let ret = { + let ctx = rqctx.context(); + let mut v: Vec = Vec::new(); + let lock = ctx.services.read().await; + for i in 0..lock.len() { + v.push(lock[i].config.clone()); + } + v + }; + Ok(HttpResponseOk(ret)) +} +#[endpoint { + method = POST, + path = "/service", +}] +pub async fn endpoint_post_service( + rqctx: RequestContext>, + body: TypedBody, +) -> Result, HttpError> { + let ctx = rqctx.context(); + let body = body.into_inner(); + let mut s: ServiceConf = ServiceConf::new(); + if let Some(name) = &body.name { + s.name = name.clone().to_owned(); + } else { + return Err(HttpError::for_bad_request( + None, + String::from("name field is required!"), + )); + } + if let Some(command) = &body.command { + s.command = command.clone().to_owned(); + } else { + return Err(HttpError::for_bad_request( + None, + String::from("command field is required!"), + )); + } + if let Some(args) = &body.args { + if let Some(args) = args { + s.args = Some(args.clone().to_owned()); + } + } + if let Some(dir) = &body.directory { + if let Some(dir) = dir { + s.directory = Some(dir.clone().to_owned()); + } + } + if let Some(auto) = &body.autostart { + s.autostart = auto.clone().to_owned(); + } else { + s.autostart = false; + } + let service: SalarymanService = + SalarymanService::from_parts(s.clone(), Arc::new(RwLock::new(Service::from_conf(&s)))); + if service.config.autostart { + let mut lock = service.service.write().await; + match lock.start_with_output().await { + Ok(_) => (), + Err(_) => (), + } + drop(lock); + } + let mut lock = ctx.config.write().await; + lock.service.push(s.clone()); + drop(lock); + let mut lock = ctx.services.write().await; + lock.push(Arc::new(service)); + drop(lock); + Ok(HttpResponseOk(ServiceConf::new())) +} +#[endpoint { + method = GET, + path = "/service/{service_uuid}", +}] +pub async fn endpoint_get_service( + rqctx: RequestContext>, + path_params: Path, +) -> Result, HttpError> { + let u = path_params.into_inner().service_uuid; + let ctx = rqctx.context(); + let mut service: Option> = None; + let lock = ctx.services.read().await; + for i in 0..lock.len() { + if lock[i].config.uuid == u { + service = Some(lock[i].clone()); + } else { + continue; + } + } + let s = match service { + Some(s) => s.config.clone(), + None => { + return Err(HttpError::for_unavail( + None, + String::from("Service Not Found"), + )); + } + }; + Ok(HttpResponseOk(s)) +} +#[endpoint { + method = GET, + path = "/service/{service_uuid}/start", +}] +pub async fn endpoint_start_service( + rqctx: RequestContext>, + path_params: Path, +) -> Result, HttpError> { + let u = path_params.into_inner().service_uuid; + let ctx = rqctx.context(); + let mut service: Option> = None; + let lock = ctx.services.read().await; + for i in 0..lock.len() { + if lock[i].config.uuid == u { + service = Some(lock[i].clone()); + } else { + continue; + } + } + match service { + Some(s) => { + let mut lock = s.service.write().await; + match lock.start_with_output().await { + Ok(_) => (), + Err(e) => return Err(HttpError::for_internal_error(e.to_string())), + } + } + None => { + return Err(HttpError::for_unavail( + None, + String::from("Service Not Found"), + )); + } + }; + Ok(HttpResponseOk(())) +} +#[endpoint { + method = GET, + path = "/service/{service_uuid}/stop", +}] +pub async fn endpoint_stop_service( + rqctx: RequestContext>, + path_params: Path, +) -> Result, HttpError> { + let u = path_params.into_inner().service_uuid; + let ctx = rqctx.context(); + let mut service: Option> = None; + let lock = ctx.services.read().await; + for i in 0..lock.len() { + if lock[i].config.uuid == u { + service = Some(lock[i].clone()); + } else { + continue; + } + } + match service { + Some(s) => { + let mut lock = s.service.write().await; + match lock.stop().await { + Ok(_) => (), + Err(e) => return Err(HttpError::for_internal_error(e.to_string())), + } + } + None => { + return Err(HttpError::for_unavail( + None, + String::from("Service Not Found"), + )); + } + }; + Ok(HttpResponseOk(())) +} +#[endpoint { + method = GET, + path = "/service/{service_uuid}/restart", +}] +pub async fn endpoint_restart_service( + rqctx: RequestContext>, + path_params: Path, +) -> Result, HttpError> { + let u = path_params.into_inner().service_uuid; + let ctx = rqctx.context(); + let mut service: Option> = None; + let lock = ctx.services.read().await; + for i in 0..lock.len() { + if lock[i].config.uuid == u { + service = Some(lock[i].clone()); + } else { + continue; + } + } + match service { + Some(s) => { + let mut lock = s.service.write().await; + match lock.restart_with_output().await { + Ok(_) => (), + Err(e) => return Err(HttpError::for_internal_error(e.to_string())), + } + } + None => { + return Err(HttpError::for_unavail( + None, + String::from("Service Not Found"), + )); + } + }; + Ok(HttpResponseOk(())) +} +#[endpoint { + method = PUT, + path = "/service/{service_uuid}/write" +}] +pub async fn endpoint_post_stdin( + rqctx: RequestContext>, + path_params: Path, + update: TypedBody, +) -> Result, HttpError> { + let ctx = rqctx.context(); + let stdin_str = update.into_inner(); + let u = path_params.into_inner().service_uuid; + let mut service: Option> = None; + let lock = ctx.services.read().await; + for i in 0..lock.len() { + if lock[i].config.uuid == u { + service = Some(lock[i].clone()); + } else { + continue; + } + } + match service { + Some(s) => { + let mut lock = s.service.write().await; + if lock.started().await { + let b = if let Some(endl) = stdin_str.endl { + if endl { + lock.writeln_stdin(stdin_str.stdin.clone()).await //TODO: PROPERLY HANDLE ERROR! + } else { + lock.write_stdin(stdin_str.stdin.clone()).await //TODO: PROPERLY HANDLE ERROR! + } + } else { + lock.writeln_stdin(stdin_str.stdin.clone()).await //TODO: PROPERLY HANDLE ERROR! + }; + match b { + Ok(_) => (), + Err(e) => return Err(HttpError::for_internal_error(e.to_string())), + } + } + drop(lock); + } + None => { + return Err(HttpError::for_unavail( + None, + String::from("Service Not Found"), + )); + } + } + Ok(HttpResponseOk(())) +} diff --git a/src/server/main.rs b/src/server/main.rs new file mode 100644 index 0000000..7557145 --- /dev/null +++ b/src/server/main.rs @@ -0,0 +1,162 @@ +mod context; +mod endpoints; + +use clap::Parser; +use dropshot::{ApiDescription, ConfigDropshot, ConfigLogging, ConfigLoggingLevel, ServerBuilder}; +use salaryman::service::{Service, ServiceConf}; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use tokio::{fs::read_to_string, sync::RwLock}; + +use std::{ + net::{IpAddr, SocketAddr}, + path::PathBuf, + sync::Arc, +}; + +use crate::context::{SalarymanDContext, SalarymanService}; +use crate::endpoints::{ + endpoint_get_config, endpoint_get_config_save, endpoint_get_service, endpoint_get_services, + endpoint_post_service, endpoint_post_stdin, endpoint_put_config, endpoint_restart_service, + endpoint_start_service, endpoint_stop_service, +}; + +#[derive(Parser, Debug)] +#[command(version, about, long_about = None)] +struct Args { + #[arg( + short, + long, + value_name = "FILE", + help = "config file override", + default_value = "salaryman.toml" + )] + config: PathBuf, + #[arg( + short, + long, + value_name = "ADDR", + help = "IP address to bind API to", + default_value = "127.0.0.1" + )] + address: IpAddr, + #[arg( + short, + long, + value_name = "PORT", + help = "TCP Port to bind API to", + default_value = "3080" + )] + port: u16, +} + +#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug)] +pub struct User { + pub username: String, + pub token: String, +} + +#[derive(Serialize, Deserialize, JsonSchema, Clone, Debug)] +pub struct Config { + pub address: Option, + pub port: Option, + pub user: Vec, + pub service: Vec, +} +impl Config { + pub fn new() -> Self { + Self { + address: None, + port: None, + user: Vec::new(), + service: Vec::new(), + } + } +} + +async fn load_config(file: &PathBuf) -> Result> { + let s: String = match read_to_string(file).await { + Ok(s) => s, + Err(_) => { + return Err(Box::new(std::io::Error::new( + std::io::ErrorKind::NotFound, + "cannot find config file", + ))); + } + }; + match toml::from_str(s.as_str()) { + Ok(c) => Ok(c), + Err(_) => Err(Box::new(std::io::Error::new( + std::io::ErrorKind::Other, + "unable to parse config file", + ))), + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let args = Args::parse(); + let conf: Config = load_config(&args.config).await?; + let addr = if let Some(addr) = conf.address { + addr + } else { + args.address + }; + let port = if let Some(port) = conf.port { + port + } else { + args.port + }; + let bind = SocketAddr::new(addr, port); + let services: RwLock>> = RwLock::new(Vec::new()); + for i in 0..conf.service.len() { + let mut lock = services.write().await; + lock.push(Arc::new(SalarymanService::from_parts( + conf.service[i].clone(), + Arc::new(RwLock::new(Service::from_conf(&conf.service[i]))), + ))); + drop(lock); + } + let lock = services.write().await; + for i in 0..lock.len() { + if lock[i].config.autostart { + let mut l = lock[i].service.write().await; + l.start().await?; + l.scan_stdout().await?; + l.scan_stderr().await?; + drop(l); + } + } + drop(lock); + let log_conf = ConfigLogging::StderrTerminal { + level: ConfigLoggingLevel::Info, + }; + let log = log_conf.to_logger("smd")?; + let ctx = Arc::new(SalarymanDContext::from_parts( + services, + args.config, + Arc::new(RwLock::new(conf)), + )); + let config = ConfigDropshot { + bind_address: bind, + ..Default::default() + }; + let mut api = ApiDescription::new(); + api.register(endpoint_get_services)?; + api.register(endpoint_get_service)?; + api.register(endpoint_start_service)?; + api.register(endpoint_stop_service)?; + api.register(endpoint_restart_service)?; + api.register(endpoint_post_stdin)?; + api.register(endpoint_post_service)?; + api.register(endpoint_get_config)?; + api.register(endpoint_put_config)?; + api.register(endpoint_get_config_save)?; + api.openapi("Salaryman", semver::Version::new(1, 0, 0)) + .write(&mut std::io::stdout())?; + let server = ServerBuilder::new(api, ctx.clone(), log) + .config(config) + .start()?; + server.await?; + Ok(()) +} -- cgit 1.4.1-2-gfad0