diff --git a/orion-server/.env b/orion-server/.env index e2ccd1d86..0f281d582 100644 --- a/orion-server/.env +++ b/orion-server/.env @@ -1,3 +1,4 @@ BUILD_LOG_DIR="/tmp/megadir/buck2ctl" DATABASE_URL="postgres://postgres:postgres@localhost/orion" -PORT=80 \ No newline at end of file +PORT=80 +MONOBASE_URL="http://git.gitmega.com" \ No newline at end of file diff --git a/orion-server/Cargo.toml b/orion-server/Cargo.toml index e6f312a43..142495dd3 100644 --- a/orion-server/Cargo.toml +++ b/orion-server/Cargo.toml @@ -9,6 +9,7 @@ orion = { workspace = true} axum = { workspace = true, features = ["macros", "ws"] } axum-extra = { workspace = true, features = ["erased-json"] } tokio = { workspace = true, features = ["rt-multi-thread", "fs", "process"] } +tokio-retry = "0.3" tracing = { workspace = true } tracing-subscriber = { workspace = true } serde = { workspace = true, features = ["derive"] } @@ -29,4 +30,6 @@ dashmap = { workspace = true } utoipa-axum.workspace = true utoipa.workspace = true utoipa-swagger-ui = { workspace = true, features = ["axum"] } -chrono = { version = "0.4", features = ["serde"] } \ No newline at end of file +chrono = { version = "0.4", features = ["serde"] } +reqwest.workspace = true + diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 855d18bcc..35ebaec8a 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -1,3 +1,4 @@ +use crate::buck2::download_and_get_buck2_targets; use crate::model::builds; use axum::{ Json, Router, @@ -61,7 +62,8 @@ fn create_log_file(task_id: &str) -> Result { #[derive(Debug, Deserialize, ToSchema)] pub struct BuildRequest { repo: String, - target: String, + buck_hash: String, + buckconfig_hash: String, args: Option>, mr: Option, } @@ -281,7 +283,19 @@ pub async fn task_output_handler( pub async fn task_handler( State(state): State, Json(req): Json, -) -> (StatusCode, Json) { +) -> impl IntoResponse { + // Download and get buck2 targets first + let target = match download_and_get_buck2_targets(&req.buck_hash, &req.buckconfig_hash).await { + Ok(target) => target, + Err(e) => { + tracing::error!("Failed to download buck2 targets: {}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "message": format!("Failed to download buck2 targets: {}", e) })), + ).into_response(); + } + }; + // Find all idle workers let idle_workers: Vec = state .workers @@ -295,7 +309,7 @@ pub async fn task_handler( return ( StatusCode::SERVICE_UNAVAILABLE, Json(serde_json::json!({"message": "No available workers at the moment"})), - ); + ).into_response(); } // Randomly select an idle worker @@ -314,14 +328,14 @@ pub async fn task_handler( return ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"message": "Failed to create log file"})), - ); + ).into_response(); } }; // Create build information structure let build_info = BuildInfo { repo: req.repo.clone(), - target: req.target.clone(), + target: target.clone(), args: req.args.clone(), start_at: chrono::Utc::now(), mr: req.mr.clone(), @@ -346,14 +360,14 @@ pub async fn task_handler( return ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"message": "Failed to create task in database"})), - ); + ).into_response(); } // Create WebSocket message for the worker let msg = WSMessage::Task { id: task_id.to_string(), repo: req.repo, - target: req.target, + target, args: req.args, mr: req.mr.unwrap_or_default(), }; @@ -367,7 +381,7 @@ pub async fn task_handler( ( StatusCode::OK, Json(serde_json::json!({"task_id": task_id.to_string(), "client_id": chosen_id})), - ) + ).into_response() } else { tracing::error!( "Failed to send task to supposedly idle worker {}. It might have just disconnected.", @@ -378,7 +392,7 @@ pub async fn task_handler( Json( serde_json::json!({"message": "Failed to dispatch task to worker. Please try again."}), ), - ) + ).into_response() } } else { tracing::error!( @@ -388,7 +402,7 @@ pub async fn task_handler( ( StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({"message": "Internal scheduler error."})), - ) + ).into_response() } } diff --git a/orion-server/src/buck2.rs b/orion-server/src/buck2.rs new file mode 100644 index 000000000..e3f3a6dd8 --- /dev/null +++ b/orion-server/src/buck2.rs @@ -0,0 +1,174 @@ +use std::process::Command; +use std::fs; +use std::path::Path; +use tokio_retry::{strategy::ExponentialBackoff, Retry}; +use uuid::Uuid; +use std::time::Duration; + + + +/// Download files from file blob API using two hash values and save them to a new folder in tmp directory +async fn download_files_to_tmp(hash1: &str, hash2: &str) -> Result> { + // Create tmp directory path + // Generate a unique folder name using UUID + let folder_name = Uuid::now_v7().to_string(); + println!("Generated folder name: {folder_name},buck:{hash1},.buckconfig:{hash2}"); + + // Create tmp directory path + let tmp_dir = std::env::temp_dir().join(folder_name); + + fs::create_dir_all(&tmp_dir)?; + + // Download first file as BUCK + let buck_path = tmp_dir.join("BUCK"); + download_file_with_retry(hash1, &buck_path, 3).await?; + + // Download second file as .buckconfig + let buckconfig_path = tmp_dir.join(".buckconfig"); + download_file_with_retry(hash2, &buckconfig_path, 3).await?; + + Ok(tmp_dir.to_string_lossy().to_string()) +} + +/// Download a single file with retry mechanism using tokio-retry +async fn download_file_with_retry( + hash: &str, + file_path: &Path, + max_retries: usize +) -> Result<(), Box> { + let api_endpoint = file_blob_endpoint(); + let url = format!("{api_endpoint}/{hash}"); + + // Create retry strategy: exponential backoff starting from 100ms, max 3 attempts + let retry_strategy = ExponentialBackoff::from_millis(100) + .max_delay(Duration::from_secs(2)) + .take(max_retries); + + Retry::spawn(retry_strategy, || { + download_single_file(&url, file_path) + }) + .await + .map_err(|e| format!("Failed to download {hash} after {max_retries} attempts: {e}").into()) +} + +/// Download a single file from URL and save to specified path +async fn download_single_file(url: &str, file_path: &Path) -> Result<(), Box> { + let response = reqwest::get(url).await?; + + if !response.status().is_success() { + return Err(format!("HTTP error: {}", response.status()).into()); + } + + let content = response.bytes().await?; + fs::write(file_path, &content)?; + + Ok(()) +} + +/// Get the base URL for API requests +fn base_url() -> String { + std::env::var("MONOBASE_URL").unwrap_or_else(|_| "http://localhost:8000".to_string()) +} + +/// Get the file blob API endpoint +pub fn file_blob_endpoint() -> String { + format!("{}/api/v1/file/blob", base_url()) +} + +/// Execute buck2 targets //... command in the given directory and return the last line string +pub fn get_buck2_targets_last_line(directory: &str) -> Result> { + let output = Command::new("buck2") + .args(["targets", "//..."]) + .current_dir(directory) + .output()?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(format!("buck2 command failed: {stderr}" ).into()); + } + + let stdout = String::from_utf8_lossy(&output.stdout); + let last_line = stdout + .lines() + .last() + .unwrap_or("") + .to_string(); + + Ok(last_line) +} + +/// Download files and execute buck2 targets command to get the last line output +pub async fn download_and_get_buck2_targets( + hash1: &str, + hash2: &str, +) -> Result> { + // First, download the files (BUCK and .buckconfig) to tmp directory + let tmp_dir_path = download_files_to_tmp(hash1, hash2).await?; + + // Then, execute buck2 targets command in the downloaded directory + let last_line = get_buck2_targets_last_line(&tmp_dir_path)?; + + // Clean up the temporary directory after getting the result + //TODO: Cleanup should happen in a finally block or using RAII to ensure temporary files are removed even if buck2 command fails. Consider using a guard or defer pattern. TempDirGuard , for example. + if Path::new(&tmp_dir_path).exists() { + fs::remove_dir_all(&tmp_dir_path)?; + } + + Ok(last_line) +} + +#[cfg(test)] +mod tests { + use super::*; + use std::fs; + use std::path::Path; + + #[test] + fn test_get_buck2_targets_last_line() { + // Use the test directory within the project + let test_dir = "./test"; + if Path::new(test_dir).exists() { + // Create a temporary directory for testing + let tmp_test_dir = std::env::temp_dir().join("buck2_test"); + + // Remove existing tmp directory if it exists + if tmp_test_dir.exists() { + fs::remove_dir_all(&tmp_test_dir).expect("Failed to remove existing tmp directory"); + } + + // Copy test directory contents to tmp directory + copy_dir_all(test_dir, &tmp_test_dir).expect("Failed to copy test directory to tmp"); + + // Run buck2 targets command in the tmp directory + match get_buck2_targets_last_line(tmp_test_dir.to_str().unwrap()) { + Ok(last_line) => { + println!("Last line: {last_line}"); + assert!(!last_line.is_empty()); + }, + Err(e) => println!("Warning: {e}"), + } + + // Clean up: remove the tmp directory after test + if tmp_test_dir.exists() { + fs::remove_dir_all(&tmp_test_dir).expect("Failed to clean up tmp directory"); + } + } else { + println!("Test directory '{test_dir}' does not exist. Skipping test."); + } + } + + /// Helper function to recursively copy directory contents + fn copy_dir_all(src: impl AsRef, dst: impl AsRef) -> std::io::Result<()> { + fs::create_dir_all(&dst)?; + for entry in fs::read_dir(src)? { + let entry = entry?; + let ty = entry.file_type()?; + if ty.is_dir() { + copy_dir_all(entry.path(), dst.as_ref().join(entry.file_name()))?; + } else { + fs::copy(entry.path(), dst.as_ref().join(entry.file_name()))?; + } + } + Ok(()) + } +} \ No newline at end of file diff --git a/orion-server/src/main.rs b/orion-server/src/main.rs index b2e429fa3..2db33d7cb 100644 --- a/orion-server/src/main.rs +++ b/orion-server/src/main.rs @@ -1,6 +1,7 @@ mod api; mod model; mod server; +mod buck2; /// Orion Build Server /// A distributed build system that manages build tasks and worker nodes