herocode/herolib_rust
12
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Add service manager support

Browse Source 
- Add a new service manager crate for dynamic service management
- Integrate service manager with Rhai for scripting
- Provide examples for circle worker management and basic usage
- Add comprehensive tests for service lifecycle and error handling
- Implement cross-platform support for macOS and Linux (zinit/systemd)
This commit is contained in:
Mahmoud-Emad
2025-07-01 18:00:21 +03:00
parent 46ad848e7e
commit 131d978450
28 changed files with 3562 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files

221
service_manager/src/launchctl.rs
View File

@@ -1,9 +1,15 @@
use crate::{ServiceConfig, ServiceManager, ServiceManagerError, ServiceStatus};
use async_trait::async_trait;
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use tokio::process::Command;
use tokio::runtime::Runtime;
// Shared runtime for async operations
static ASYNC_RUNTIME: Lazy<Runtime> = Lazy::new(|| {
Runtime::new().expect("Failed to create async runtime for LaunchctlServiceManager")
});
#[derive(Debug)]
pub struct LaunchctlServiceManager {
@@ -18,7 +24,10 @@ struct LaunchDaemon {
program_arguments: Vec<String>,
#[serde(rename = "WorkingDirectory", skip_serializing_if = "Option::is_none")]
working_directory: Option<String>,
#[serde(rename = "EnvironmentVariables", skip_serializing_if = "Option::is_none")]
#[serde(
rename = "EnvironmentVariables",
skip_serializing_if = "Option::is_none"
)]
environment_variables: Option<HashMap<String, String>>,
#[serde(rename = "KeepAlive", skip_serializing_if = "Option::is_none")]
keep_alive: Option<bool>,
@@ -85,7 +94,11 @@ impl LaunchctlServiceManager {
} else {
Some(config.environment.clone())
},
keep_alive: if config.auto_restart { Some(true) } else { None },
keep_alive: if config.auto_restart {
Some(true)
} else {
None
},
run_at_load: true,
standard_out_path: Some(log_path.to_string_lossy().to_string()),
standard_error_path: Some(log_path.to_string_lossy().to_string()),
@@ -94,8 +107,9 @@ impl LaunchctlServiceManager {
let mut plist_content = Vec::new();
plist::to_writer_xml(&mut plist_content, &launch_daemon)
.map_err(|e| ServiceManagerError::Other(format!("Failed to serialize plist: {}", e)))?;
let plist_content = String::from_utf8(plist_content)
.map_err(|e| ServiceManagerError::Other(format!("Failed to convert plist to string: {}", e)))?;
let plist_content = String::from_utf8(plist_content).map_err(|e| {
ServiceManagerError::Other(format!("Failed to convert plist to string: {}", e))
})?;
tokio::fs::write(&plist_path, plist_content).await?;
@@ -103,10 +117,7 @@ impl LaunchctlServiceManager {
}
async fn run_launchctl(&self, args: &[&str]) -> Result<String, ServiceManagerError> {
let output = Command::new("launchctl")
.args(args)
.output()
.await?;
let output = Command::new("launchctl").args(args).output().await?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr);
@@ -119,12 +130,16 @@ impl LaunchctlServiceManager {
Ok(String::from_utf8_lossy(&output.stdout).to_string())
}
async fn wait_for_service_status(&self, service_name: &str, timeout_secs: u64) -> Result<(), ServiceManagerError> {
use tokio::time::{sleep, Duration, timeout};
async fn wait_for_service_status(
&self,
service_name: &str,
timeout_secs: u64,
) -> Result<(), ServiceManagerError> {
use tokio::time::{sleep, timeout, Duration};
let timeout_duration = Duration::from_secs(timeout_secs);
let poll_interval = Duration::from_millis(500);
let result = timeout(timeout_duration, async {
loop {
match self.status(service_name) {
@@ -140,45 +155,65 @@ impl LaunchctlServiceManager {
// Extract error lines from logs
let error_lines: Vec<&str> = logs
.lines()
.filter(|line| line.to_lowercase().contains("error") || line.to_lowercase().contains("failed"))
.filter(|line| {
line.to_lowercase().contains("error")
|| line.to_lowercase().contains("failed")
})
.take(3)
.collect();
if error_lines.is_empty() {
format!("Service failed to start. Recent logs:\n{}",
logs.lines().rev().take(5).collect::<Vec<_>>().into_iter().rev().collect::<Vec<_>>().join("\n"))
format!(
"Service failed to start. Recent logs:\n{}",
logs.lines()
.rev()
.take(5)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect::<Vec<_>>()
.join("\n")
)
} else {
format!("Service failed to start. Errors:\n{}", error_lines.join("\n"))
format!(
"Service failed to start. Errors:\n{}",
error_lines.join("\n")
)
}
};
return Err(ServiceManagerError::StartFailed(service_name.to_string(), error_msg));
return Err(ServiceManagerError::StartFailed(
service_name.to_string(),
error_msg,
));
}
Ok(ServiceStatus::Stopped) | Ok(ServiceStatus::Unknown) => {
// Still starting, continue polling
sleep(poll_interval).await;
}
Err(ServiceManagerError::ServiceNotFound(_)) => {
return Err(ServiceManagerError::ServiceNotFound(service_name.to_string()));
return Err(ServiceManagerError::ServiceNotFound(
service_name.to_string(),
));
}
Err(e) => {
return Err(e);
}
}
}
}).await;
})
.await;
match result {
Ok(Ok(())) => Ok(()),
Ok(Err(e)) => Err(e),
Err(_) => Err(ServiceManagerError::StartFailed(
service_name.to_string(),
format!("Service did not start within {} seconds", timeout_secs)
format!("Service did not start within {} seconds", timeout_secs),
)),
}
}
}
#[async_trait]
impl ServiceManager for LaunchctlServiceManager {
fn exists(&self, service_name: &str) -> Result<bool, ServiceManagerError> {
let plist_path = self.get_plist_path(service_name);
@@ -186,15 +221,16 @@ impl ServiceManager for LaunchctlServiceManager {
}
fn start(&self, config: &ServiceConfig) -> Result<(), ServiceManagerError> {
// For synchronous version, we'll use blocking operations
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
// Use the shared runtime for async operations
ASYNC_RUNTIME.block_on(async {
let label = self.get_service_label(&config.name);
// Check if service is already loaded
let list_output = self.run_launchctl(&["list"]).await?;
if list_output.contains(&label) {
return Err(ServiceManagerError::ServiceAlreadyExists(config.name.clone()));
return Err(ServiceManagerError::ServiceAlreadyExists(
config.name.clone(),
));
}
// Create the plist file
@@ -204,23 +240,26 @@ impl ServiceManager for LaunchctlServiceManager {
let plist_path = self.get_plist_path(&config.name);
self.run_launchctl(&["load", &plist_path.to_string_lossy()])
.await
.map_err(|e| ServiceManagerError::StartFailed(config.name.clone(), e.to_string()))?;
.map_err(|e| {
ServiceManagerError::StartFailed(config.name.clone(), e.to_string())
})?;
Ok(())
})
}
fn start_existing(&self, service_name: &str) -> Result<(), ServiceManagerError> {
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
ASYNC_RUNTIME.block_on(async {
let label = self.get_service_label(service_name);
let plist_path = self.get_plist_path(service_name);
// Check if plist file exists
if !plist_path.exists() {
return Err(ServiceManagerError::ServiceNotFound(service_name.to_string()));
return Err(ServiceManagerError::ServiceNotFound(
service_name.to_string(),
));
}
// Check if service is already loaded and running
let list_output = self.run_launchctl(&["list"]).await?;
if list_output.contains(&label) {
@@ -231,53 +270,69 @@ impl ServiceManager for LaunchctlServiceManager {
}
_ => {
// Service is loaded but not running, try to start it
self.run_launchctl(&["start", &label])
.await
.map_err(|e| ServiceManagerError::StartFailed(service_name.to_string(), e.to_string()))?;
self.run_launchctl(&["start", &label]).await.map_err(|e| {
ServiceManagerError::StartFailed(
service_name.to_string(),
e.to_string(),
)
})?;
return Ok(());
}
}
}
// Service is not loaded, load it
self.run_launchctl(&["load", &plist_path.to_string_lossy()])
.await
.map_err(|e| ServiceManagerError::StartFailed(service_name.to_string(), e.to_string()))?;
.map_err(|e| {
ServiceManagerError::StartFailed(service_name.to_string(), e.to_string())
})?;
Ok(())
})
}
async fn start_and_confirm(&self, config: &ServiceConfig, timeout_secs: u64) -> Result<(), ServiceManagerError> {
fn start_and_confirm(
&self,
config: &ServiceConfig,
timeout_secs: u64,
) -> Result<(), ServiceManagerError> {
// First start the service
self.start(config)?;
// Then wait for confirmation
self.wait_for_service_status(&config.name, timeout_secs).await
// Then wait for confirmation using the shared runtime
ASYNC_RUNTIME.block_on(async {
self.wait_for_service_status(&config.name, timeout_secs)
.await
})
}
async fn run(&self, config: &ServiceConfig, timeout_secs: u64) -> Result<(), ServiceManagerError> {
self.start_and_confirm(config, timeout_secs).await
}
async fn start_existing_and_confirm(&self, service_name: &str, timeout_secs: u64) -> Result<(), ServiceManagerError> {
fn start_existing_and_confirm(
&self,
service_name: &str,
timeout_secs: u64,
) -> Result<(), ServiceManagerError> {
// First start the existing service
self.start_existing(service_name)?;
// Then wait for confirmation
self.wait_for_service_status(service_name, timeout_secs).await
// Then wait for confirmation using the shared runtime
ASYNC_RUNTIME.block_on(async {
self.wait_for_service_status(service_name, timeout_secs)
.await
})
}
fn stop(&self, service_name: &str) -> Result<(), ServiceManagerError> {
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
ASYNC_RUNTIME.block_on(async {
let _label = self.get_service_label(service_name);
let plist_path = self.get_plist_path(service_name);
// Unload the service
self.run_launchctl(&["unload", &plist_path.to_string_lossy()])
.await
.map_err(|e| ServiceManagerError::StopFailed(service_name.to_string(), e.to_string()))?;
.map_err(|e| {
ServiceManagerError::StopFailed(service_name.to_string(), e.to_string())
})?;
Ok(())
})
@@ -288,7 +343,10 @@ impl ServiceManager for LaunchctlServiceManager {
if let Err(e) = self.stop(service_name) {
// If stop fails because service doesn't exist, that's ok for restart
if !matches!(e, ServiceManagerError::ServiceNotFound(_)) {
return Err(ServiceManagerError::RestartFailed(service_name.to_string(), e.to_string()));
return Err(ServiceManagerError::RestartFailed(
service_name.to_string(),
e.to_string(),
));
}
}
@@ -301,18 +359,19 @@ impl ServiceManager for LaunchctlServiceManager {
}
fn status(&self, service_name: &str) -> Result<ServiceStatus, ServiceManagerError> {
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
ASYNC_RUNTIME.block_on(async {
let label = self.get_service_label(service_name);
let plist_path = self.get_plist_path(service_name);
// First check if the plist file exists
if !plist_path.exists() {
return Err(ServiceManagerError::ServiceNotFound(service_name.to_string()));
return Err(ServiceManagerError::ServiceNotFound(
service_name.to_string(),
));
}
let list_output = self.run_launchctl(&["list"]).await?;
if !list_output.contains(&label) {
return Ok(ServiceStatus::Stopped);
}
@@ -333,11 +392,14 @@ impl ServiceManager for LaunchctlServiceManager {
})
}
fn logs(&self, service_name: &str, lines: Option<usize>) -> Result<String, ServiceManagerError> {
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
fn logs(
&self,
service_name: &str,
lines: Option<usize>,
) -> Result<String, ServiceManagerError> {
ASYNC_RUNTIME.block_on(async {
let log_path = self.get_log_path(service_name);
if !log_path.exists() {
return Ok(String::new());
}
@@ -359,10 +421,9 @@ impl ServiceManager for LaunchctlServiceManager {
}
fn list(&self) -> Result<Vec<String>, ServiceManagerError> {
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
ASYNC_RUNTIME.block_on(async {
let list_output = self.run_launchctl(&["list"]).await?;
let services: Vec<String> = list_output
.lines()
.filter_map(|line| {
@@ -370,7 +431,9 @@ impl ServiceManager for LaunchctlServiceManager {
// Extract service name from label
line.split_whitespace()
.last()
.and_then(|label| label.strip_prefix(&format!("{}.", self.service_prefix)))
.and_then(|label| {
label.strip_prefix(&format!("{}.", self.service_prefix))
})
.map(|s| s.to_string())
} else {
None
@@ -383,12 +446,18 @@ impl ServiceManager for LaunchctlServiceManager {
}
fn remove(&self, service_name: &str) -> Result<(), ServiceManagerError> {
// Stop the service first
let _ = self.stop(service_name);
// Try to stop the service first, but don't fail if it's already stopped or doesn't exist
if let Err(e) = self.stop(service_name) {
// Log the error but continue with removal
log::warn!(
"Failed to stop service '{}' before removal: {}",
service_name,
e
);
}
// Remove the plist file
let rt = tokio::runtime::Runtime::new().map_err(|e| ServiceManagerError::Other(e.to_string()))?;
rt.block_on(async {
// Remove the plist file using the shared runtime
ASYNC_RUNTIME.block_on(async {
let plist_path = self.get_plist_path(service_name);
if plist_path.exists() {
tokio::fs::remove_file(&plist_path).await?;
@@ -396,4 +465,4 @@ impl ServiceManager for LaunchctlServiceManager {
Ok(())
})
}
}
}