pig2014/honey-biscuit-workshop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Package Manager

Browse Source
This commit is contained in:
Catty Steve
2026-03-11 14:49:10 +08:00
parent c80500f6ac
commit 69056a3a4a
10 changed files with 1111 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AGENTS.md
+21 -99
View File
@@ -1,6 +1,6 @@
# Agent Guidelines for HoneyBiscuitWorkshop
This repository is a monorepo of independent Rust projects using a unified coding style and conventions.
This repository is a monorepo of independent Rust projects using unified coding style and conventions.
## Project Structure
@@ -13,115 +13,72 @@ This repository is a monorepo of independent Rust projects using a unified codin
```bash
# Build specific project
cd workshop-executor && cargo build --release
# Run all tests for a project
cargo test
# Run a single test
cargo test test_function_name
# Run tests in a specific module
cargo test module_name
# Run tests with verbose output
cargo test --verbose
# Run ignored tests
cargo test -- --ignored
# Lint with clippy
cargo clippy -- -D warnings
# Format code
cargo fmt
# Check formatting without making changes
cargo fmt --all -- --check
# Run tests with specific features (RustyVault)
cd RustyVault && cargo test --features sync_handler --verbose
# Generate coverage report
cargo install cargo-tarpaulin
cargo tarpaulin --out Lcov
```
## Code Style Guidelines
### Edition and Rust Version
- All workshop projects use Rust 2024 edition
- RustyVault uses Rust 2021 edition
### Naming Conventions
### Edition and Naming
- Workshop projects: Rust 2024 edition, RustyVault: Rust 2021 edition
- **Crates/Packages**: `workshop-{name}` (snake_case)
- **Modules/Files**: `snake_case` (e.g., `error.rs`, `config.rs`)
- **Structs/Enums**: `PascalCase` (e.g., `VaultError`, `CertificateBundle`)
- **Functions/Methods**: `snake_case` (e.g., `generate_device_id()`, `store_cicd_variable()`)
- **Functions/Methods**: `snake_case` (e.g., `generate_device_id()`)
- **Constants**: `SCREAMING_SNAKE_CASE` (e.g., `EXIT_CODE_OK`, `VERSION`)
### Import Organization
```rust
// Standard library imports first
use std::collections::HashMap;
use std::path::PathBuf;
// External crates (grouped)
use tokio::sync::mpsc;
use serde::{Serialize, Deserialize};
use anyhow::Result;
use log::info;
// Local crate modules
use crate::module_name::Type;
```
### Module Structure
### Module Structure and Error Handling
```rust
// lib.rs or main.rs
mod config;
mod error;
mod handler;
mod state;
// Re-export commonly used types
pub use config::*;
pub use error::*;
```
### Error Handling
Use `thiserror` for custom error types and `anyhow` for generic error handling:
```rust
use thiserror::Error;
#[derive(Error, Debug)]
pub enum VaultError {
#[error("Vault operation failed: {0}")]
OperationFailed(String),
#[error("Authentication failed: {0}")]
AuthFailed(String),
#[error("Serialization error: {0}")]
SerializationError(#[from] serde_json::Error),
}
// Create type alias for convenience
pub type Result<T> = std::result::Result<T, VaultError>;
// Implement From trait for error conversions
impl From<std::io::Error> for VaultError {
fn from(err: std::io::Error) -> Self {
VaultError::IoError(err.to_string())
}
}
```
### Async Patterns
### Async Patterns and Serialization
Use `tokio` with "full" features for async code:
```rust
#[tokio::main]
async fn main() {
@@ -129,65 +86,27 @@ async fn main() {
// async code...
}
#[tokio::test]
async fn test_async_function() {
// test code...
}
```
### Serialization
Use `serde` derive macros with `#[serde(rename)]` for reserved words:
```rust
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Artifact {
pub name: String,
#[serde(rename = "type")]
pub type_: String,
pub path: String,
}
```
### Documentation
### Documentation and Testing
Add doc comments (`///`) for all public items:
Add doc comments (`///`) for all public items. Place unit tests in `#[cfg(test)]` modules within source files.
Use `#[tokio::test]` for async tests, `#[ignore]` for tests requiring external resources. Use `maybe_async` crate for conditional async tests:
```rust
/// Creates a new Vault client with the specified address and token
pub fn create_vault_client(address: String, token: String) -> Result<RustyVaultClient> {
// implementation...
#[maybe_async::test(feature = "sync_handler", async(all(not(feature = "sync_handler")), tokio::test))]
async fn test_with_feature_flag() {
// test code...
}
```
### Testing
- Place unit tests in `#[cfg(test)]` modules within source files
- Use `#[tokio::test]` for async tests
- Use `#[ignore]` for tests requiring external resources (e.g., Vault instances)
- Use descriptive test names
```rust
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_device_id_generation() {
let device_id = DeviceIdGenerator::generate_device_id().unwrap();
assert_eq!(device_id.len(), 64);
}
#[tokio::test]
#[ignore]
async fn test_vault_operations() {
let client = create_vault_client_from_env().unwrap();
// test code...
}
}
```
### Formatting
### Formatting and Dependencies
Follow rustfmt settings from RustyVault:
- Max width: 120 characters
@@ -196,8 +115,7 @@ Follow rustfmt settings from RustyVault:
- Imports granularity: `Crate`
- Trailing comma: `Vertical`
### Common Dependencies
Common dependencies:
- `tokio` (v1.40-1.48) with "full" features
- `serde` and `serde_json` for serialization
- `thiserror` for custom errors
@@ -221,3 +139,7 @@ Use similar allowances for workshop projects when appropriate.
Use environment variable prefixes for configuration:
- `WSEXECUTOR_` for workshop-executor
- Common pattern: `env_logger::init()` at the start of `main()`
### Cursor/Copilot Rules
No Cursor or Copilot rules found in the repository.
workshop-executor/Cargo.lock
Generated
+197
View File
@@ -326,6 +326,15 @@ dependencies = [
"generic-array",
]
[[package]]
name = "block2"
version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cdeb9d870516001442e364c5220d3574d2da8dc765554b4a617230d33fa58ef5"
dependencies = [
"objc2",
]
[[package]]
name = "blocking"
version = "1.6.2"
@@ -649,6 +658,16 @@ dependencies = [
"crypto-common",
]
[[package]]
name = "dispatch2"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1e0e367e4e7da84520dedcac1901e4da967309406d1e51017ae1abfb97adbd38"
dependencies = [
"bitflags 2.10.0",
"objc2",
]
[[package]]
name = "displaydoc"
version = "0.2.5"
@@ -1583,6 +1602,165 @@ dependencies = [
"autocfg",
]
[[package]]
name = "objc2"
version = "0.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3a12a8ed07aefc768292f076dc3ac8c48f3781c8f2d5851dd3d98950e8c5a89f"
dependencies = [
"objc2-encode",
]
[[package]]
name = "objc2-cloud-kit"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73ad74d880bb43877038da939b7427bba67e9dd42004a18b809ba7d87cee241c"
dependencies = [
"bitflags 2.10.0",
"objc2",
"objc2-foundation",
]
[[package]]
name = "objc2-core-data"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0b402a653efbb5e82ce4df10683b6b28027616a2715e90009947d50b8dd298fa"
dependencies = [
"objc2",
"objc2-foundation",
]
[[package]]
name = "objc2-core-foundation"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
dependencies = [
"bitflags 2.10.0",
"dispatch2",
"objc2",
]
[[package]]
name = "objc2-core-graphics"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e022c9d066895efa1345f8e33e584b9f958da2fd4cd116792e15e07e4720a807"
dependencies = [
"bitflags 2.10.0",
"dispatch2",
"objc2",
"objc2-core-foundation",
"objc2-io-surface",
]
[[package]]
name = "objc2-core-image"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e5d563b38d2b97209f8e861173de434bd0214cf020e3423a52624cd1d989f006"
dependencies = [
"objc2",
"objc2-foundation",
]
[[package]]
name = "objc2-core-location"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca347214e24bc973fc025fd0d36ebb179ff30536ed1f80252706db19ee452009"
dependencies = [
"objc2",
"objc2-foundation",
]
[[package]]
name = "objc2-core-text"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0cde0dfb48d25d2b4862161a4d5fcc0e3c24367869ad306b0c9ec0073bfed92d"
dependencies = [
"bitflags 2.10.0",
"objc2",
"objc2-core-foundation",
"objc2-core-graphics",
]
[[package]]
name = "objc2-encode"
version = "4.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ef25abbcd74fb2609453eb695bd2f860d389e457f67dc17cafc8b8cbc89d0c33"
[[package]]
name = "objc2-foundation"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e3e0adef53c21f888deb4fa59fc59f7eb17404926ee8a6f59f5df0fd7f9f3272"
dependencies = [
"bitflags 2.10.0",
"block2",
"libc",
"objc2",
"objc2-core-foundation",
]
[[package]]
name = "objc2-io-surface"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "180788110936d59bab6bd83b6060ffdfffb3b922ba1396b312ae795e1de9d81d"
dependencies = [
"bitflags 2.10.0",
"objc2",
"objc2-core-foundation",
]
[[package]]
name = "objc2-quartz-core"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "96c1358452b371bf9f104e21ec536d37a650eb10f7ee379fff67d2e08d537f1f"
dependencies = [
"bitflags 2.10.0",
"objc2",
"objc2-core-foundation",
"objc2-foundation",
]
[[package]]
name = "objc2-ui-kit"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d87d638e33c06f577498cbcc50491496a3ed4246998a7fbba7ccb98b1e7eab22"
dependencies = [
"bitflags 2.10.0",
"block2",
"objc2",
"objc2-cloud-kit",
"objc2-core-data",
"objc2-core-foundation",
"objc2-core-graphics",
"objc2-core-image",
"objc2-core-location",
"objc2-core-text",
"objc2-foundation",
"objc2-quartz-core",
"objc2-user-notifications",
]
[[package]]
name = "objc2-user-notifications"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9df9128cbbfef73cda168416ccf7f837b62737d748333bfe9ab71c245d76613e"
dependencies = [
"objc2",
"objc2-foundation",
]
[[package]]
name = "once_cell"
version = "1.21.3"
@@ -1635,6 +1813,22 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "os_info"
version = "3.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e4022a17595a00d6a369236fdae483f0de7f0a339960a53118b818238e132224"
dependencies = [
"android_system_properties",
"log",
"nix 0.30.1",
"objc2",
"objc2-foundation",
"objc2-ui-kit",
"serde",
"windows-sys 0.61.2",
]
[[package]]
name = "parking"
version = "2.2.1"
@@ -3182,6 +3376,7 @@ name = "workshop-executor"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"bollard",
"cgroups-rs",
"chrono",
@@ -3190,8 +3385,10 @@ dependencies = [
"env_logger",
"futures-util",
"lazy_static",
"libc",
"log",
"minijinja",
"os_info",
"regex",
"semver",
"serde",
workshop-executor/Cargo.toml
+3
View File
@@ -26,4 +26,7 @@ tempfile = "3.24.0"
thiserror = "2.0.17"
tokio = { version = "1.48.0", features = ["full"] }
topological-sort = "0.2.2"
async-trait = "0.1.87"
uuid = { version = "1.19.0", features = ["v4"] }
libc = "0.2"
os_info = "3.14.0"
workshop-executor/src/engine.rs
-229
View File
@@ -1,229 +0,0 @@
use crate::{decorator::Decorator, parser::Function};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use std::time::Duration;
use tokio::process::Command;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUsage {
pub cpu_time_secs: f64,
pub max_memory_kb: u64,
pub io_read_kb: u64,
pub io_write_kb: u64,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
pub struct RetryPolicy {
pub max_attempts: usize,
pub backoff_ms: u64,
pub jitter: bool,
}
#[derive(Debug, Clone)]
pub struct ExecutionContext {
pub script_path: PathBuf,
pub pipeline_name: String,
pub username: String,
pub task_id: String,
pub working_dir: PathBuf,
pub env_vars: HashMap<String, String>,
pub timeout: Option<Duration>,
pub retry_policy: RetryPolicy,
}
#[derive(Debug, Clone, Serialize)]
pub struct ExecutionResult {
pub task_id: String,
pub exit_code: u32,
pub success: bool,
pub duration: Duration,
pub stdout_bytes: usize,
pub stderr_bytes: usize,
pub new_env_vars: HashMap<String, String>,
pub resource_usage: Option<ResourceUsage>,
}
#[derive(Debug, Clone, Serialize)]
pub enum StreamType {
Stdout,
Stderr,
}
#[derive(Debug, Clone, Serialize)]
pub enum ExecutionEvent {
TaskStarted {
task_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
OutputChunk {
task_id: String,
stream: StreamType,
data: Vec<u8>,
is_partial: bool,
},
ProgressUpdate {
task_id: String,
progress: f32,
message: String,
},
EnvUpdate {
task_id: String,
key: String,
value: String,
},
TaskCompleted {
task_id: String,
result: ExecutionResult,
},
TaskFailed {
task_id: String,
error: String,
retryable: bool,
attempt: usize,
},
}
pub async fn execute(
script_path: PathBuf,
pipeline_name: String,
username: String,
function: Function,
working_dir: Option<PathBuf>,
env_vars: HashMap<String, String>,
event_tx: Option<tokio::sync::mpsc::Sender<ExecutionEvent>>,
) -> Result<ExecutionResult, Box<dyn std::error::Error + Send + Sync>> {
let mut retry_policy = RetryPolicy {
max_attempts: 1,
backoff_ms: 0,
jitter: false,
};
let retry_config = function.find_decorator(|d| match d {
Decorator::Retry(count, interval) => Some((*count, *interval)), // 解引用复制值
_ => None,
});
if let Some((count, interval)) = retry_config {
retry_policy.max_attempts = count;
retry_policy.backoff_ms = interval as u64;
} else {
retry_policy.max_attempts = 1;
retry_policy.backoff_ms = 0;
}
let timeout = function.find_decorator(|d| match d {
Decorator::Timeout(seconds) => Some(Duration::from_secs(*seconds as u64)),
_ => None,
});
let working_dir = working_dir.unwrap_or_else(|| script_path.parent().unwrap().to_path_buf());
let context = ExecutionContext {
script_path,
pipeline_name,
username,
task_id: function.name.clone(),
working_dir,
env_vars,
timeout,
retry_policy,
};
let mut attempt = 1;
// let start_time = std::time::Instant::now();
loop {
if let Some(tx) = &event_tx {
let _ = tx.send(ExecutionEvent::TaskStarted {
task_id: function.name.clone(),
timestamp: chrono::Utc::now(),
});
}
match execute_single(&context, attempt, event_tx.clone()).await {
Ok(result) => {
log::info!("Task {} succeeded on attempt {}", context.task_id, attempt);
return Ok(result);
}
Err((error, retryable)) => {
if let Some(tx) = &event_tx {
let _ = tx.send(ExecutionEvent::TaskFailed { task_id: context.task_id.clone(), error:error.clone(), retryable, attempt }).await;
}
if retryable && attempt < retry_policy.max_attempts {
attempt += 1;
log::info!("Task {} failed on attempt {}, retrying...", context.task_id, attempt);
tokio::time::sleep(Duration::from_millis(retry_policy.backoff_ms)).await;
continue;
} else {
return Err(format!("Task {} failed after {} attempts", context.task_id, attempt).into());
}
}
}
}
}
async fn execute_single(
context: &ExecutionContext,
attempt: usize,
event_tx: Option<tokio::sync::mpsc::Sender<ExecutionEvent>>,
) -> Result<ExecutionResult, (String, bool)> {
let start_time = std::time::Instant::now();
let mut command = Command::new("bash"); // TODO: 换成可变的解释器
command.current_dir(&context.working_dir);
command.envs(&context.env_vars);
command.env("WS_TASKID", &context.task_id);
command.env("WS_PIPELINE", &context.pipeline_name);
command.env("WS_ATTEMPT", attempt.to_string());
command.arg(&context.script_path); // TODO: 适配可变的解释器
let child = match command.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
{
Ok(child) => child,
Err(err) => {
return Err((format!("Failed to spawn command: {}", err), false));
}
};
let pid = child.id();
log::debug!("Task {} started with PID {}", context.task_id, pid.unwrap_or(0));
todo!();
/*
let output_handler = OutputHandler::new(); // TODO: 完成OutputHandler
let result = if let Some(timeout) = context.timeout && timeout.as_secs() > 0{
tokio::time::timeout(timeout, output_handler.wait()).await
} else {
Ok(output_handler.wait().await)
};
match result.flatten() {
Ok((exit_code, stdout_len, stderr_len, new_env)) => {
let duration = start_time.elapsed();
let result = ExecutionResult {
task_id: context.task_id.clone(),
exit_code,
success: exit_code == 0,
duration,
stdout_bytes: stdout_len,
stderr_bytes: stderr_len,
new_env_vars: new_env,
resource_usage: None,
};
if let Some(tx) = &event_tx {
let _ = tx.send(ExecutionEvent::TaskCompleted {
task_id: context.task_id.clone(),
result: result.clone(),
}).await;
}
return Ok(result)
}
Err(e) => {
return Err((format!("Execution error: {}", e), true))
}
}
*/
}
workshop-executor/src/engine/cgroups.rs
+105
View File
@@ -0,0 +1,105 @@
use std::fs;
use std::io::{self, Write};
use std::path::PathBuf;
pub struct CgroupManager {
path: PathBuf,
}
impl CgroupManager {
pub fn new(build_id: &str) -> io::Result<Self> {
let path = PathBuf::from(format!("/sys/fs/cgroup/ws-executor/{}", build_id));
fs::create_dir_all(&path)?;
Ok(Self { path })
}
pub fn set_memory_limit(&self, limit_bytes: u64) -> io::Result<()> {
let mem_max_path = self.path.join("memory.max");
let content = if limit_bytes == 0 {
"max".to_string()
} else {
limit_bytes.to_string()
};
fs::write(&mem_max_path, content)?;
let mem_swap_path = self.path.join("memory.swap.max");
fs::write(&mem_swap_path, "max")?;
Ok(())
}
pub fn set_cpu_weight(&self, weight: u64) -> io::Result<()> {
let cpu_weight_path = self.path.join("cpu.weight");
let weight = weight.max(1).min(10000);
fs::write(&cpu_weight_path, weight.to_string())?;
Ok(())
}
pub fn set_cpu_max(&self, max_us: u64, period_us: u64) -> io::Result<()> {
let cpu_max_path = self.path.join("cpu.max");
let content = if max_us == 0 {
"max".to_string()
} else {
format!("{} {}", max_us, period_us)
};
fs::write(&cpu_max_path, content)?;
Ok(())
}
pub fn add_process(&self, pid: u32) -> io::Result<()> {
let cgroup_procs_path = self.path.join("cgroup.procs");
let mut file = fs::OpenOptions::new()
.write(true)
.append(true)
.open(&cgroup_procs_path)?;
writeln!(file, "{}", pid)?;
Ok(())
}
pub fn destroy(&self) -> io::Result<()> {
if self.path.exists() {
fs::remove_dir(&self.path)?;
}
Ok(())
}
pub fn path(&self) -> &PathBuf {
&self.path
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
#[ignore]
fn test_cgroup_manager_new() {
let build_id = "test-build-123";
let manager = CgroupManager::new(build_id).unwrap();
assert!(manager.path().exists());
manager.destroy().unwrap();
}
#[test]
#[ignore]
fn test_set_memory_limit() {
let build_id = "test-mem-123";
let manager = CgroupManager::new(build_id).unwrap();
manager.set_memory_limit(1024 * 1024 * 1024).unwrap();
manager.destroy().unwrap();
}
#[test]
#[ignore]
fn test_set_cpu_weight() {
let build_id = "test-cpu-123";
let manager = CgroupManager::new(build_id).unwrap();
manager.set_cpu_weight(1024).unwrap();
manager.destroy().unwrap();
}
}
workshop-executor/src/engine/mod.rs
+255
View File
@@ -0,0 +1,255 @@
pub mod cgroups;
pub mod pm;
use std::collections::HashMap;
use std::path::PathBuf;
use std::process::Stdio;
use std::time::Duration;
use tokio::process::Command;
use tokio::sync::mpsc;
use serde::{Deserialize, Serialize};
use chrono::Utc;
pub use crate::engine::cgroups::CgroupManager;
use crate::engine::pm::PackageManager;
// pub use crate::engine::pm::{PackageManager};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUsage {
pub cpu_time_secs: f64,
pub max_memory_kb: u64,
pub io_read_kb: u64,
pub io_write_kb: u64,
}
#[derive(Debug, Clone)]
pub struct ExecutionContext {
pub task_id: String,
pub pipeline_name: String,
pub username: String,
pub euid: u32,
pub egid: u32,
pub working_dir: PathBuf,
pub env_vars: HashMap<String, String>,
pub timeout: Option<Duration>,
pub cgroup_path: Option<PathBuf>,
pub shell: String,
}
impl Default for ExecutionContext {
fn default() -> Self {
Self {
task_id: String::new(),
pipeline_name: String::new(),
username: String::new(),
euid: unsafe { libc::geteuid() },
egid: unsafe { libc::getegid() },
working_dir: std::env::current_dir().unwrap_or_else(|_| PathBuf::from("/")),
env_vars: HashMap::new(),
timeout: None,
cgroup_path: None,
shell: String::from("bash"),
}
}
}
#[derive(Debug, Clone, Default)]
pub struct ResourceLimits {
pub memory_bytes: Option<u64>,
pub cpu_weight: Option<u64>,
pub cpu_max_us: Option<u64>,
pub cpu_period_us: u64,
}
#[derive(Debug, Clone, Serialize)]
pub struct ExecutionResult {
pub task_id: String,
pub exit_code: i32,
pub success: bool,
pub duration: Duration,
pub stdout: String,
pub stderr: String,
pub resource_usage: Option<ResourceUsage>,
}
#[derive(Debug, Clone, Serialize)]
pub enum StreamType {
Stdout,
Stderr,
}
#[derive(Debug, Clone, Serialize)]
pub enum ExecutionEvent {
TaskStarted {
task_id: String,
timestamp: chrono::DateTime<chrono::Utc>,
},
OutputChunk {
task_id: String,
stream: StreamType,
data: String,
},
TaskCompleted {
task_id: String,
result: ExecutionResult,
},
TaskFailed {
task_id: String,
error: String,
},
}
pub type EventSender = Option<mpsc::Sender<ExecutionEvent>>;
#[derive(Debug, Clone, Serialize)]
pub enum ExecutionError {
InvalidCommand(String),
ExecutionFailed(String),
PermissionDenied { euid: u32, egid: u32, reason: String },
Timeout,
IoError(String),
ScriptNotFound(PathBuf),
PackageManagerNotFound,
}
pub struct LocalExecutor {
cgroup_manager: Option<CgroupManager>,
}
impl LocalExecutor {
pub fn new() -> Self {
Self { cgroup_manager: None }
}
pub fn with_cgroup(limits: ResourceLimits, build_id: &str) -> Result<Self, String> {
let cgroup_manager = CgroupManager::new(build_id)
.map_err(|e| format!("Failed to create cgroup: {}", e))?;
if let Some(mem) = limits.memory_bytes {
cgroup_manager.set_memory_limit(mem)
.map_err(|e| format!("Failed to set memory limit: {}", e))?;
}
if let Some(weight) = limits.cpu_weight {
cgroup_manager.set_cpu_weight(weight)
.map_err(|e| format!("Failed to set cpu weight: {}", e))?;
}
if let Some(max) = limits.cpu_max_us {
cgroup_manager.set_cpu_max(max, limits.cpu_period_us)
.map_err(|e| format!("Failed to set cpu max: {}", e))?;
}
Ok(Self { cgroup_manager: Some(cgroup_manager) })
}
fn apply_cgroup(&self, pid: u32) -> Result<(), ExecutionError> {
if let Some(ref cgroup) = self.cgroup_manager {
cgroup.add_process(pid)
.map_err(|e| ExecutionError::IoError(format!("Failed to add process to cgroup: {}", e)))?;
}
Ok(())
}
pub async fn execute_script(&self, script_path: &PathBuf, ctx: &ExecutionContext, event_tx: EventSender) -> Result<ExecutionResult, ExecutionError> {
let abs_path = if script_path.is_absolute() {
script_path.clone()
} else {
ctx.working_dir.join(script_path)
};
let mut command = Command::new(ctx.shell.as_str());
command.arg("-c");
command.arg(abs_path);
self.execute(command, ctx, event_tx).await
}
pub async fn execute_command(&self, cmd: String, ctx: &ExecutionContext, event_tx: EventSender) -> Result<ExecutionResult, ExecutionError> {
let mut command = Command::new(ctx.shell.as_str());
command.arg("-c");
command.arg(cmd);
self.execute(command, ctx, event_tx).await
}
pub async fn execute(&self, cmd: Command, ctx: &ExecutionContext, event_tx: EventSender) -> Result<ExecutionResult, ExecutionError> {
let start_time = std::time::Instant::now();
if let Some(ref tx) = event_tx {
let _ = tx.send(ExecutionEvent::TaskStarted {
task_id: ctx.task_id.clone(),
timestamp: Utc::now(),
}).await;
}
let mut command = cmd;
command.current_dir(&ctx.working_dir);
command.envs(&ctx.env_vars);
command.env("WS_TASKID", &ctx.task_id);
command.env("WS_PIPELINE", &ctx.pipeline_name);
command.env("WS_EUID", ctx.euid.to_string());
command.env("WS_EGID", ctx.egid.to_string());
command.env("WS_USERNAME", &ctx.username);
command.stdin(Stdio::null());
command.stdout(Stdio::piped());
command.stderr(Stdio::piped());
let child = command.spawn()
.map_err(|e| ExecutionError::InvalidCommand(format!("Failed to spawn command: {}", e)))?;
let pid = child.id().ok_or_else(|| ExecutionError::IoError("Failed to get PID".to_string()))?;
self.apply_cgroup(pid)?;
let output = child.wait_with_output()
.await
.map_err(|e| ExecutionError::IoError(format!("Failed to wait for process: {}", e)))?;
let exit_code = output.status.code().unwrap_or(-1);
let duration = start_time.elapsed();
let result = ExecutionResult {
task_id: ctx.task_id.clone(),
exit_code,
success: exit_code == 0,
duration,
stdout: String::from_utf8_lossy(&output.stdout).to_string(),
stderr: String::from_utf8_lossy(&output.stderr).to_string(),
resource_usage: None,
};
if let Some(ref tx) = event_tx {
let event = if result.success {
ExecutionEvent::TaskCompleted {
task_id: ctx.task_id.clone(),
result: result.clone(),
}
} else {
ExecutionEvent::TaskFailed {
task_id: ctx.task_id.clone(),
error: format!("Exit code: {}", exit_code),
}
};
let _ = tx.send(event).await;
}
Ok(result)
}
pub async fn install_dependency(&self, package: &[&str], pkgman: &dyn PackageManager, ctx: &ExecutionContext) -> Result<ExecutionResult, ExecutionError> {
let cmd = pkgman.install(package);
self.execute(cmd, ctx, None).await
}
pub async fn cleanup(&self) -> Result<(), String> {
if let Some(ref cgroup) = self.cgroup_manager {
cgroup.destroy()
.map_err(|e| format!("Failed to destroy cgroup: {}", e))?;
}
Ok(())
}
}
impl Default for LocalExecutor {
fn default() -> Self {
Self::new()
}
}
workshop-executor/src/engine/pm/apt.rs
+219
View File
@@ -0,0 +1,219 @@
use crate::engine::pm::PackageManager;
use os_info::{Info, Version};
use tokio::process::Command;
#[derive(Clone)]
pub struct Apt;
impl PackageManager for Apt {
fn name(&self) -> &'static str {
"apt"
}
fn adopt(&self, distro: &str) -> bool {
// 匹配所有主要的使用 apt 的发行版
// 注意：os_info 返回的 Type 转字符串通常是全小写或特定格式，这里做宽松匹配
let d = distro.to_lowercase();
matches!(
d.as_str(),
"debian"
| "ubuntu"
| "linuxmint"
| "mint"
| "pop"
| "pop!_os"
| "kali"
| "raspbian"
| "aosc"
| "deepin"
| "uos"
| "elementary"
)
}
fn binary(&self) -> &'static str {
"apt"
}
fn update(&self) -> Command {
let mut cmd = Command::new("apt");
cmd.arg("update");
cmd
}
fn install(&self, package: &[&str]) -> Command {
let mut cmd = Command::new("apt");
cmd.arg("install");
cmd.arg("-y"); // 自动确认
cmd.args(package);
cmd
}
fn change_mirror(&self, repo: &str, osinfo: Option<Info>) -> Command {
// 目标仓库 URL (例如: https://mirrors.tuna.tsinghua.edu.cn/debian)
// 我们需要构建一个 sed 命令来替换 /etc/apt/sources.list 中的旧 URL
let codename = extract_codename(osinfo.as_ref()).unwrap_or_else(|| {
log::warn!("无法自动获取版本代号，将尝试替换所有行，请谨慎使用。");
String::new()
});
// 构造 sed 表达式
// 逻辑：找到包含当前 codename 或 generic 的行，替换掉其中的 http/https 域名部分
// 为了安全，我们通常建议备份，但这里只生成命令，用户执行前可自行备份或使用 --backup
// 简单的策略：替换 sources.list 中第一个出现的 http/https 开头直到第一个空格的部分
// 更稳健的策略是针对特定发行版模板，但通用 sed 更灵活
let mut cmd = Command::new("sed");
// 注意：sed -i 需要权限，通常由用户 sudo 运行此命令
cmd.arg("-i.bak"); // 自动备份为 .bak
if codename.is_empty() {
// 如果不知道代号，尝试宽泛替换 (风险较高，仅替换常见源)
// 匹配 http://... 或 https://... 直到空格
cmd.arg("-E");
cmd.arg(format!(
r"s|^https?://[^\s]+|{}|g",
repo.trim_end_matches('/')
));
} else {
// 如果知道代号，尽量只替换包含该代号的行，减少误伤
// 匹配： deb http://old-url codename ... -> deb new-repo codename ...
cmd.arg("-E");
cmd.arg(format!(
r"s|(deb\s+https?://)[^\s]+(\s+{}\s)|\1{}{}\2|g",
codename,
repo.trim_end_matches('/'),
if codename.is_empty() { "" } else { "" } // 占位，实际逻辑在 regex 中
));
// 修正上面的正则逻辑，使其更准确：
// 捕获组1: "deb http://"
// 匹配中间: [^\s]+ (旧域名)
// 捕获组2: " codename " (包含代号的部分)
// 替换为: \1 + 新域名 + \2
// 重新构建更准确的 sed 参数
cmd = Command::new("sed");
cmd.arg("-i.bak");
cmd.arg("-E");
let pattern = format!(
r"s#(deb\s+https?://)[^\s]+(\s+{}(\s|$))#\1{}\2#g",
codename,
repo.trim_end_matches('/')
);
cmd.arg(pattern);
}
cmd.arg("/etc/apt/sources.list");
// 提示：对于 /etc/apt/sources.list.d/ 下的文件，此命令不会处理，通常需要额外逻辑
// 但作为基础实现，修改主文件是最常见的操作
cmd
}
fn add_repository(&self, repo: &str, osinfo: Option<Info>) -> Option<Command> {
// 策略 1: 优先尝试使用 add-apt-repository (来自 software-properties-common)
// 这个命令会自动处理 GPG key 和 list 文件
let mut cmd_ppa = Command::new("add-apt-repository");
cmd_ppa.arg("-y");
cmd_ppa.arg(repo);
// 我们无法在这里完美检测 add-apt-repository 是否存在而不执行
// 但我们可以假设如果它是 Ubuntu/Debian 标准环境，可能有用。
// 为了稳妥，我们返回一个 sh -c 脚本，先检查命令是否存在，不存在则 fallback 到手动写入
let codename = extract_codename(osinfo.as_ref())?;
// 构建一个 shell 脚本来处理逻辑
let script = format!(
r#"
if command -v add-apt-repository >/dev/null 2>&1; then
add-apt-repository -y "{repo}"
else
echo "deb {repo} {codename} main" | tee /etc/apt/sources.list.d/custom-repo.list
fi
"#,
repo = repo,
codename = codename
);
let mut cmd = Command::new("sh");
cmd.arg("-c").arg(script);
Some(cmd)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_adopt_true_for_ubuntu_and_debian() {
let apt = Apt;
assert!(apt.adopt("ubuntu"));
assert!(apt.adopt("debian"));
assert!(!apt.adopt("fedora"));
}
#[test]
fn test_update_command_contents() {
let cmd = Apt.update();
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "apt");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["update"]);
}
#[test]
fn test_install_command_contents() {
let cmd = Apt.install(&["curl"]);
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "apt");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["install", "-y", "curl"]);
}
}
/// 辅助函数：从 os_info 中提取 Codename
fn extract_codename(info: Option<&Info>) -> Option<String> {
let info = info?;
// 1. 直接获取 codename (os_info 较新版本支持)
if let Some(codename) = info.codename() {
if !codename.is_empty() {
return Some(codename.to_string());
}
}
// 2. 尝试从 version 中推断 (针对某些把 codename 放在 Custom 里的情况)
match info.version() {
Version::Custom(v) => {
// 有些系统可能把 codename 放在这里，虽然不标准
if !v.chars().next().map(|c| c.is_numeric()).unwrap_or(true) {
return Some(v.clone());
}
}
Version::Semantic(major, minor, _) => {
// 对于 Ubuntu/Debian，如果没有 codename，有时可以用版本号兜底 (如 22.04)
// 但 apt 源通常更喜欢 codename。如果实在没有，返回版本号字符串作为最后手段
return Some(format!("{}.{}", major, minor));
}
_ => {}
}
// 3. 如果 os_info 没提供，且是常见发行版，可以尝试根据 Type 猜测？
// 不推荐硬编码猜测，因为版本太多。
// 最好返回 None，让调用者决定是报错还是让用户手动指定。
None
}
workshop-executor/src/engine/pm/dnf.rs
+133
View File
@@ -0,0 +1,133 @@
use crate::engine::pm::PackageManager;
use os_info::Info;
use tokio::process::Command;
#[derive(Clone)]
pub struct Dnf;
impl PackageManager for Dnf {
fn name(&self) -> &'static str {
"dnf"
}
fn adopt(&self, distro: &str) -> bool {
let d = distro.to_lowercase();
matches!(
d.as_str(),
"fedora" | "rhel" | "centos" | "almalinux" | "rocky"
)
}
fn binary(&self) -> &'static str {
"dnf"
}
fn update(&self) -> Command {
let mut cmd = Command::new("dnf");
cmd.arg("makecache");
cmd
}
fn install(&self, package: &[&str]) -> Command {
let mut cmd = Command::new("dnf");
cmd.arg("install");
cmd.arg("-y"); // auto-confirm
cmd.args(package);
cmd
}
fn change_mirror(&self, repo: &str, _osinfo: Option<Info>) -> Command {
// Change baseurl in yum repos under /etc/yum.repos.d
// This is a best-effort: replace baseurl with the provided repo URL
let mut cmd = Command::new("sed");
cmd.arg("-i.bak");
cmd.arg(format!("s#baseurl=.*#{}#g", repo));
// The target files
cmd.arg("/etc/yum.repos.d/*.repo");
cmd
}
fn add_repository(&self, repo: &str, _osinfo: Option<Info>) -> Option<Command> {
// Append a minimal repo file under /etc/yum.repos.d
let script = format!(
"printf '%s' '[custom]\nname=Custom Repo\nbaseurl={}%\nenabled=1\n' >> /etc/yum.repos.d/custom.repo",
repo
);
// Run via bash -lc to avoid escaping issues
let mut cmd = Command::new("bash");
cmd.arg("-lc").arg(script);
Some(cmd)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_adopt_fedora_family() {
let dnf = Dnf;
assert!(dnf.adopt("Fedora"));
assert!(dnf.adopt("RHEL"));
assert!(dnf.adopt("centos"));
assert!(dnf.adopt("AlmaLinux"));
assert!(dnf.adopt("Rocky"));
assert!(!dnf.adopt("ubuntu"));
}
#[test]
fn test_update_command() {
let cmd = Dnf.update();
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "dnf");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["makecache"]);
}
#[test]
fn test_install_command() {
let cmd = Dnf.install(&["git", "curl"]);
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "dnf");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["install", "-y", "git", "curl"]);
}
#[test]
fn test_change_mirror_command_basic() {
// Provide a dummy repo url and verify the command program and a couple of args
let cmd = Dnf.change_mirror("https://example.repo/", None);
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "sed");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
// Should include -i.bak and the target path
assert!(args.contains(&"-i.bak".to_string()));
assert!(args.iter().any(|s| s.contains("/etc/yum.repos.d/*.repo")));
}
#[test]
fn test_add_repository_command_exists() {
let maybe_cmd = Dnf.add_repository("https://example.repo/", None);
assert!(maybe_cmd.is_some());
let cmd = maybe_cmd.unwrap();
let cmd = cmd.as_std();
// The command should be a shell invocation executing a script
assert_eq!(cmd.get_program().to_string_lossy(), "bash");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
// Ensure the script contains the repo URL
assert!(args.iter().any(|s| s.contains("https://example.repo/")));
}
}
workshop-executor/src/engine/pm/mod.rs
+54
View File
@@ -0,0 +1,54 @@
use os_info::Info;
use tokio::process::Command;
mod apt;
mod dnf;
mod pacman;
pub trait PackageManager {
/// Name of package manager
fn name(&self) -> &'static str;
/// Adopt for a distro
fn adopt(&self, distro: &str) -> bool;
/// Binary name
fn binary(&self) -> &'static str;
/// The command of updating
fn update(&self) -> Command;
/// The command of installing packages
fn install(&self, package: &[&str]) -> Command;
// Check whether a package is installed
// Currently not planned
// Package manager should handle installed packages themselves
// just as --needed in pacman
// Also the PM module should generate command, not execute them
// fn is_installed(&self, _package: &str) -> Option<bool> {unimplemented!()}
/// Change major mirror of package manager
fn change_mirror(&self, repo: &str, osinfo: Option<Info>) -> Command;
/// add a custom repository
fn add_repository(&self, repo: &str, osinfo: Option<Info>) -> Option<Command>;
}
pub fn all_managers() -> Vec<Box<dyn PackageManager>> {
vec![
Box::new(apt::Apt),
Box::new(pacman::Pacman),
Box::new(dnf::Dnf)
]
}
pub fn detect(distro: os_info::Type) -> Option<Box<dyn PackageManager>> {
let distro_str = distro.to_string();
all_managers().into_iter().find(|pm| pm.adopt(&distro_str))
}
pub fn select(manager: &str) -> Option<Box<dyn PackageManager>> {
log::warn!("Manually select package manager is not recommended.");
log::warn!("Use at your own risk.");
all_managers().into_iter().find(|pm| pm.name() == manager)
}
workshop-executor/src/engine/pm/pacman.rs
+124
View File
@@ -0,0 +1,124 @@
use crate::engine::pm::PackageManager;
use os_info::Info;
use tokio::process::Command;
#[derive(Clone)]
pub struct Pacman;
impl PackageManager for Pacman {
fn name(&self) -> &'static str {
"pacman"
}
fn adopt(&self, distro: &str) -> bool {
let d = distro.to_lowercase();
matches!(
d.as_str(),
"arch"
| "manjaro"
| "endeavouros"
| "garuda"
| "cachyos"
| "artix"
| "mabox"
| "nobara"
| "instantos"
| "archlinux"
)
}
fn binary(&self) -> &'static str {
"pacman"
}
fn update(&self) -> Command {
let mut cmd = Command::new("pacman");
cmd.arg("-Sy"); // 仅刷新数据库，不升级包（通常换源后需要这一步）
cmd
}
fn install(&self, package: &[&str]) -> Command {
let mut cmd = Command::new("pacman");
cmd.arg("-S"); // 同步/安装
cmd.arg("--noconfirm"); // 自动确认
cmd.arg("--needed");
cmd.args(package);
cmd
}
fn change_mirror(&self, repo: &str, osinfo: Option<Info>) -> Command {
let mirrorlist_path = "/etc/pacman.d/mirrorlist";
let content = format!(
"# Generated by WSExecutor-PM\n\
# Date: $(date)\n\
# Original content overwritten.\n\
\n\
Server = {}\n",
repo
);
// echo "content" | sudo tee /path/to/file > /dev/null
let safe_content = content.replace("'", "'\"'\"'"); // 转义单引号: ' -> '\''
let mut cmd = Command::new("sh");
cmd.arg("-c").arg(format!(
"printf '%s' '{}' | sudo tee {} > /dev/null",
safe_content, mirrorlist_path
));
cmd
}
fn add_repository(&self, repo: &str, osinfo: Option<Info>) -> Option<Command> {
let pacman_conf = "/etc/pacman.conf";
let content = format!(
"[custom]\nServer = {}\nSigLevel = Optional TrustAll\n",
repo,
);
let safe_content = content.replace("'", "'\"'\"'");
let mut cmd = Command::new("sh");
cmd.arg("-c")
.arg(format!("echo '{}' >> {}", safe_content, pacman_conf));
Some(cmd)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_adopt_arch_like() {
let pac = Pacman;
assert!(pac.adopt("arch"));
assert!(pac.adopt("manjaro"));
assert!(!pac.adopt("ubuntu"));
}
#[test]
fn test_update_command_contents() {
let cmd = Pacman.update();
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "pacman");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["-Sy"]);
}
#[test]
fn test_install_command_contents() {
let cmd = Pacman.install(&["vim"]);
let cmd = cmd.as_std();
assert_eq!(cmd.get_program().to_string_lossy(), "pacman");
let args: Vec<String> = cmd
.get_args()
.map(|a| a.to_string_lossy().into_owned())
.collect();
assert_eq!(args, vec!["-S", "--noconfirm", "--needed", "vim"]);
}
}