diff --git a/orion-server/Cargo.toml b/orion-server/Cargo.toml index 8f6913afe..8639d83ad 100644 --- a/orion-server/Cargo.toml +++ b/orion-server/Cargo.toml @@ -11,6 +11,7 @@ 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" +tokio-stream = "0.1.17" tracing = { workspace = true } tracing-subscriber = { workspace = true } serde = { workspace = true, features = ["derive"] } diff --git a/orion-server/README.md b/orion-server/README.md index 5b87520f0..ae6cf9326 100644 --- a/orion-server/README.md +++ b/orion-server/README.md @@ -34,16 +34,16 @@ Orion Server is a Buck2 build task scheduling service written in Rust. It provid #### 1. WebSocket Endpoint -- **`/ws`** - Establishes a WebSocket connection for build clients (agents). +- **`/ws`** + Establishes a WebSocket connection for build clients (agents). - **Purpose:** Register build clients, receive build tasks, and report build logs/status. - **Protocol:** Custom JSON messages (see `WSMessage` in code). #### 2. Submit Build Task -- **`POST /task`** +- **`POST /task`** Submits a new build task to the server. - - **Request Body:** + - **Request Body:** ```json { "repo": "string", @@ -52,19 +52,19 @@ Orion Server is a Buck2 build task scheduling service written in Rust. It provid "mr": "string" // optional, Merge Request number } ``` - - **Response:** + - **Response:** ```json { "task_id": "string", "client_id": "string" } ``` - - **Errors:** + - **Errors:** - `{ "message": "No clients connected" }` if no build agents are available. ```bash curl -X POST http://localhost:8004/task \ -H "Content-Type: application/json" \ - -d '{ + -d '{ "repo": "buck2-rust-third-party", "target": "root//:rust-third-party", "args": [""], @@ -73,9 +73,9 @@ curl -X POST http://localhost:8004/task \ ``` #### 3. Query Task Status -- **`GET /task-status/{id}`** +- **`GET /task-status/{id}`** Query the status of a build task by its ID. - - **Response:** + - **Response:** ```json { "status": "Building|Interrupted|Failed|Completed|NotFound", @@ -83,25 +83,25 @@ curl -X POST http://localhost:8004/task \ "message": "string" // optional } ``` - - **Status Codes:** - - `200 OK` if found - - `404 Not Found` if task does not exist + - **Status Codes:** + - `200 OK` if found + - `404 Not Found` if task does not exist - `400 Bad Request` if ID is invalid #### 4. Real-time Task Output (Logs) -- **`GET /task-output/{id}`** +- **`GET /task-output/{id}`** Streams real-time build logs for a task using Server-Sent Events (SSE). - - **Response:** + - **Response:** - SSE stream of log lines as they are produced. - - If the log file does not exist: + - If the log file does not exist: `data: Task output file not found` #### 5. Query Builds by Merge Request -- **`GET /mr-task/{mr}`** +- **`GET /mr-task/{mr}`** Query historical build records by Merge Request (MR) number. - - **Response:** + - **Response:** - `200 OK` with a JSON array of build records if found. - `404 Not Found` with `{ "message": "No builds found for the given MR" }` if none. - `500 Internal Server Error` on database errors. @@ -111,7 +111,69 @@ curl -X POST http://localhost:8004/task \ curl -X GET http://localhost:8004/mr-task/123 ``` -**Note:** +**Note:** - All endpoints except `/ws` are intended for HTTP clients (frontends, automation, etc.). - WebSocket clients must implement the protocol defined in `orion::ws::WSMessage` for task handling and reporting. - SSE endpoints require clients to support Server-Sent Events. +#### 6. Query Historical Task Logs + +- **`GET /task-history-output/{id}`** + Provides the ability to read historical task logs, supporting either retrieving the **entire log at once** or **retrieving by line segments**. + + - **Path Parameters:** + - `id` *(string)*: Task ID whose log to read. + + - **Query Parameters:** + - `type` *(string, required)*: Type of log retrieval. + - `"full"` → Return the entire log file. + - `"segment"` → Return a portion of the log by line number and limit. + - `offset` *(integer, optional)*: Starting line number (**1-based**). Defaults to `1`. + - `limit` *(integer, optional)*: Maximum number of lines to return. Defaults to `4096`. + + - **Responses:** + - `200 OK` → Returns the log content in JSON: + ```json + { "data": "log content..." } + ``` + - `400 Bad Request` → Invalid parameters: + ```json + { "message": "Invalid type" } + ``` + - `404 Not Found` → Log file does not exist: + ```json + { "message": "Error: Log File Not Found" } + ``` + + + - **Examples:** + + Retrieve the full log: + ```bash + curl -X GET "http://localhost:8004/task-history-output/abc123?type=full" + ``` + + Retrieve log lines 100–150: + ```bash + curl -X GET "http://localhost:8004/task-history-output/abc123?type=segment&offset=100&limit=50" + ``` +#### 7. Real-time Task Output via SSE + +- **`GET /task-output/{id}`** + Streams the build output logs for a specific task in real time using Server-Sent Events (SSE). + - **Path Parameter:** + - `id` — Task ID for which to stream the logs. + - **Response:** + - `200 OK` — A continuous SSE stream of log lines as they are produced. Each log line is sent as an SSE `data` event. + - `404 Not Found` — If the log file for the given task ID does not exist. + - **Behavior:** + - Starts streaming from the end of the log file. + - Keeps the connection alive, sending heartbeat comments every 15 seconds to prevent client timeouts. + - Continues streaming until the build completes and no new logs are appended. + - **Example using `curl`:** + ```bash + curl -N http://localhost:8004/task-output/ + ``` + - **Notes:** + - Replace `` with the actual task ID returned from the `/task` endpoint. + - The SSE stream sends both new log lines (`data`) and periodic heartbeat comments (`: heartbeat`). + - Frontend clients should handle incremental updates as log lines arrive. diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 4fe2ec5dd..727f86eda 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -3,7 +3,6 @@ use crate::scheduler::{ BuildInfo, BuildRequest, TaskQueueStats, TaskScheduler, WorkerInfo, WorkerStatus, create_log_file, get_build_log_dir, }; -use crate::scheduler::{LogReadError, LogSegment}; use axum::{ Json, Router, extract::{ @@ -16,7 +15,7 @@ use axum::{ }; use chrono::{DateTime, Utc}; use dashmap::DashMap; -use futures_util::{SinkExt, Stream, StreamExt, stream}; +use futures_util::{SinkExt, Stream, StreamExt}; use orion::ws::WSMessage; use rand::Rng; use sea_orm::{ @@ -30,10 +29,11 @@ use std::net::SocketAddr; use std::ops::ControlFlow; use std::sync::Arc; use std::time::Duration; -use tokio::io::AsyncReadExt; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt}; use tokio::sync::Mutex; use tokio::sync::mpsc; use tokio::sync::mpsc::UnboundedSender; +use tokio_stream::wrappers::UnboundedReceiverStream; use utoipa::ToSchema; use uuid::Uuid; @@ -50,6 +50,11 @@ pub enum TaskStatusEnum { NotFound, } +/// Default log limit for segmented log retrieval +const DEFAULT_LOG_LIMIT: usize = 4096; +/// Default log offset for segmented log retrieval +const DEFAULT_LOG_OFFSET: u64 = 1; + /// Response structure for task status queries #[derive(Debug, Serialize, Default, ToSchema)] pub struct TaskStatus { @@ -89,8 +94,8 @@ pub fn routers() -> Router { .route("/task-status/{id}", get(task_status_handler)) .route("/task-output/{id}", get(task_output_handler)) .route( - "/task-output-segment/{id}", - get(task_output_segment_handler), + "/task-history-output/{id}", + get(task_history_output_handler), ) .route("/mr-task/{mr}", get(task_query_by_mr)) .route("/tasks/{mr}", get(tasks_handler)) @@ -217,89 +222,148 @@ pub async fn task_output_handler( } let file = tokio::fs::File::open(log_path).await.unwrap(); - let reader = tokio::io::BufReader::new(file); - - // Create a stream that continuously reads from the log file - let stream = stream::unfold(reader, move |mut reader| { - let id_c = id.clone(); - let active_builds = state.scheduler.active_builds.clone(); - async move { + let mut reader = tokio::io::BufReader::new(file); + reader.seek(tokio::io::SeekFrom::End(0)).await.unwrap(); + + // Build an asynchronous data channel + // Spawn background task: handle both log + heartbeat with select + let (tx, rx) = mpsc::unbounded_channel::(); + tokio::spawn(async move { + let mut heartbeat = tokio::time::interval(Duration::from_secs(15)); + loop { let mut buf = String::new(); - let is_building = active_builds.contains_key(&id_c); - let size = reader.read_to_string(&mut buf).await.unwrap(); - - if size == 0 { - if is_building { - // Wait for more content if build is still active - tokio::time::sleep(Duration::from_secs(1)).await; - let size = reader.read_to_string(&mut buf).await.unwrap(); - if size > 0 { - Some((Ok(Event::default().data(buf)), reader)) + let active_builds = state.scheduler.active_builds.clone(); + let is_building = active_builds.contains_key(&id); + + tokio::select! { + size = reader.read_to_string(&mut buf) => { + let size = size.unwrap(); + if size == 0 { + if is_building { + continue; + } else { + break; + } } else { - // Send keep-alive comment - Some((Ok(Event::default().comment("")), reader)) + let _ = tx.send(Event::default().data(buf.trim_end())); } - } else { - // Build finished and no more content - None } - } else { - // Send new content - Some((Ok(Event::default().data(buf)), reader)) + _ = heartbeat.tick() => { + let _ = tx.send(Event::default().comment("heartbeat")); + } } } }); - Ok(Sse::new(stream.boxed()).keep_alive(KeepAlive::new())) + let stream = UnboundedReceiverStream::new(rx).map(Ok::<_, Infallible>); + + Ok(Sse::new(stream).keep_alive(KeepAlive::new())) } +/// Provides the ability to read historical task logs +/// supporting either retrieving the entire log at once or segmenting it by line count. #[utoipa::path( get, - path = "/task-output-segment/{id}", + path = "/task-history-output/{id}", params( ("id" = String, Path, description = "Task ID whose log to read"), - ("offset" = u64, Query, description = "Start byte offset", example = 0), - ("len" = usize, Query, description = "Max bytes to read", example = 4096) + ("type" = String, Query, description = "The type of log retrieval: \"full\" indicates full retrieval, while \"segment\" indicates retrieval of segments.",example = "full"), + ("offset" = Option, Query, description = "Start line number (1-based)"), + ("limit" = Option, Query, description = "Max number of lines to return"), ), responses( - (status = 200, description = "Log segment", body = LogSegment), + (status = 200, description = "History Log"), + (status = 400, description = "Invalid parameters"), (status = 404, description = "Log file not found"), - (status = 416, description = "Offset out of range"), - (status = 400, description = "Invalid parameters") + (status = 500, description = "Failed to operate log file"), ) )] -pub async fn task_output_segment_handler( - State(state): State, +pub async fn task_history_output_handler( Path(id): Path, axum::extract::Query(params): axum::extract::Query>, ) -> impl IntoResponse { - let offset = params - .get("offset") - .and_then(|v| v.parse::().ok()) - .unwrap_or(0); - let len = params - .get("len") - .and_then(|v| v.parse::().ok()) - .unwrap_or(4096); - // Cap len to reasonable maximum (e.g., 1MB) - let len = len.min(1024 * 1024); - - match state.scheduler.read_log_segment(&id, offset, len).await { - Ok(seg) => (StatusCode::OK, Json(seg)).into_response(), - Err(LogReadError::NotFound) => StatusCode::NOT_FOUND.into_response(), - Err(LogReadError::OffsetOutOfRange { size }) => ( - StatusCode::RANGE_NOT_SATISFIABLE, + let log_path_str = format!("{}/{}", get_build_log_dir(), id); + let log_path = std::path::Path::new(&log_path_str); + + // Return error message if log file doesn't exist + if !log_path.exists() { + return ( + StatusCode::NOT_FOUND, Json(serde_json::json!({ - "message": "Offset out of range", - "file_size": size + "message": "Error: Log File Not Found" })), - ) - .into_response(), - Err(LogReadError::Io(e)) => ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": format!("IO error: {e}")})), - ) - .into_response(), + ); + } + + let log_type = params.get("type").map(|s| s.as_str()); + match log_type { + Some("full") => { + // Read the entire log file + let file = match tokio::fs::File::open(log_path).await { + Ok(f) => f, + Err(_) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "message": "Failed to open log file" })), + ); + } + }; + let mut reader = tokio::io::BufReader::new(file); + let mut buf = String::new(); + if reader.read_to_string(&mut buf).await.is_err() { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "message": "Failed to read log file" })), + ); + } + + (StatusCode::OK, Json(serde_json::json!({ "data": buf }))) + } + Some("segment") => { + // Parse offset + let offset = params + .get("offset") + .and_then(|v| v.parse::().ok()) + .unwrap_or(DEFAULT_LOG_OFFSET) + .saturating_sub(1); + let limit = params + .get("limit") + .and_then(|v| v.parse::().ok()) + .unwrap_or(DEFAULT_LOG_LIMIT); + + // Read Range Log + let file = match tokio::fs::File::open(log_path).await { + Ok(f) => f, + Err(_) => { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + Json(serde_json::json!({ "message": "Failed to open log file" })), + ); + } + }; + let reader = tokio::io::BufReader::new(file); + let mut buf = String::new(); + let mut lines = reader.lines(); + let mut idx = 0; + let mut count = 0; + + while let Ok(Some(line)) = lines.next_line().await { + if idx >= offset { + buf.push_str(&line); + buf.push('\n'); + count += 1; + if count >= limit { + break; + } + } + idx += 1; + } + (StatusCode::OK, Json(serde_json::json!({ "data": buf }))) + } + _ => ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "message": "Invalid type" })), + ), } } diff --git a/orion-server/src/scheduler.rs b/orion-server/src/scheduler.rs index 9959487c9..f500771a6 100644 --- a/orion-server/src/scheduler.rs +++ b/orion-server/src/scheduler.rs @@ -186,6 +186,7 @@ pub struct LogSegment { } /// Errors when reading a log segment +#[allow(dead_code)] #[derive(Debug)] pub enum LogReadError { NotFound, @@ -218,7 +219,8 @@ impl TaskScheduler { } /// Read a segment of a log file by task id at given offset, limited to max_len bytes. - pub async fn read_log_segment( + #[allow(dead_code)] + async fn read_log_segment( &self, task_id: &str, offset: u64, diff --git a/orion-server/src/server.rs b/orion-server/src/server.rs index c04a7c9a3..ae01e5608 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -23,7 +23,7 @@ use crate::model::builds; api::task_handler, api::task_status_handler, api::task_output_handler, - api::task_output_segment_handler, + api::task_history_output_handler, api::task_query_by_mr, api::tasks_handler, ),