From 059f4b12ee6c3f0e33fc2f3303d60e31eabd3c42 Mon Sep 17 00:00:00 2001 From: Xiaoyang Han Date: Wed, 23 Jul 2025 14:43:33 +0800 Subject: [PATCH] Orion integrated with Scorpio. Added support for MR. Signed-off-by: Xiaoyang Han --- orion-server/src/api.rs | 1 + orion/Cargo.toml | 3 ++- orion/src/api.rs | 2 ++ orion/src/buck_controller.rs | 43 ++++++++++++++++++++++++++++++++++++ orion/src/ws.rs | 34 ++++++++++++++++++++++++---- 5 files changed, 78 insertions(+), 5 deletions(-) diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index e3476842f..a8866db61 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -247,6 +247,7 @@ async fn task_handler( repo: req.repo, target: req.target, args: req.args, + mr: req.mr.clone().unwrap_or_default(), }; state.clients.get(chosen_id).unwrap().send(msg).unwrap(); // TODO client maybe disconnected diff --git a/orion/Cargo.toml b/orion/Cargo.toml index a0cc4ac78..05baaa35b 100644 --- a/orion/Cargo.toml +++ b/orion/Cargo.toml @@ -16,4 +16,5 @@ once_cell = { workspace = true } dotenvy = { workspace = true } tokio-tungstenite = { workspace = true } tungstenite = { workspace = true } -serde_json = { workspace = true } \ No newline at end of file +serde_json = { workspace = true } +reqwest.workspace = true diff --git a/orion/src/api.rs b/orion/src/api.rs index 48495ca2b..77831335e 100644 --- a/orion/src/api.rs +++ b/orion/src/api.rs @@ -11,6 +11,7 @@ pub struct BuildRequest { pub repo: String, pub target: String, pub args: Option>, + pub mr: String, // merge request id // pub webhook: Option, // post } @@ -41,6 +42,7 @@ pub async fn buck_build( req.repo.clone(), req.target.clone(), req.args.unwrap_or_default(), + req.mr.clone(), sender.clone(), ) .await diff --git a/orion/src/buck_controller.rs b/orion/src/buck_controller.rs index 666c2c1a5..abb770473 100644 --- a/orion/src/buck_controller.rs +++ b/orion/src/buck_controller.rs @@ -1,5 +1,6 @@ use crate::ws::WSMessage; use once_cell::sync::Lazy; +use serde_json::json; use std::io; use std::process::{ExitStatus, Stdio}; use tokio::io::AsyncBufReadExt; @@ -9,13 +10,55 @@ use tokio::sync::mpsc::UnboundedSender; static PROJECT_ROOT: Lazy = Lazy::new(|| std::env::var("BUCK_PROJECT_ROOT").expect("BUCK_PROJECT_ROOT must be set")); + +/// Sends a filesystem mount request to the specified API endpoint +/// Parameters: +/// - repo: Repository path to mount +/// - mr: Merge request number +/// Returns: Result containing the response body string on success, or an error on failure +pub async fn mount_fs(repo: &str, mr: &str) -> Result { + // Create HTTP client + let client = reqwest::Client::new(); + + // Construct JSON request payload + let payload = json!({ + "path": repo, + "mr": mr, + }); + + // Send POST request + let res = client + .post("http://localhost:2725/api/fs/mount") + .header("Content-Type", "application/json") + .body(payload.to_string()) + .send() + .await?; + + // Print status code + println!("Mount request status: {}", res.status()); + + // Get and return response body + let body = res.text().await?; + println!("Mount response body: {body}"); + + Ok(body) +} + pub async fn build( id: String, repo: String, target: String, args: Vec, + mr: String, sender: UnboundedSender, ) -> io::Result { + + tracing::info!("Building {} in repo {} with target {}", id, repo, target); + // Prepare the command to run + // Note: `args` is a list of additional arguments to pass to the `buck + + let _ = mount_fs(&repo, &mr).await; + let mut cmd = Command::new("buck2"); let cmd = cmd .arg("build") diff --git a/orion/src/ws.rs b/orion/src/ws.rs index 6fcb30c05..e4ce1181b 100644 --- a/orion/src/ws.rs +++ b/orion/src/ws.rs @@ -18,6 +18,7 @@ pub enum WSMessage { repo: String, target: String, args: Option>, + mr: String, }, TaskAck { id: String, @@ -36,6 +37,30 @@ pub enum WSMessage { }, } +const MAX_RETRIES: u32 = 3; + +/// send message and retry +async fn send_with_retry(sender: &mut (impl SinkExt + Unpin), msg: &WSMessage) -> Result<(), String> { + let msg_str = serde_json::to_string(msg) + .map_err(|e| format!("seriaz fail: {e}"))?; + let ws_msg = Message::Text(Utf8Bytes::from(msg_str)); + + for attempt in 0..=MAX_RETRIES { + match sender.send(ws_msg.clone()).await { + Ok(_) => return Ok(()), + Err(_) => { + if attempt == MAX_RETRIES { + return Err(format!("max retry times:({MAX_RETRIES}), ")); + } + println!("send fail (retry {}/{})", attempt + 1, MAX_RETRIES); + const RETRY_DELAY_MS: u64 = 200; + tokio::time::sleep(std::time::Duration::from_millis(RETRY_DELAY_MS)).await; + } + } + } + Err("max retry error".into()) +} + pub async fn spawn_client(server: &str) { let ws_stream = match connect_async(server).await { Ok((stream, response)) => { @@ -55,8 +80,8 @@ pub async fn spawn_client(server: &str) { let mut send_task = tokio::spawn(async move { while let Some(msg) = rx.recv().await { - let msg = serde_json::to_string(&msg).unwrap(); - if let Err(e) = sender.send(Message::Text(Utf8Bytes::from(msg))).await { + //let msg = serde_json::to_string(&msg).unwrap(); + if let Err(e) = send_with_retry(&mut sender, &msg).await { println!("Error sending message: {e}"); break; } @@ -91,11 +116,12 @@ async fn process_message(msg: Message) -> ControlFlow<(), ()> { repo, target, args, + mr, } => { - println!(">>> got task: id:{id}, repo:{repo}, target:{target}, args:{args:?}"); + println!(">>> got task: id:{id}, repo:{repo}, target:{target}, args:{args:?}, mr:{mr}"); let Json(res) = buck_build( id.parse().unwrap(), - BuildRequest { repo, target, args }, + BuildRequest { repo, target, args, mr }, SENDER.get().unwrap().clone(), ) .await;