From 1474eb2544cb709a582f21637959e4b40962bcaf Mon Sep 17 00:00:00 2001 From: MYUU <1405758738@qq.com> Date: Mon, 1 Sep 2025 08:34:50 +0800 Subject: [PATCH 1/3] feat(orion-server): historical logs API & task_output refactor(#1379) Signed-off-by: MYUU <1405758738@qq.com> --- orion-server/Cargo.toml | 1 + orion-server/README.md | 98 +++++++++++++++++++----- orion-server/src/api.rs | 148 ++++++++++++++++++++++++++++++------- orion-server/src/server.rs | 1 + 4 files changed, 202 insertions(+), 46 deletions(-) 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..57886b9fe 100644 --- a/orion-server/src/api.rs +++ b/orion-server/src/api.rs @@ -16,7 +16,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 +30,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; @@ -88,6 +89,10 @@ pub fn routers() -> Router { .route("/task", axum::routing::post(task_handler)) .route("/task-status/{id}", get(task_status_handler)) .route("/task-output/{id}", get(task_output_handler)) + .route( + "/task-history-output/{id}", + get(task_history_output_handler), + ) .route( "/task-output-segment/{id}", get(task_output_segment_handler), @@ -217,40 +222,127 @@ 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-history-output/{id}", + params( + ("id" = String, Path, description = "Task ID whose log to read"), + ("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 = "History Log"), + (status = 400, description = "Invalid parameters"), + (status = 404, description = "Log file not found"), + ) +)] +pub async fn task_history_output_handler( + Path(id): Path, + axum::extract::Query(params): axum::extract::Query>, +) -> impl IntoResponse { + 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": "Error: Log File Not Found" + })), + ); + } + + let log_type = params.get("type").map(|s| s.as_str()); + match log_type { + Some("full") => { + // Read the entire log file + let file = tokio::fs::File::open(log_path).await.unwrap(); + let mut reader = tokio::io::BufReader::new(file); + let mut buf = String::new(); + reader.read_to_string(&mut buf).await.unwrap(); + + (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(1) + .saturating_sub(1); + let limit = params + .get("limit") + .and_then(|v| v.parse::().ok()) + .unwrap_or(4096); + + // Read Range Log + let file = tokio::fs::File::open(log_path).await.unwrap(); + 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" })), + ), + } } #[utoipa::path( diff --git a/orion-server/src/server.rs b/orion-server/src/server.rs index c04a7c9a3..ed029121f 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -23,6 +23,7 @@ use crate::model::builds; api::task_handler, api::task_status_handler, api::task_output_handler, + api::task_history_output_handler, api::task_output_segment_handler, api::task_query_by_mr, api::tasks_handler, From f8515eb2a4c222d237063e572f79311d67333dd8 Mon Sep 17 00:00:00 2001 From: Li WenLin <2839681263@qq.com> Date: Mon, 1 Sep 2025 09:52:03 +0800 Subject: [PATCH 2/3] feat(UI): add gpg-key setting (#1397) --- moon/apps/web/components/Setting/GPGKeys.tsx | 217 +++++++++++++++++++ moon/apps/web/hooks/useDeleteGPGKeyById.ts | 9 + moon/apps/web/hooks/useGetGPGList.ts | 32 +++ moon/apps/web/hooks/usePostGPGKey.ts | 9 + moon/apps/web/pages/me/settings/index.tsx | 2 + 5 files changed, 269 insertions(+) create mode 100644 moon/apps/web/components/Setting/GPGKeys.tsx create mode 100644 moon/apps/web/hooks/useDeleteGPGKeyById.ts create mode 100644 moon/apps/web/hooks/useGetGPGList.ts create mode 100644 moon/apps/web/hooks/usePostGPGKey.ts diff --git a/moon/apps/web/components/Setting/GPGKeys.tsx b/moon/apps/web/components/Setting/GPGKeys.tsx new file mode 100644 index 000000000..c5e089036 --- /dev/null +++ b/moon/apps/web/components/Setting/GPGKeys.tsx @@ -0,0 +1,217 @@ +import React, { useState } from 'react'; +import { LoadingSpinner, LockIcon, Button, TextField, PlusIcon } from '@gitmono/ui' +import * as Dialog from '@gitmono/ui/src/Dialog' +import { DateAndTimePicker } from "@/components/DateAndTimePicker"; +import { useGetGPGList } from '@/hooks/useGetGPGList' +import { usePostGPGKey } from '@/hooks/usePostGPGKey' +import { useDeleteGPGKeyById } from '@/hooks/useDeleteGPGKeyById' +import { legacyApiClient } from "@/utils/queryClient"; +import { useQueryClient } from "@tanstack/react-query"; +import HandleTime from "@/components/MrView/components/HandleTime"; +import { GpgKey } from "@gitmono/types"; + +const GpgKeyItem = ({ keyData } : { keyData: GpgKey }) => { + const { mutate: deleteGPGKey } = useDeleteGPGKeyById() + const fetchGPGList = legacyApiClient.v1.getApiGpgList() + const queryClient = useQueryClient() + + return ( +
+
+
+ +
+ ) +} + +interface NewGPGKeyDialogProps { + open: boolean; + setOpen: (open: boolean) => void; +} + +const NewGPGKeyDialog = ({ open, setOpen }: NewGPGKeyDialogProps) => { + const { mutate: postGPGKey, isPending } = usePostGPGKey() + // const [title, setTitle] = useState('') + const [gpg_content, setGpg_content] = useState('') + const [errors, setErrors] = useState<{ title?: string; gpgKey?: string }>({}) + const [expires_days, setExpiresDays] = useState(new Date()) + + const fetchGPGList = legacyApiClient.v1.getApiGpgList() + const queryClient = useQueryClient() + + const handleSubmit = (e?: React.FormEvent | React.MouseEvent) => { + if (e) e.preventDefault() + const nextErrors: { title?: string; gpgKey?: string } = {} + + // if (!title.trim()) nextErrors.title = 'Title is required' + if (!gpg_content.trim()) nextErrors.gpgKey = 'GPG key is required' + setErrors(nextErrors) + if (Object.keys(nextErrors).length > 0) return + + // Normalize Windows-style line endings to Unix-style + const normalizedGpgContent = gpg_content.replace(/\r\n/g, '\n') + + postGPGKey( + { + data: { + expires_days: 10, + gpg_content: normalizedGpgContent + } + }, + { + onSuccess: () => { + setOpen(false) + // setTitle('') + setGpg_content('') + setErrors({}) + + queryClient.invalidateQueries({ queryKey: fetchGPGList.requestKey() }) + } + } + ) + } + + return ( + + + Add GPG key + + + {/*
*/ } + {/* */ } + {/* {errors.title && {errors.title}}*/ } + {/*
*/ } + +
+ +
+ +
+
+ +
+ + { errors.gpgKey && { errors.gpgKey } } +
+ + + + + + + + + + ) +} + +const GPGKeys = () => { + const { gpgKeys, isLoading: isGPGLoading } = useGetGPGList() + const [open, setOpen] = useState(false) + + return ( + <> +
+
+

GPG keys

+ +
+ +

+ This is a list of GPG keys associated with your account. Remove any keys that you do not recognize. +

+ +
+

+ Authentication keys +

+ { isGPGLoading? ( +
+ +
+ ) : ( +
+ { gpgKeys?.map((key) => ( + + )) } +
+ ) } +
+
+ + + ); +}; + +export default GPGKeys; \ No newline at end of file diff --git a/moon/apps/web/hooks/useDeleteGPGKeyById.ts b/moon/apps/web/hooks/useDeleteGPGKeyById.ts new file mode 100644 index 000000000..354dd7d35 --- /dev/null +++ b/moon/apps/web/hooks/useDeleteGPGKeyById.ts @@ -0,0 +1,9 @@ +import { useMutation } from "@tanstack/react-query"; +import { DeleteApiGpgRemoveData, RemoveGpgRequest } from "@gitmono/types"; +import { legacyApiClient } from "@/utils/queryClient"; + +export const useDeleteGPGKeyById = () => { + return useMutation({ + mutationFn: ({ data }) => legacyApiClient.v1.deleteApiGpgRemove().request(data) + }) +}; \ No newline at end of file diff --git a/moon/apps/web/hooks/useGetGPGList.ts b/moon/apps/web/hooks/useGetGPGList.ts new file mode 100644 index 000000000..66af23fa3 --- /dev/null +++ b/moon/apps/web/hooks/useGetGPGList.ts @@ -0,0 +1,32 @@ +import { legacyApiClient } from "@/utils/queryClient"; +import { atomFamily } from "jotai/utils"; +import { atomWithWebStorage } from "@/utils/atomWithWebStorage"; +import { GpgKey } from "@gitmono/types"; +import { useAtom } from "jotai"; +import { useQuery } from "@tanstack/react-query"; +import { useEffect } from "react"; + +const fetchGPGList = legacyApiClient.v1.getApiGpgList() +const getGPGListAtom = atomFamily(() => + atomWithWebStorage(`gpg-key`, []) +) + +export const useGetGPGList = () => { + const [gpgKeys, setGpgKeys] = useAtom(getGPGListAtom('gpg-key')) + const { data, isLoading } = useQuery({ + queryKey: fetchGPGList.requestKey(), + queryFn: async () => { + const response = await fetchGPGList.request() + + return response.data + } + }); + + useEffect(() => { + if(data) { + setGpgKeys(data) + } + }, [data, setGpgKeys]); + + return { gpgKeys, isLoading } +}; \ No newline at end of file diff --git a/moon/apps/web/hooks/usePostGPGKey.ts b/moon/apps/web/hooks/usePostGPGKey.ts new file mode 100644 index 000000000..a7066b84a --- /dev/null +++ b/moon/apps/web/hooks/usePostGPGKey.ts @@ -0,0 +1,9 @@ +import { useMutation } from "@tanstack/react-query"; +import { NewGpgRequest, PostApiGpgAddData } from "@gitmono/types"; +import { legacyApiClient } from "@/utils/queryClient"; + +export const usePostGPGKey = () => { + return useMutation({ + mutationFn: ({ data }) => legacyApiClient.v1.postApiGpgAdd().request(data) + }) +} \ No newline at end of file diff --git a/moon/apps/web/pages/me/settings/index.tsx b/moon/apps/web/pages/me/settings/index.tsx index 191e60f0d..933f6d216 100644 --- a/moon/apps/web/pages/me/settings/index.tsx +++ b/moon/apps/web/pages/me/settings/index.tsx @@ -17,6 +17,7 @@ import { PersonalCallLinks } from '@/components/UserSettings/PersonalCallLinks'; import { SlackNotificationSettings } from '@/components/UserSettings/SlackNotificationSettings'; import { Timezone } from '@/components/UserSettings/Timezone' import { PageWithProviders } from '@/utils/types'; +import GPGKeys from "@/components/Setting/GPGKeys"; const UserSettingsPage: PageWithProviders = () => { useEffect(() => { @@ -43,6 +44,7 @@ const UserSettingsPage: PageWithProviders = () => { + From d1991496307f7e73124d7b39952b9b87ae31b1bf Mon Sep 17 00:00:00 2001 From: MYUU <1405758738@qq.com> Date: Mon, 1 Sep 2025 10:54:39 +0800 Subject: [PATCH 3/3] feat(orion-server): add error handling for file operations, remove task-output-segment Signed-off-by: MYUU <1405758738@qq.com> --- orion-server/src/api.rs | 96 +++++++++++++---------------------- orion-server/src/scheduler.rs | 4 +- orion-server/src/server.rs | 1 - 3 files changed, 37 insertions(+), 64 deletions(-) diff --git a/orion-server/src/api.rs b/orion-server/src/api.rs index 57886b9fe..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::{ @@ -51,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 { @@ -93,10 +97,6 @@ pub fn routers() -> Router { "/task-history-output/{id}", get(task_history_output_handler), ) - .route( - "/task-output-segment/{id}", - get(task_output_segment_handler), - ) .route("/mr-task/{mr}", get(task_query_by_mr)) .route("/tasks/{mr}", get(tasks_handler)) .route("/queue-stats", get(queue_stats_handler)) @@ -267,14 +267,15 @@ pub async fn task_output_handler( path = "/task-history-output/{id}", params( ("id" = String, Path, description = "Task ID whose log to read"), - ("type" = String, Query, description = "The type of log retrieval: “full” indicates full retrieval, while “segment” indicates retrieval of segments.",example = "full"), + ("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"), + ("limit" = Option, Query, description = "Max number of lines to return"), ), responses( (status = 200, description = "History Log"), (status = 400, description = "Invalid parameters"), (status = 404, description = "Log file not found"), + (status = 500, description = "Failed to operate log file"), ) )] pub async fn task_history_output_handler( @@ -298,10 +299,23 @@ pub async fn task_history_output_handler( match log_type { Some("full") => { // Read the entire log file - let file = tokio::fs::File::open(log_path).await.unwrap(); + 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(); - reader.read_to_string(&mut buf).await.unwrap(); + 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 }))) } @@ -310,15 +324,23 @@ pub async fn task_history_output_handler( let offset = params .get("offset") .and_then(|v| v.parse::().ok()) - .unwrap_or(1) + .unwrap_or(DEFAULT_LOG_OFFSET) .saturating_sub(1); let limit = params .get("limit") .and_then(|v| v.parse::().ok()) - .unwrap_or(4096); + .unwrap_or(DEFAULT_LOG_LIMIT); // Read Range Log - let file = tokio::fs::File::open(log_path).await.unwrap(); + 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(); @@ -345,56 +367,6 @@ pub async fn task_history_output_handler( } } -#[utoipa::path( - get, - path = "/task-output-segment/{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) - ), - responses( - (status = 200, description = "Log segment", body = LogSegment), - (status = 404, description = "Log file not found"), - (status = 416, description = "Offset out of range"), - (status = 400, description = "Invalid parameters") - ) -)] -pub async fn task_output_segment_handler( - State(state): State, - 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, - Json(serde_json::json!({ - "message": "Offset out of range", - "file_size": size - })), - ) - .into_response(), - Err(LogReadError::Io(e)) => ( - StatusCode::INTERNAL_SERVER_ERROR, - Json(serde_json::json!({"message": format!("IO error: {e}")})), - ) - .into_response(), - } -} - #[utoipa::path( post, path = "/task", 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 ed029121f..ae01e5608 100644 --- a/orion-server/src/server.rs +++ b/orion-server/src/server.rs @@ -24,7 +24,6 @@ use crate::model::builds; api::task_status_handler, api::task_output_handler, api::task_history_output_handler, - api::task_output_segment_handler, api::task_query_by_mr, api::tasks_handler, ),