No OneTemporary
Actions

Size

48 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/src/daemon/api/routes.rs b/src/daemon/api/routes.rs
	index 81e6400..585506f 100644
	--- a/src/daemon/api/routes.rs
	+++ b/src/daemon/api/routes.rs
	@@ -1,834 +1,834 @@
	#![allow(non_snake_case)]

	use chrono::{DateTime, Utc};
	use global_placeholders::global;
	use macros_rs::{fmtstr, string, ternary, then};
	use prometheus::{Encoder, TextEncoder};
	use psutil::process::{MemoryInfo, Process};
	use reqwest::header::HeaderValue;
	use serde::Deserialize;
	use tera::{Context, Tera};
	use utoipa::ToSchema;

	use rocket::{
	get,
	http::{ContentType, Status},
	post,
	serde::{json::Json, Serialize},
	State,
	};

	use super::{
	helpers::{generic_error, not_found, GenericError, NotFound},
	structs::ErrorMessage,
	EnableWebUI, TeraState,
	};

	use pmc::{
	config, file, helpers,
	process::{dump, http::client, ItemSingle, ProcessItem, Runner},
	};

	use crate::daemon::{
	api::{HTTP_COUNTER, HTTP_REQ_HISTOGRAM},
	pid,
	};

	use std::{
	collections::BTreeMap,
	env,
	fs::{self, File},
	io::{self, BufRead, BufReader},
	path::PathBuf,
	};

	pub(crate) struct Token;
	type EnvList = Json<BTreeMap<String, String>>;

	#[allow(dead_code)]
	#[derive(ToSchema)]
	#[schema(as = MemoryInfo)]
	pub(crate) struct DocMemoryInfo {
	rss: u64,
	vms: u64,
	#[cfg(target_os = "linux")]
	shared: u64,
	#[cfg(target_os = "linux")]
	text: u64,
	#[cfg(target_os = "linux")]
	data: u64,
	#[cfg(target_os = "macos")]
	page_faults: u64,
	#[cfg(target_os = "macos")]
	pageins: u64,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub(crate) struct ActionBody {
	#[schema(example = "restart")]
	method: String,
	}

	#[derive(Serialize, ToSchema)]
	pub(crate) struct ConfigBody {
	#[schema(example = "bash")]
	shell: String,
	#[schema(min_items = 1, example = json!(["-c"]))]
	args: Vec<String>,
	#[schema(example = "/home/user/.pmc/logs")]
	log_path: String,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub(crate) struct CreateBody {
	#[schema(example = "app")]
	name: Option<String>,
	#[schema(example = "node index.js")]
	script: String,
	#[schema(value_type = String, example = "/projects/app")]
	path: PathBuf,
	#[schema(example = "src")]
	watch: Option<String>,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub(crate) struct ActionResponse {
	#[schema(example = true)]
	done: bool,
	#[schema(example = "name")]
	action: String,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub(crate) struct LogResponse {
	logs: Vec<String>,
	}

	#[derive(Serialize, ToSchema)]
	pub struct MetricsRoot {
	pub version: Version,
	pub daemon: Daemon,
	}

	#[derive(Serialize, ToSchema)]
	pub struct Version {
	#[schema(example = "v1.0.0")]
	pub pkg: String,
	pub hash: Option<&'static str>,
	#[schema(example = "2000-01-01")]
	pub build_date: &'static str,
	#[schema(example = "release")]
	pub target: &'static str,
	}

	#[derive(Serialize, ToSchema)]
	pub struct Daemon {
	pub pid: Option<i32>,
	#[schema(example = true)]
	pub running: bool,
	pub uptime: String,
	pub process_count: usize,
	#[schema(example = "default")]
	pub daemon_type: String,
	pub stats: Stats,
	}

	#[derive(Serialize, ToSchema)]
	pub struct Stats {
	pub memory_usage: String,
	pub cpu_percent: String,
	}

	fn attempt(done: bool, method: &str) -> ActionResponse {
	ActionResponse {
	done,
	action: ternary!(done, Box::leak(Box::from(method)), "DOES_NOT_EXIST").to_string(),
	}
	}

	fn render(name: &str, tmpl: &Tera, ctx: &Context) -> Result<String, NotFound> { tmpl.render(name, &ctx).or(Err(not_found("Page was not found"))) }

	#[get("/")]
	pub async fn dashboard(state: &State<TeraState>, _webui: EnableWebUI) -> Result<(ContentType, String), NotFound> {
	let mut ctx = Context::new();

	ctx.insert("base_path", &state.path);
	let payload = render("dashboard", &state.tera, &ctx)?;
	Ok((ContentType::HTML, payload))
	}

	#[get("/login")]
	pub async fn login(state: &State<TeraState>, _webui: EnableWebUI) -> Result<(ContentType, String), NotFound> {
	let mut ctx = Context::new();

	ctx.insert("base_path", &state.path);
	let payload = render("login", &state.tera, &ctx)?;
	Ok((ContentType::HTML, payload))
	}

	#[get("/view/<id>")]
	pub async fn view_process(id: usize, state: &State<TeraState>, _webui: EnableWebUI) -> Result<(ContentType, String), NotFound> {
	let mut ctx = Context::new();

	ctx.insert("base_path", &state.path);
	ctx.insert("process_id", &id);

	let payload = render("view", &state.tera, &ctx)?;
	Ok((ContentType::HTML, payload))
	}

	#[get("/daemon/prometheus")]
	#[utoipa::path(get, tag = "Daemon", path = "/daemon/prometheus", security((), ("api_key" = [])),
	responses(
	(
	description = "Get prometheus metrics", body = String, status = 200,
	example = json!("# HELP daemon_cpu_percentage The cpu usage graph of the daemon.\n# TYPE daemon_cpu_percentage histogram\ndaemon_cpu_percentage_bucket{le=\"0.005\"} 0"),
	),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn prometheus_handler(_t: Token) -> String {
	let encoder = TextEncoder::new();
	let mut buffer = Vec::<u8>::new();
	let metric_families = prometheus::gather();

	encoder.encode(&metric_families, &mut buffer).unwrap();
	String::from_utf8(buffer.clone()).unwrap()
	}

	#[get("/daemon/servers")]
	#[utoipa::path(get, tag = "Daemon", path = "/daemon/servers", security((), ("api_key" = [])),
	responses(
	(status = 200, description = "Get daemon servers successfully", body = [String]),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn servers_handler(_t: Token) -> Result<Json<Vec<String>>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["servers"]).start_timer();

	if let Some(servers) = config::servers().servers {
	HTTP_COUNTER.inc();
	timer.observe_duration();

	Ok(Json(servers.into_keys().collect()))
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[get("/remote/<name>/list")]
	#[utoipa::path(get, tag = "Remote", path = "/remote/{name}/list", security((), ("api_key" = [])),
	params(("name" = String, Path, description = "Name of remote daemon", example = "example"),),
	responses(
	(status = 200, description = "Get list from remote daemon successfully", body = [ProcessItem]),
	(status = NOT_FOUND, description = "Remote daemon does not exist", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn remote_list(name: String, _t: Token) -> Result<Json<Vec<ProcessItem>>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["list"]).start_timer();

	if let Some(servers) = config::servers().servers {
	let (address, (client, headers)) = match servers.get(&name) {
	Some(server) => (&server.address, client(&server.token).await),
	None => return Err(generic_error(Status::NotFound, string!("Server was not found"))),
	};

	HTTP_COUNTER.inc();
	timer.observe_duration();

	match client.get(fmtstr!("{address}/list")).headers(headers).send().await {
	Ok(data) => {
	if data.status() != 200 {
	let err = data.json::<ErrorMessage>().await.unwrap();
	Err(generic_error(err.code, err.message))
	} else {
	Ok(Json(data.json::<Vec<ProcessItem>>().await.unwrap()))
	}
	}
	Err(err) => Err(generic_error(Status::InternalServerError, err.to_string())),
	}
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[get("/remote/<name>/info/<id>")]
	#[utoipa::path(get, tag = "Remote", path = "/remote/{name}/info/{id}", security((), ("api_key" = [])),
	params(
	("name" = String, Path, description = "Name of remote daemon", example = "example"),
	("id" = usize, Path, description = "Process id to get information for", example = 0)
	),
	responses(
	(status = 200, description = "Get process info from remote daemon successfully", body = [ProcessItem]),
	(status = NOT_FOUND, description = "Remote daemon does not exist", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn remote_info(name: String, id: usize, _t: Token) -> Result<Json<ItemSingle>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["info"]).start_timer();

	if let Some(servers) = config::servers().servers {
	let (address, (client, headers)) = match servers.get(&name) {
	Some(server) => (&server.address, client(&server.token).await),
	None => return Err(generic_error(Status::NotFound, string!("Server was not found"))),
	};

	HTTP_COUNTER.inc();
	timer.observe_duration();

	match client.get(fmtstr!("{address}/process/{id}/info")).headers(headers).send().await {
	Ok(data) => {
	if data.status() != 200 {
	let err = data.json::<ErrorMessage>().await.unwrap();
	Err(generic_error(err.code, err.message))
	} else {
	Ok(Json(data.json::<ItemSingle>().await.unwrap()))
	}
	}
	Err(err) => Err(generic_error(Status::InternalServerError, err.to_string())),
	}
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[get("/remote/<name>/logs/<id>/<kind>")]
	#[utoipa::path(get, tag = "Remote", path = "/remote/{name}/logs/{id}/{kind}", security((), ("api_key" = [])),
	params(
	("name" = String, Path, description = "Name of remote daemon", example = "example"),
	("id" = usize, Path, description = "Process id to get information for", example = 0),
	("kind" = String, Path, description = "Log output type", example = "out")
	),
	responses(
	(status = 200, description = "Remote process logs of {type} fetched", body = LogResponse),
	(status = NOT_FOUND, description = "Remote daemon does not exist", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn remote_logs(name: String, id: usize, kind: String, _t: Token) -> Result<Json<LogResponse>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["info"]).start_timer();

	if let Some(servers) = config::servers().servers {
	let (address, (client, headers)) = match servers.get(&name) {
	Some(server) => (&server.address, client(&server.token).await),
	None => return Err(generic_error(Status::NotFound, string!("Server was not found"))),
	};

	HTTP_COUNTER.inc();
	timer.observe_duration();

	match client.get(fmtstr!("{address}/process/{id}/logs/{kind}")).headers(headers).send().await {
	Ok(data) => {
	if data.status() != 200 {
	let err = data.json::<ErrorMessage>().await.unwrap();
	Err(generic_error(err.code, err.message))
	} else {
	Ok(Json(data.json::<LogResponse>().await.unwrap()))
	}
	}
	Err(err) => Err(generic_error(Status::InternalServerError, err.to_string())),
	}
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[post("/remote/<name>/rename/<id>", format = "text", data = "<body>")]
	#[utoipa::path(post, tag = "Remote", path = "/remote/{name}/rename/{id}",
	security((), ("api_key" = [])),
	request_body(content = String, example = json!("example_name")),
	params(
	("id" = usize, Path, description = "Process id to rename", example = 0),
	("name" = String, Path, description = "Name of remote daemon", example = "example"),
	),
	responses(
	(
	description = "Remote rename process successful", body = ActionResponse,
	example = json!({"action": "rename", "done": true }), status = 200,
	),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn remote_rename(name: String, id: usize, body: String, _t: Token) -> Result<Json<ActionResponse>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["rename"]).start_timer();

	if let Some(servers) = config::servers().servers {
	let (address, (client, mut headers)) = match servers.get(&name) {
	Some(server) => (&server.address, client(&server.token).await),
	None => return Err(generic_error(Status::NotFound, string!("Server was not found"))),
	};

	HTTP_COUNTER.inc();
	timer.observe_duration();
	headers.insert("content-type", HeaderValue::from_static("text/plain"));

	match client.post(fmtstr!("{address}/process/{id}/rename")).body(body).headers(headers).send().await {
	Ok(data) => {
	if data.status() != 200 {
	let err = data.json::<ErrorMessage>().await.unwrap();
	Err(generic_error(err.code, err.message))
	} else {
	Ok(Json(data.json::<ActionResponse>().await.unwrap()))
	}
	}
	Err(err) => Err(generic_error(Status::InternalServerError, err.to_string())),
	}
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[post("/remote/<name>/action/<id>", format = "json", data = "<body>")]
	#[utoipa::path(post, tag = "Remote", path = "/remote/{name}/action/{id}", request_body = ActionBody,
	security((), ("api_key" = [])),
	params(
	("id" = usize, Path, description = "Process id to run action on", example = 0),
	("name" = String, Path, description = "Name of remote daemon", example = "example")
	),
	responses(
	(status = 200, description = "Run action on remote process successful", body = ActionResponse),
	(status = NOT_FOUND, description = "Process/action was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn remote_action(name: String, id: usize, body: Json<ActionBody>, _t: Token) -> Result<Json<ActionResponse>, GenericError> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["action"]).start_timer();

	if let Some(servers) = config::servers().servers {
	let (address, (client, headers)) = match servers.get(&name) {
	Some(server) => (&server.address, client(&server.token).await),
	None => return Err(generic_error(Status::NotFound, string!("Server was not found"))),
	};

	HTTP_COUNTER.inc();
	timer.observe_duration();

	match client.post(fmtstr!("{address}/process/{id}/action")).json(&body.0).headers(headers).send().await {
	Ok(data) => {
	if data.status() != 200 {
	let err = data.json::<ErrorMessage>().await.unwrap();
	Err(generic_error(err.code, err.message))
	} else {
	Ok(Json(data.json::<ActionResponse>().await.unwrap()))
	}
	}
	Err(err) => Err(generic_error(Status::InternalServerError, err.to_string())),
	}
	} else {
	Err(generic_error(Status::BadRequest, string!("No servers have been added")))
	}
	}

	#[get("/daemon/dump")]
	#[utoipa::path(get, tag = "Daemon", path = "/daemon/dump", security((), ("api_key" = [])),
	responses(
	(status = 200, description = "Dump processes successfully", body = [u8]),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn dump_handler(_t: Token) -> Vec<u8> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["dump"]).start_timer();

	HTTP_COUNTER.inc();
	timer.observe_duration();

	dump::raw()
	}

	#[get("/daemon/config")]
	#[utoipa::path(get, tag = "Daemon", path = "/daemon/config", security((), ("api_key" = [])),
	responses(
	(status = 200, description = "Get daemon config successfully", body = ConfigBody),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn config_handler(_t: Token) -> Json<ConfigBody> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["dump"]).start_timer();
	let config = config::read().runner;

	HTTP_COUNTER.inc();
	timer.observe_duration();

	Json(ConfigBody {
	shell: config.shell,
	args: config.args,
	log_path: config.log_path,
	})
	}

	#[get("/list")]
	#[utoipa::path(get, path = "/list", tag = "Process", security((), ("api_key" = [])),
	responses(
	(status = 200, description = "List processes successfully", body = [ProcessItem]),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn list_handler(_t: Token) -> Json<Vec<ProcessItem>> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["list"]).start_timer();
	let data = Runner::new().fetch();

	HTTP_COUNTER.inc();
	timer.observe_duration();

	Json(data)
	}

	#[get("/process/<id>/logs/<kind>")]
	#[utoipa::path(get, tag = "Process", path = "/process/{id}/logs/{kind}",
	security((), ("api_key" = [])),
	params(
	("id" = usize, Path, description = "Process id to get logs for", example = 0),
	("kind" = String, Path, description = "Log output type", example = "out")
	),
	responses(
	(status = 200, description = "Process logs of {type} fetched", body = LogResponse),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn logs_handler(id: usize, kind: String, _t: Token) -> Result<Json<LogResponse>, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["log"]).start_timer();

	HTTP_COUNTER.inc();
	match Runner::new().info(id) {
	Some(item) => {
	let log_file = match kind.as_str() {
	"out" \| "stdout" => item.logs().out,
	"error" \| "stderr" => item.logs().error,
	_ => item.logs().out,
	};

	match File::open(log_file) {
	Ok(data) => {
	let reader = BufReader::new(data);
	let logs: Vec<String> = reader.lines().collect::<io::Result<_>>().unwrap();

	timer.observe_duration();
	Ok(Json(LogResponse { logs }))
	}
	Err(_) => Ok(Json(LogResponse { logs: vec![] })),
	}
	}
	None => {
	timer.observe_duration();
	Err(not_found("Process was not found"))
	}
	}
	}

	#[get("/process/<id>/logs/<kind>/raw")]
	#[utoipa::path(get, tag = "Process", path = "/process/{id}/logs/{kind}/raw",
	security((), ("api_key" = [])),
	params(
	("id" = usize, Path, description = "Process id to get logs for", example = 0),
	("kind" = String, Path, description = "Log output type", example = "out")
	),
	responses(
	(
	description = "Process logs of {type} fetched raw", body = String, status = 200,
	example = json!("# PATH path/of/file.log\nserver started on port 3000")
	),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn logs_raw_handler(id: usize, kind: String, _t: Token) -> Result<String, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["log"]).start_timer();

	HTTP_COUNTER.inc();
	match Runner::new().info(id) {
	Some(item) => {
	let log_file = match kind.as_str() {
	"out" \| "stdout" => item.logs().out,
	"error" \| "stderr" => item.logs().error,
	_ => item.logs().out,
	};

	let data = match fs::read_to_string(&log_file) {
	Ok(data) => format!("# PATH {log_file}\n{data}"),
	Err(err) => err.to_string(),
	};

	timer.observe_duration();
	Ok(data)
	}
	None => {
	timer.observe_duration();
	Err(not_found("Process was not found"))
	}
	}
	}

	#[get("/process/<id>/info")]
	#[utoipa::path(get, tag = "Process", path = "/process/{id}/info", security((), ("api_key" = [])),
	params(("id" = usize, Path, description = "Process id to get information for", example = 0)),
	responses(
	(status = 200, description = "Current process info retrieved", body = ItemSingle),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn info_handler(id: usize, _t: Token) -> Result<Json<ItemSingle>, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["info"]).start_timer();
	let runner = Runner::new();

	if runner.exists(id) {
	let item = runner.get(id);
	HTTP_COUNTER.inc();
	timer.observe_duration();
	Ok(Json(item.fetch()))
	} else {
	Err(not_found("Process was not found"))
	}
	}

	#[post("/process/create", format = "json", data = "<body>")]
	#[utoipa::path(post, tag = "Process", path = "/process/create", request_body(content = CreateBody),
	security((), ("api_key" = [])),
	responses(
	(
	description = "Create process successful", body = ActionResponse,
	example = json!({"action": "create", "done": true }), status = 200,
	),
	(status = INTERNAL_SERVER_ERROR, description = "Failed to create process", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn create_handler(body: Json<CreateBody>, _t: Token) -> Result<Json<ActionResponse>, ()> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["create"]).start_timer();
	let mut runner = Runner::new();

	HTTP_COUNTER.inc();

	let name = match &body.name {
	Some(name) => string!(name),
	None => string!(body.script.split_whitespace().next().unwrap_or_default()),
	};

	runner.start(&name, &body.script, body.path.clone(), &body.watch).save();
	timer.observe_duration();

	Ok(Json(attempt(true, "create")))
	}

	#[post("/process/<id>/rename", format = "text", data = "<body>")]
	#[utoipa::path(post, tag = "Process", path = "/process/{id}/rename",
	security((), ("api_key" = [])),
	request_body(content = String, example = json!("example_name")),
	params(("id" = usize, Path, description = "Process id to rename", example = 0)),
	responses(
	(
	description = "Rename process successful", body = ActionResponse,
	example = json!({"action": "rename", "done": true }), status = 200,
	),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn rename_handler(id: usize, body: String, _t: Token) -> Result<Json<ActionResponse>, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["rename"]).start_timer();
	let runner = Runner::new();

	match runner.clone().info(id) {
	Some(process) => {
	HTTP_COUNTER.inc();
	let mut item = runner.get(id);
	item.rename(body.trim().replace("\n", ""));
	then!(process.running, item.restart());
	timer.observe_duration();
	Ok(Json(attempt(true, "rename")))
	}
	None => {
	timer.observe_duration();
	Err(not_found("Process was not found"))
	}
	}
	}

	#[get("/process/<id>/env")]
	#[utoipa::path(get, tag = "Process", path = "/process/{id}/env",
	params(("id" = usize, Path, description = "Process id to fetch env from", example = 0)),
	responses(
	(
	description = "Current process env", body = HashMap<String, String>,
	example = json!({"ENV_TEST_VALUE": "example_value"}), status = 200
	),
	(status = NOT_FOUND, description = "Process was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn env_handler(id: usize, _t: Token) -> Result<EnvList, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["env"]).start_timer();

	HTTP_COUNTER.inc();
	match Runner::new().info(id) {
	Some(item) => {
	timer.observe_duration();
	Ok(Json(item.clone().env))
	}
	None => {
	timer.observe_duration();
	Err(not_found("Process was not found"))
	}
	}
	}

	#[post("/process/<id>/action", format = "json", data = "<body>")]
	#[utoipa::path(post, tag = "Process", path = "/process/{id}/action", request_body = ActionBody,
	security((), ("api_key" = [])),
	params(("id" = usize, Path, description = "Process id to run action on", example = 0)),
	responses(
	(status = 200, description = "Run action on process successful", body = ActionResponse),
	(status = NOT_FOUND, description = "Process/action was not found", body = ErrorMessage),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn action_handler(id: usize, body: Json<ActionBody>, _t: Token) -> Result<Json<ActionResponse>, NotFound> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["action"]).start_timer();
	let mut runner = Runner::new();
	let method = body.method.as_str();

	if runner.exists(id) {
	HTTP_COUNTER.inc();
	match method {
	"start" \| "restart" => {
	- runner.restart(id, false);
	+ runner.get(id).restart();
	timer.observe_duration();
	Ok(Json(attempt(true, method)))
	}
	"stop" \| "kill" => {
	runner.get(id).stop();
	timer.observe_duration();
	Ok(Json(attempt(true, method)))
	}
	"remove" \| "delete" => {
	runner.remove(id);
	timer.observe_duration();
	Ok(Json(attempt(true, method)))
	}
	_ => {
	timer.observe_duration();
	Err(not_found("Process was not found"))
	}
	}
	} else {
	Err(not_found("Process was not found"))
	}
	}

	#[get("/daemon/metrics")]
	#[utoipa::path(get, tag = "Daemon", path = "/daemon/metrics", security((), ("api_key" = [])),
	responses(
	(status = 200, description = "Get daemon metrics", body = MetricsRoot),
	(
	status = UNAUTHORIZED, description = "Authentication failed or not provided", body = ErrorMessage,
	example = json!({"code": 401, "message": "Unauthorized"})
	)
	)
	)]
	pub async fn metrics_handler(_t: Token) -> Json<MetricsRoot> {
	let timer = HTTP_REQ_HISTOGRAM.with_label_values(&["metrics"]).start_timer();
	let mut pid: Option<i32> = None;
	let mut cpu_percent: Option<f32> = None;
	let mut uptime: Option<DateTime<Utc>> = None;
	let mut memory_usage: Option<MemoryInfo> = None;
	let mut runner: Runner = file::read_object(global!("pmc.dump"));

	HTTP_COUNTER.inc();
	if pid::exists() {
	if let Ok(process_id) = pid::read() {
	if let Ok(mut process) = Process::new(process_id as u32) {
	pid = Some(process_id);
	uptime = Some(pid::uptime().unwrap());
	memory_usage = process.memory_info().ok();
	cpu_percent = process.cpu_percent().ok();
	}
	}
	}

	let memory_usage = match memory_usage {
	Some(usage) => helpers::format_memory(usage.rss()),
	None => string!("0b"),
	};

	let cpu_percent = match cpu_percent {
	Some(percent) => format!("{:.2}%", percent),
	None => string!("0%"),
	};

	let uptime = match uptime {
	Some(uptime) => helpers::format_duration(uptime),
	None => string!("none"),
	};

	timer.observe_duration();
	Json(MetricsRoot {
	version: Version {
	target: env!("PROFILE"),
	build_date: env!("BUILD_DATE"),
	pkg: format!("v{}", env!("CARGO_PKG_VERSION")),
	hash: ternary!(env!("GIT_HASH_FULL") == "", None, Some(env!("GIT_HASH_FULL"))),
	},
	daemon: Daemon {
	pid,
	uptime,
	running: pid::exists(),
	process_count: runner.count(),
	daemon_type: global!("pmc.daemon.kind"),
	stats: Stats { memory_usage, cpu_percent },
	},
	})
	}
	diff --git a/src/process/mod.rs b/src/process/mod.rs
	index ca9bc9c..f6fe68a 100644
	--- a/src/process/mod.rs
	+++ b/src/process/mod.rs
	@@ -1,566 +1,570 @@
	mod unix;

	use crate::{
	config,
	config::structs::Server,
	file, helpers,
	service::{run, stop, ProcessMetadata},
	};

	use std::{
	env,
	path::PathBuf,
	sync::{Arc, Mutex},
	};

	use nix::{
	sys::signal::{kill, Signal},
	unistd::Pid,
	};

	use chrono::serde::ts_milliseconds;
	use chrono::{DateTime, Utc};
	use global_placeholders::global;
	use macros_rs::{crashln, string, ternary, then};
	use psutil::process;
	use serde::{Deserialize, Serialize};
	use std::collections::BTreeMap;
	use utoipa::ToSchema;

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct ItemSingle {
	pub info: Info,
	pub stats: Stats,
	pub watch: Watch,
	pub log: Log,
	pub raw: Raw,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct Info {
	pub id: usize,
	pub pid: i64,
	pub name: String,
	pub status: String,
	#[schema(value_type = String, example = "/path")]
	pub path: PathBuf,
	pub uptime: String,
	pub command: String,
	pub children: Vec<i64>,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct Stats {
	pub restarts: u64,
	pub start_time: i64,
	pub cpu_percent: Option<f32>,
	pub memory_usage: Option<MemoryInfo>,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct MemoryInfo {
	pub rss: u64,
	pub vms: u64,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct Log {
	pub out: String,
	pub error: String,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct Raw {
	pub running: bool,
	pub crashed: bool,
	pub crashes: u64,
	}

	#[derive(Clone)]
	pub struct LogInfo {
	pub out: String,
	pub error: String,
	}

	#[derive(Serialize, Deserialize, ToSchema)]
	pub struct ProcessItem {
	pid: i64,
	id: usize,
	cpu: String,
	mem: String,
	name: String,
	restarts: u64,
	status: String,
	uptime: String,
	#[schema(example = "/path")]
	watch_path: String,
	#[schema(value_type = String, example = "2000-01-01T01:00:00.000Z")]
	start_time: DateTime<Utc>,
	}

	#[derive(Clone)]
	pub struct ProcessWrapper {
	pub id: usize,
	pub runner: Arc<Mutex<Runner>>,
	}

	type Env = BTreeMap<String, String>;

	#[derive(Clone, Debug, Deserialize, Serialize)]
	pub struct Process {
	pub id: usize,
	pub pid: i64,
	pub env: Env,
	pub name: String,
	pub path: PathBuf,
	pub script: String,
	pub restarts: u64,
	pub running: bool,
	pub crash: Crash,
	pub watch: Watch,
	pub children: Vec<i64>,
	#[serde(with = "ts_milliseconds")]
	pub started: DateTime<Utc>,
	}

	#[derive(Clone, Debug, Deserialize, Serialize)]
	pub struct Crash {
	pub crashed: bool,
	pub value: u64,
	}

	#[derive(Clone, Debug, Deserialize, Serialize, ToSchema)]
	pub struct Watch {
	pub enabled: bool,
	#[schema(example = "/path")]
	pub path: String,
	pub hash: String,
	}

	#[derive(Clone, Debug, Deserialize, Serialize)]
	pub struct Runner {
	pub id: id::Id,
	#[serde(skip)]
	pub remote: Option<Remote>,
	pub list: BTreeMap<usize, Process>,
	}

	#[derive(Clone, Debug)]
	pub struct Remote {
	address: String,
	token: Option<String>,
	pub config: RemoteConfig,
	}

	#[derive(Clone, Debug, Deserialize)]
	pub struct RemoteConfig {
	pub shell: String,
	pub args: Vec<String>,
	pub log_path: String,
	}

	pub enum Status {
	Offline,
	Running,
	}

	impl Status {
	pub fn to_bool(&self) -> bool {
	match self {
	Status::Offline => false,
	Status::Running => true,
	}
	}
	}

	macro_rules! lock {
	($runner:expr) => {{
	match $runner.lock() {
	Ok(runner) => runner,
	Err(err) => crashln!("Unable to lock mutex: {err}"),
	}
	}};
	}

	fn kill_children(children: Vec<i64>) {
	for pid in children {
	if let Err(err) = kill(Pid::from_raw(pid as i32), Signal::SIGTERM) {
	log::error!("Failed to stop pid {pid}: {err:?}");
	};
	}
	}

	impl Runner {
	pub fn new() -> Self { dump::read() }

	pub fn connect(name: String, Server { address, token }: Server, verbose: bool) -> Option<Self> {
	let remote_config = match config::from(&address, token.as_deref()) {
	Ok(config) => config,
	Err(err) => {
	log::error!("{err}");
	return None;
	}
	};

	if let Ok(dump) = dump::from(&address, token.as_deref()) {
	then!(verbose, println!("{} Fetched remote (name={name}, address={address})", *helpers::SUCCESS));
	Some(Runner {
	remote: Some(Remote {
	token,
	address: string!(address),
	config: remote_config,
	}),
	..dump
	})
	} else {
	None
	}
	}

	pub fn start(&mut self, name: &String, command: &String, path: PathBuf, watch: &Option<String>) -> &mut Self {
	if let Some(remote) = &self.remote {
	if let Err(err) = http::create(remote, name, command, path, watch) {
	crashln!("{} Failed to start create {name}\nError: {:#?}", *helpers::FAIL, err);
	};
	} else {
	let id = self.id.next();
	let config = config::read().runner;
	let crash = Crash { crashed: false, value: 0 };

	let watch = match watch {
	Some(watch) => Watch {
	enabled: true,
	path: string!(watch),
	hash: hash::create(file::cwd().join(watch)),
	},
	None => Watch {
	enabled: false,
	path: string!(""),
	hash: string!(""),
	},
	};

	let pid = run(ProcessMetadata {
	args: config.args,
	name: name.clone(),
	shell: config.shell,
	command: command.clone(),
	log_path: config.log_path,
	env: unix::env(),
	});

	self.list.insert(
	id,
	Process {
	id,
	pid,
	path,
	watch,
	crash,
	restarts: 0,
	running: true,
	children: vec![],
	name: name.clone(),
	started: Utc::now(),
	script: command.clone(),
	env: env::vars().collect(),
	},
	);
	}

	return self;
	}

	pub fn restart(&mut self, id: usize, dead: bool) -> &mut Self {
	if let Some(remote) = &self.remote {
	if let Err(err) = http::restart(remote, id) {
	crashln!("{} Failed to start process {id}\nError: {:#?}", *helpers::FAIL, err);
	};
	} else {
	let process = self.process(id);
	let config = config::read().runner;
	let Process { path, script, name, .. } = process.clone();

	kill_children(process.children.clone());
	stop(process.pid);

	if let Err(err) = std::env::set_current_dir(&path) {
	crashln!("{} Failed to set working directory {:?}\nError: {:#?}", *helpers::FAIL, path, err);
	};

	process.pid = run(ProcessMetadata {
	args: config.args,
	name: name.clone(),
	shell: config.shell,
	log_path: config.log_path,
	command: script.to_string(),
	env: unix::env(),
	});

	+ println!("{:?}", process.pid);
	+
	process.running = true;
	process.children = vec![];
	process.started = Utc::now();
	process.crash.crashed = false;
	process.env = env::vars().collect();

	+ println!("{:?}", process);
	+
	then!(dead, process.restarts += 1);
	then!(dead, process.crash.value += 1);
	then!(!dead, process.crash.value = 0);
	}

	return self;
	}

	pub fn remove(&mut self, id: usize) {
	if let Some(remote) = &self.remote {
	if let Err(err) = http::remove(remote, id) {
	crashln!("{} Failed to stop remove {id}\nError: {:#?}", *helpers::FAIL, err);
	};
	} else {
	self.stop(id);
	self.list.remove(&id);
	dump::write(&self);
	}
	}

	pub fn set_id(&mut self, id: id::Id) {
	self.id = id;
	self.id.next();
	dump::write(&self);
	}

	pub fn set_status(&mut self, id: usize, status: Status) {
	self.process(id).running = status.to_bool();
	dump::write(&self);
	}

	pub fn items(&self) -> BTreeMap<usize, Process> { self.list.clone() }
	pub fn items_mut(&mut self) -> &mut BTreeMap<usize, Process> { &mut self.list }

	pub fn save(&self) { then!(self.remote.is_none(), dump::write(&self)) }
	pub fn count(&mut self) -> usize { self.list().count() }
	pub fn is_empty(&self) -> bool { self.list.is_empty() }
	pub fn exists(&self, id: usize) -> bool { self.list.contains_key(&id) }
	pub fn info(&self, id: usize) -> Option<&Process> { self.list.get(&id) }
	pub fn list<'l>(&'l mut self) -> impl Iterator<Item = (&'l usize, &'l mut Process)> { self.list.iter_mut().map(\|(k, v)\| (k, v)) }
	pub fn process(&mut self, id: usize) -> &mut Process { self.list.get_mut(&id).unwrap_or_else(\|\| crashln!("{} Process ({id}) not found", *helpers::FAIL)) }
	pub fn pid(&self, id: usize) -> i64 { self.list.get(&id).unwrap_or_else(\|\| crashln!("{} Process ({id}) not found", *helpers::FAIL)).pid }

	pub fn get(self, id: usize) -> ProcessWrapper {
	ProcessWrapper {
	id,
	runner: Arc::new(Mutex::new(self)),
	}
	}

	pub fn set_crashed(&mut self, id: usize) -> &mut Self {
	self.process(id).crash.crashed = true;
	return self;
	}

	pub fn set_children(&mut self, id: usize, children: Vec<i64>) -> &mut Self {
	self.process(id).children = children;
	return self;
	}

	pub fn new_crash(&mut self, id: usize) -> &mut Self {
	self.process(id).crash.value += 1;
	return self;
	}

	pub fn stop(&mut self, id: usize) -> &mut Self {
	if let Some(remote) = &self.remote {
	if let Err(err) = http::stop(remote, id) {
	crashln!("{} Failed to stop process {id}\nError: {:#?}", *helpers::FAIL, err);
	};
	} else {
	let process = self.process(id);

	kill_children(process.children.clone());
	stop(process.pid);

	process.running = false;
	process.crash.crashed = false;
	process.crash.value = 0;
	process.children = vec![];
	}

	return self;
	}

	pub fn rename(&mut self, id: usize, name: String) -> &mut Self {
	if let Some(remote) = &self.remote {
	if let Err(err) = http::rename(remote, id, name) {
	crashln!("{} Failed to rename process {id}\nError: {:#?}", *helpers::FAIL, err);
	};
	} else {
	self.process(id).name = name;
	}

	return self;
	}

	pub fn watch(&mut self, id: usize, path: &str, enabled: bool) -> &mut Self {
	let process = self.process(id);
	process.watch = Watch {
	enabled,
	path: string!(path),
	hash: ternary!(enabled, hash::create(process.path.join(path)), string!("")),
	};

	return self;
	}

	pub fn fetch(&self) -> Vec<ProcessItem> {
	let mut processes: Vec<ProcessItem> = Vec::new();

	for (id, item) in self.items() {
	let mut memory_usage: Option<MemoryInfo> = None;
	let mut cpu_percent: Option<f32> = None;

	if let Ok(mut process) = process::Process::new(item.pid as u32) {
	let mem_info_psutil = process.memory_info().ok();

	cpu_percent = process.cpu_percent().ok();
	memory_usage = Some(MemoryInfo {
	rss: mem_info_psutil.as_ref().unwrap().rss(),
	vms: mem_info_psutil.as_ref().unwrap().vms(),
	});
	}

	let cpu_percent = match cpu_percent {
	Some(percent) => format!("{:.2}%", percent),
	None => string!("0.00%"),
	};

	let memory_usage = match memory_usage {
	Some(usage) => helpers::format_memory(usage.rss),
	None => string!("0b"),
	};

	let status = if item.running {
	string!("online")
	} else {
	match item.crash.crashed {
	true => string!("crashed"),
	false => string!("stopped"),
	}
	};

	processes.push(ProcessItem {
	id,
	status,
	pid: item.pid,
	cpu: cpu_percent,
	mem: memory_usage,
	restarts: item.restarts,
	name: item.name.clone(),
	start_time: item.started,
	watch_path: item.watch.path.clone(),
	uptime: helpers::format_duration(item.started),
	});
	}

	return processes;
	}
	}

	impl Process {
	/// Get a log paths of the process item
	pub fn logs(&self) -> LogInfo {
	let name = self.name.replace(" ", "_");

	LogInfo {
	out: global!("pmc.logs.out", name.as_str()),
	error: global!("pmc.logs.error", name.as_str()),
	}
	}
	}

	impl ProcessWrapper {
	/// Stop the process item
	pub fn stop(&mut self) { lock!(self.runner).stop(self.id).save(); }

	/// Restart the process item
	pub fn restart(&mut self) { lock!(self.runner).restart(self.id, false).save(); }

	/// Rename the process item
	pub fn rename(&mut self, name: String) { lock!(self.runner).rename(self.id, name).save(); }

	/// Enable watching a path on the process item
	pub fn watch(&mut self, path: &str) { lock!(self.runner).watch(self.id, path, true).save(); }

	/// Disable watching on the process item
	pub fn disable_watch(&mut self) { lock!(self.runner).watch(self.id, "", false).save(); }

	/// Set the process item as crashed
	pub fn crashed(&mut self) { lock!(self.runner).restart(self.id, true).save(); }

	/// Get a json dump of the process item
	pub fn fetch(&self) -> ItemSingle {
	let mut runner = lock!(self.runner);

	let item = runner.process(self.id);
	let config = config::read().runner;

	let mut memory_usage: Option<MemoryInfo> = None;
	let mut cpu_percent: Option<f32> = None;

	if let Ok(mut process) = process::Process::new(item.pid as u32) {
	let mem_info_psutil = process.memory_info().ok();

	cpu_percent = process.cpu_percent().ok();
	memory_usage = Some(MemoryInfo {
	rss: mem_info_psutil.as_ref().unwrap().rss(),
	vms: mem_info_psutil.as_ref().unwrap().vms(),
	});
	}

	let status = if item.running {
	string!("online")
	} else {
	match item.crash.crashed {
	true => string!("crashed"),
	false => string!("stopped"),
	}
	};

	ItemSingle {
	info: Info {
	status,
	id: item.id,
	pid: item.pid,
	name: item.name.clone(),
	path: item.path.clone(),
	children: item.children.clone(),
	uptime: helpers::format_duration(item.started),
	command: format!("{} {} '{}'", config.shell, config.args.join(" "), item.script.clone()),
	},
	stats: Stats {
	cpu_percent,
	memory_usage,
	restarts: item.restarts,
	start_time: item.started.timestamp_millis(),
	},
	watch: Watch {
	enabled: item.watch.enabled,
	hash: item.watch.hash.clone(),
	path: item.watch.path.clone(),
	},
	log: Log {
	out: item.logs().out,
	error: item.logs().error,
	},
	raw: Raw {
	running: item.running,
	crashed: item.crash.crashed,
	crashes: item.crash.value,
	},
	}
	}
	}

	pub mod dump;
	pub mod hash;
	pub mod http;
	pub mod id;

File Metadata

Mime Type: text/x-diff
Expires: Mon, Feb 2, 1:26 AM (1 d, 15 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 494978
Default Alt Text: (48 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions