From 80fa3bea84204e4de978970ebabdd651a1db97d4 Mon Sep 17 00:00:00 2001 From: Hazem Nureldin Date: Tue, 4 Aug 2026 08:20:13 +0100 Subject: [PATCH] fix:lohs retrival using wss fix:taskname added, log seletion done refactor: the log handling on on the webpage Support archived S3 logs in subscriptions fix: display logs like the archieved ones fix: the link to s3 main.log working fix: preserve task logs after pod completion while workflow continues test: align log retrieval tests with S3-first archive lookup feat(logs): retrieve archived task logs when live stream ends fix: error mssages fix: update task selection tests and React Flow mocks fix: resolve workflow view tests and lint issues fix: resolve workflow nullability and lint errors fix: resolve workflow nullability and lint errors fix: format SingleWorkflowView with Prettier --- .../graph-proxy/src/graphql/subscription.rs | 598 ++++++++++++++++-- backend/graph-proxy/src/graphql/workflows.rs | 11 +- backend/graph-proxy/src/main.rs | 1 - frontend/.devcontainer/devcontainer-lock.json | 14 + .../src/routes/SingleWorkflowPage.tsx | 11 +- .../lib/components/BaseWorkflowRelay.tsx | 23 +- .../lib/components/RelayEnvironment.ts | 56 +- .../lib/components/TasksFlow.tsx | 14 +- frontend/relay-workflows-lib/lib/main.ts | 1 + .../lib/utils/coreUtils.ts | 11 +- .../lib/views/BaseSingleWorkflowView.tsx | 106 +++- .../lib/views/LiveSingleWorkflowView.tsx | 6 + .../lib/views/SingleWorkflowView.tsx | 17 +- .../lib/views/TaskLogViewer.tsx | 262 ++++++++ .../tests/components/TasksDynamic.test.tsx | 1 + .../views/BaseSingleWorkflowView.test.tsx | 2 + 16 files changed, 1012 insertions(+), 122 deletions(-) create mode 100644 frontend/.devcontainer/devcontainer-lock.json create mode 100644 frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx diff --git a/backend/graph-proxy/src/graphql/subscription.rs b/backend/graph-proxy/src/graphql/subscription.rs index a3dcea797..0752dbff6 100644 --- a/backend/graph-proxy/src/graphql/subscription.rs +++ b/backend/graph-proxy/src/graphql/subscription.rs @@ -12,6 +12,7 @@ use crate::{ workflows::{Workflow, WorkflowParsingError}, VisitInput, }, + s3client::{Client as S3Client, S3Bucket}, validate_token::ValidatedAuthToken, ArgoServerUrl, }; @@ -46,7 +47,7 @@ struct LogContent { pod_name: String, } -/// Succees/fail events from Workflows API +/// Success/fail events from Workflows API #[derive(Debug, Deserialize)] struct WatchEvent { /// Successful event @@ -58,6 +59,7 @@ struct WatchEvent { /// Get authentication token pub fn get_auth_token(ctx: &Context<'_>) -> anyhow::Result { let auth_token = ctx.data_unchecked::().as_token(); + auth_token .as_ref() .map(|auth| auth.token().to_string()) @@ -66,7 +68,11 @@ pub fn get_auth_token(ctx: &Context<'_>) -> anyhow::Result { #[Subscription(guard = "AuthGuard")] impl WorkflowsSubscription { - /// Processing to subscribe to logs for a single pod of a workflow + /// Subscribe to logs for a single pod of a workflow. + /// + /// Logs are streamed live from Argo while the pod is running. + /// Once the Argo stream finishes, the archived main.log is retrieved + /// from S3 and any lines not already sent are emitted. async fn logs( &self, ctx: &Context<'_>, @@ -76,10 +82,17 @@ impl WorkflowsSubscription { ) -> anyhow::Result>> { let auth_token = get_auth_token(ctx)?; - let namespace = visit.to_string(); + if task_id.is_empty() || task_id == "__NO_TASK_SELECTED__" { + return Err(anyhow::anyhow!( + "A valid task ID is required to retrieve task logs" + )); + } + let server_url = ctx.data_unchecked::().deref().clone(); let mut url = server_url; + let namespace = visit.to_string(); + url.path_segments_mut().expect("Invalid base URL").extend([ "api", "v1", @@ -94,53 +107,386 @@ impl WorkflowsSubscription { .append_pair("logOptions.container", "main") .append_pair("logOptions.follow", "true"); - let client = reqwest::Client::new(); - let response = client - .get(url) - .bearer_auth(auth_token) - .header("Accept", "text/plain") + let s3_client = ctx + .data::() + .map_err(|_| anyhow::anyhow!("Missing S3 client"))? + .clone(); + + let s3_bucket = ctx + .data::() + .map_err(|_| anyhow::anyhow!("Missing S3 bucket"))? + .clone(); + + let s3_key = format!("{workflow_name}/{task_id}/main.log"); + + // Check S3 first because the task pod may already have been deleted + // while the overall workflow is still running. + let initial_archive = match s3_client + .get_object() + .bucket(s3_bucket.clone()) + .key(&s3_key) .send() - .await?; + .await + { + Ok(response) => { + tracing::info!( + "ARCHIVE_FOUND_BEFORE_LIVE task={} workflow={} key={}", + task_id, + workflow_name, + s3_key + ); + + Some(response) + } + + Err(_) => { + tracing::info!( + "ARCHIVE_NOT_FOUND_BEFORE_LIVE task={} workflow={}", + task_id, + workflow_name + ); + + None + } + }; + + // Only contact the Argo live-log endpoint when the archived + // main.log is not already available in S3. + let live_response = if initial_archive.is_none() { + tracing::info!( + "STARTING_LIVE_STREAM namespace={} workflow={} task={}", + namespace, + workflow_name, + task_id + ); + + let client = reqwest::Client::new(); + + Some( + client + .get(url) + .bearer_auth(auth_token) + .header("Accept", "text/plain") + .send() + .await?, + ) + } else { + None + }; + + let live_status = live_response.as_ref().map(|response| response.status()); + + let mut byte_stream = live_response.map(|response| response.bytes_stream()); - let status = response.status(); - let byte_stream = response.bytes_stream(); let log_stream = stream! { - for await chunk_result in byte_stream { - match chunk_result { - Ok(chunk) if status.is_success() => { - let text = String::from_utf8_lossy(&chunk).to_string(); - for line in text.lines() { - match serde_json::from_str::(line) { - Ok(parsed) => { - if let Some(result) = parsed.result { - yield Ok(LogEntry { - content: result.content, - pod_name: result.pod_name, - }); - } else { - yield Err("Missing result in log response".to_string()); + if let Some(archive_response) = initial_archive { + let archive_bytes = match archive_response.body.collect().await { + Ok(bytes) => bytes, + + Err(err) => { + yield Err(format!( + "Failed to read archived log artifact: {err}" + )); + return; + } + }; + + let archived_text = String::from_utf8_lossy( + archive_bytes.into_bytes().as_ref(), + ) + .to_string(); + + for line in archived_text.lines() { + let content = line.trim_end(); + + if content.is_empty() + || should_skip_log_line(content) + { + continue; + } + + yield Ok(LogEntry { + content: content.to_string(), + pod_name: task_id.clone(), + }); + } + + tracing::info!( + "ARCHIVE_EMITTED_BEFORE_LIVE task={} workflow={} key={}", + task_id, + workflow_name, + s3_key + ); + + return; + } + + let mut live_lines = Vec::new(); + let mut archive_check = + tokio::time::interval(std::time::Duration::from_secs(2)); + archive_check.tick().await; + archive_check.set_missed_tick_behavior( + tokio::time::MissedTickBehavior::Skip, + ); + + 'live_stream: loop { + tokio::select! { + chunk_result = async { + match byte_stream.as_mut() { + Some(stream) => stream.next().await, + None => None, + } + } => { + match chunk_result { + Some(Ok(chunk)) + if live_status + .map(|status| status.is_success()) + .unwrap_or(false) => + { + let text = + String::from_utf8_lossy(&chunk).to_string(); + + for line in text.lines() { + match serde_json::from_str::(line) { + Ok(parsed) => { + if let Some(result) = parsed.result { + let content = result.content; + + let skip_line = + content.contains("capturing logs") + || content.contains("waiting for signals") + || content.contains("sub-process exited") + || content.contains("file signal handler exiting") + || content.contains("no need to save artifact") + || content.contains("no need to save parameter"); + + if skip_line { + continue; + } + + live_lines.push(content.clone()); + + yield Ok(LogEntry { + content, + pod_name: result.pod_name, + }); + } else { + yield Err( + "Missing result in log response" + .to_string() + ); + } + } + + Err(_) => { + let content = line.trim().to_string(); + + if content.starts_with("{\"result\"") { + continue; + } + + if !content.is_empty() + && !should_skip_log_line(&content) + { + live_lines.push(content.clone()); + + yield Ok(LogEntry { + content, + pod_name: task_id.clone(), + }); + } + } } } - Err(_) => { - yield Ok(LogEntry { - content: line.trim().to_string(), - pod_name: task_id.clone(), - }); - } } + + Some(Ok(_)) => { + tracing::info!( + "Live log request unavailable for task {}; checking archive", + task_id + ); + + break 'live_stream; + } + + Some(Err(err)) => { + tracing::warn!( + "Live log stream ended for task {}: {}; checking archive", + task_id, + err + ); + + break 'live_stream; + } + + None => { + tracing::info!( + "Live log stream completed for task {}; checking archive", + task_id + ); + + break 'live_stream; + } + } + } + + _ = archive_check.tick() => { + let archive_available = s3_client + .head_object() + .bucket(s3_bucket.clone()) + .key(&s3_key) + .send() + .await + .is_ok(); + + if archive_available { + tracing::info!( + "Archived log is available before workflow completion: {}", + s3_key + ); + + break 'live_stream; + } + } + } + } + + // The live Argo stream has finished. The durable log should now + // be available in the S3 artifact. + tracing::info!( + "LIVE_STREAM_ENDED task={} workflow={}", + task_id, + workflow_name + ); + + tracing::info!( + "ARCHIVE_LOOKUP bucket={} key={}", + s3_bucket.0, + s3_key + ); + + let archive_response = { + let mut attempts = 0; + + loop { + attempts += 1; + + match s3_client + .get_object() + .bucket(s3_bucket.clone()) + .key(&s3_key) + .send() + .await + { + Ok(response) => { + tracing::info!( + "ARCHIVE_RETRIEVED task={} workflow={} key={}", + task_id, + workflow_name, + s3_key + ); + + break response; + } + + Err(_) if attempts < 10 => { + tracing::info!( + "Archived log not available yet for task {}; \ + retrying S3 lookup, attempt {}", + task_id, + attempts + ); + + tokio::time::sleep( + std::time::Duration::from_secs(1) + ) + .await; + } + + Err(_err) => { + yield Err("No logs available".to_string()); + return; } } - Ok(_) | Err(_) => { - yield Err("Failed to read log chunk".to_string()); + } + }; + let archive_bytes = match archive_response.body.collect().await { + Ok(bytes) => bytes, + + Err(err) => { + yield Err(format!( + "Failed to read archived log artifact: {err}" + )); + return; + } + }; + + let archived_text = + String::from_utf8_lossy(archive_bytes.into_bytes().as_ref()).to_string(); + + let archived_lines: Vec = archived_text + .lines() + .filter(|line| { + !line.contains("capturing logs") + && !line.contains("waiting for signals") + && !line.contains("sub-process exited") + && !line.contains("file signal handler exiting") + && !line.contains("no need to save artifact") + && !line.contains("no need to save parameter") + }) + .map(str::to_string) + .collect(); + + // Determine where the archived log begins beyond what was already + // sent by the live Argo stream. + let mut archive_start = 0; + + while archive_start < live_lines.len() + && archive_start < archived_lines.len() + && live_lines[archive_start] == archived_lines[archive_start] + { + archive_start += 1; + } + + // If the archive and live stream don't share the same prefix, + // try to locate the final live line in the archive. + if archive_start < live_lines.len() { + if let Some(last_live_line) = live_lines.last() { + if let Some(position) = archived_lines + .iter() + .rposition(|line| line == last_live_line) + { + archive_start = position + 1; + } else { + yield Err( + "Unable to reconcile live and archived logs".to_string() + ); + return; } } } + tracing::info!( + "ARCHIVE_DEBUG task={} live_lines={} archived_lines={} archive_start={}", + task_id, + live_lines.len(), + archived_lines.len(), + archive_start + ); + + // Send only the archived lines that were not already emitted + // from the live Argo stream. + for line in archived_lines.into_iter().skip(archive_start) { + yield Ok(LogEntry { + content: line, + pod_name: task_id.clone(), + }); + } }; Ok(log_stream) } - /// Processing to subscribe to data for all workflows in a session + /// Subscribe to data for all workflows in a session. async fn workflow( &self, ctx: &Context<'_>, @@ -166,6 +512,7 @@ impl WorkflowsSubscription { ); let client = reqwest::Client::new(); + let response = client .get(url) .bearer_auth(auth_token) @@ -177,6 +524,7 @@ impl WorkflowsSubscription { let stream = response.then(move |event_result| { let session_clone = visit.clone(); + async move { match event_result { Ok(event) => { @@ -191,14 +539,18 @@ impl WorkflowsSubscription { Err("No workflow object returned".to_string()) } } + (None, Some(err)) => Err(err.message), + (None, None) => Err("Missing result and error in event".to_string()), + (Some(_), Some(_)) => { Err("Conflicting result and error in event".to_string()) } } } - Err(_err) => Err("Failed to read event from stream".to_string()), + + Err(_) => Err("Failed to read event from stream".to_string()), } } }); @@ -207,6 +559,16 @@ impl WorkflowsSubscription { } } +/// Returns true for Argo executor messages that should not be displayed. +fn should_skip_log_line(content: &str) -> bool { + content.contains("capturing logs") + || content.contains("waiting for signals") + || content.contains("sub-process exited") + || content.contains("file signal handler exiting") + || content.contains("no need to save artifact") + || content.contains("no need to save parameter") +} + /// Struct for storing message of StreamError #[derive(Debug, Deserialize)] struct StreamError { @@ -216,7 +578,6 @@ struct StreamError { #[cfg(test)] mod tests { - use std::{env, fs, path::PathBuf}; use async_graphql::Request; @@ -227,20 +588,134 @@ mod tests { use serde_json::{json, Value}; use url::Url; - use crate::graphql::Visit; - use crate::ArgoServerUrl; - use crate::graphql::root_schema_builder; + use crate::graphql::Visit; use crate::validate_token::ValidatedAuthToken; + use crate::{ArgoServerUrl, Client, S3Bucket, S3ClientArgs}; fn test_token() -> ValidatedAuthToken { let token = Authorization::bearer("test-token").expect("token always valid"); + ValidatedAuthToken::Valid(token) } + #[tokio::test] + async fn logs_subscription_reads_archived_s3_log_before_live_stream() { + let workflow_name = "numpy-benchmark-wdkwj"; + let task_id = "numpy-benchmark-wdkwj"; + + let visit = Visit { + proposal_code: "mg".to_string(), + proposal_number: 36964, + number: 1, + }; + + let mut server = mockito::Server::new_async().await; + + // No Argo log request is expected. + // + // The implementation now checks S3 first and returns the + // archived log immediately when main.log exists. + + // Mock the archived S3 main.log. + // + // Path-style S3 addressing produces: + // + // /test-bucket/numpy-benchmark-wdkwj/numpy-benchmark-wdkwj/main.log + let _s3_key = format!("{workflow_name}/{task_id}/main.log"); + // let s3_path = format!("/test-bucket/{s3_key}"); + + let s3_log_endpoint = server + .mock("GET", mockito::Matcher::Any) + .with_status(200) + .with_header("content-type", "text/plain") + .with_body("line 1\nline 2\nline 3\nline 4\n") + .create_async() + .await; + + let s3_bucket = S3Bucket("test-bucket".to_string()); + + let s3_client_args = S3ClientArgs { + s3_endpoint_url: Some(Url::parse(&server.url()).unwrap()), + s3_access_key_id: Some("test-access-key".to_string()), + s3_secret_access_key: Some("test-secret-key".to_string()), + s3_force_path_style: true, + s3_region: Some("us-west-2".to_string()), + }; + + let s3_client = Client::from(s3_client_args); + + let argo_server_url = Url::parse(&server.url()).unwrap(); + + let schema = root_schema_builder() + .data(ArgoServerUrl(argo_server_url)) + .data(test_token()) + .data(s3_client) + .data(s3_bucket) + .finish(); + + let request = Request::new(format!( + r#" + subscription {{ + logs( + visit: {{ + proposalCode: "{}", + proposalNumber: {}, + number: {} + }} + workflowName: "{}" + taskId: "{}" + ) {{ + content + podName + }} + }} + "#, + visit.proposal_code, visit.proposal_number, visit.number, workflow_name, task_id, + )); + + let mut response_stream = schema.execute_stream(request); + + let mut logs = Vec::new(); + + while let Some(response) = response_stream.next().await { + assert!( + response.errors.is_empty(), + "unexpected GraphQL errors: {:?}", + response.errors + ); + + let data = response.data.into_json().expect("invalid response JSON"); + + if let Some(log) = data.get("logs").and_then(|value| value.as_object()) { + logs.push(( + log["content"].as_str().unwrap().to_string(), + log["podName"].as_str().unwrap().to_string(), + )); + } + + if logs.len() == 4 { + break; + } + } + + assert_eq!( + logs, + vec![ + ("line 1".to_string(), task_id.to_string()), + ("line 2".to_string(), task_id.to_string()), + ("line 3".to_string(), task_id.to_string()), + ("line 4".to_string(), task_id.to_string()), + ] + ); + + s3_log_endpoint.assert_async().await; + } + #[tokio::test] async fn single_workflow_subscription_returns_first_event() { let workflow_name = "numpy-benchmark-wdkwj"; + let visit = Visit { proposal_code: "mg".to_string(), proposal_number: 36964, @@ -248,6 +723,7 @@ mod tests { }; let mut workflow_file_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")); + workflow_file_path.push("test-assets"); workflow_file_path.push("get-workflow-wdkwj.json"); @@ -270,7 +746,9 @@ mod tests { ); let mut server = mockito::Server::new_async().await; + let path = format!("/api/v1/workflow-events/{visit}"); + let workflow_events_endpoint = server .mock("GET", path.as_str()) .match_query(Matcher::UrlEncoded( @@ -292,15 +770,19 @@ mod tests { let request = Request::new(format!( r#" - subscription {{ - workflow( - name: "{}", - visit: {{ proposalCode: "{}", proposalNumber: {}, number: {} }} - ) {{ - name + subscription {{ + workflow( + name: "{}", + visit: {{ + proposalCode: "{}", + proposalNumber: {}, + number: {} + }} + ) {{ + name + }} }} - }} - "#, + "#, workflow_name, visit.proposal_code, visit.proposal_number, visit.number )); @@ -346,15 +828,19 @@ mod tests { let request = Request::new( r#" - subscription { - workflow( - name: "workflowName", - visit: { proposalCode: "xy", proposalNumber: 1234, number: 5678 } - ) { - name + subscription { + workflow( + name: "workflowName", + visit: { + proposalCode: "xy", + proposalNumber: 1234, + number: 5678 + } + ) { + name + } } - } - "#, + "#, ); let mut response_stream = schema.execute_stream(request); @@ -365,6 +851,7 @@ mod tests { .expect("subscription stream ended before first response"); let expected_data = json!(null); + assert_eq!( first_response .data @@ -384,6 +871,7 @@ mod tests { .expect("invalid json"); let expected_value = json!(AuthErrorCode::Unauthenticated.to_string()); + assert_eq!(error_code, expected_value); } } diff --git a/backend/graph-proxy/src/graphql/workflows.rs b/backend/graph-proxy/src/graphql/workflows.rs index e40240d8f..488a6a115 100644 --- a/backend/graph-proxy/src/graphql/workflows.rs +++ b/backend/graph-proxy/src/graphql/workflows.rs @@ -373,16 +373,17 @@ impl Artifact<'_> { .expires_in(std::time::Duration::from_secs(3600)) .build() .unwrap(); - s3_client + let req = s3_client .get_object() .bucket(s3_bucket.clone()) .key(key) .presigned(presigning_config) .await - .map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) - .and_then(|req| { - Url::parse(req.uri()).map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) - }) + .map_err(|_| WorkflowParsingError::InvalidPresignedS3Url)?; + + tracing::info!("PRESIGNED URL: {}", req.uri()); + + Url::parse(req.uri()).map_err(|_| WorkflowParsingError::InvalidPresignedS3Url) } /// The MIME type of the artifact data diff --git a/backend/graph-proxy/src/main.rs b/backend/graph-proxy/src/main.rs index b83e08c6e..92d307f13 100644 --- a/backend/graph-proxy/src/main.rs +++ b/backend/graph-proxy/src/main.rs @@ -138,7 +138,6 @@ async fn main() { // it's necessary. probably needs a tweak on the telemetry lib. let metrics = otlp_guard.meter_provider.as_ref().unwrap(); let metrics_state = Arc::new(Metrics::new(metrics)); - info!(?args, "Starting GraphQL Server"); let s3_client = Client::from(args.s3_client); let schema = root_schema_builder() diff --git a/frontend/.devcontainer/devcontainer-lock.json b/frontend/.devcontainer/devcontainer-lock.json new file mode 100644 index 000000000..f1bff2f3e --- /dev/null +++ b/frontend/.devcontainer/devcontainer-lock.json @@ -0,0 +1,14 @@ +{ + "features": { + "ghcr.io/devcontainers/features/common-utils:2.4.2": { + "version": "2.4.2", + "resolved": "ghcr.io/devcontainers/features/common-utils@sha256:bebfdcd6097a35506bf0f064a31e52ad4205467d9d7a226a688f51b851c88b65", + "integrity": "sha256:bebfdcd6097a35506bf0f064a31e52ad4205467d9d7a226a688f51b851c88b65" + }, + "ghcr.io/devcontainers/features/git-lfs:1.2.3": { + "version": "1.2.3", + "resolved": "ghcr.io/devcontainers/features/git-lfs@sha256:7acbf0a99325949b6a2906ebf5aa421dad72b89ab8045031a60e69cb394a16b1", + "integrity": "sha256:7acbf0a99325949b6a2906ebf5aa421dad72b89ab8045031a60e69cb394a16b1" + } + } +} diff --git a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx index 9a0fa0f36..e38056187 100644 --- a/frontend/dashboard/src/routes/SingleWorkflowPage.tsx +++ b/frontend/dashboard/src/routes/SingleWorkflowPage.tsx @@ -3,11 +3,12 @@ import { useParams, Link, useSearchParams } from "react-router-dom"; import { Suspense, useMemo } from "react"; import "react-resizable/css/styles.css"; import { Breadcrumbs } from "@diamondlightsource/sci-react-ui"; -import { SingleWorkflowView, WorkflowsNavbar } from "relay-workflows-lib"; + import { visitTextToVisit, WorkflowErrorBoundaryWithRetry, } from "workflows-lib"; +import { SingleWorkflowView, WorkflowsNavbar } from "relay-workflows-lib"; function SingleWorkflowPage() { const { visitid, workflowName } = useParams<{ @@ -65,6 +66,13 @@ function SingleWorkflowPage() { workflowName={workflowName} taskIds={taskIds} /> + + {/* Real-time Task Log Viewer */} + {/* */} )} @@ -79,7 +87,6 @@ function SingleWorkflowPage() { mb={4} > No valid workflow selected - {/* Go to instrumentSession or home page */} )} diff --git a/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx b/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx index fb92e1a42..36dccaf52 100644 --- a/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx +++ b/frontend/relay-workflows-lib/lib/components/BaseWorkflowRelay.tsx @@ -35,6 +35,7 @@ interface BaseWorkflowRelayProps { expanded?: boolean; onChange?: () => void; fragmentRef: BaseWorkflowRelayFragment$key; + onSelectTask?: (taskId: string) => void; } export default function BaseWorkflowRelay({ @@ -43,13 +44,18 @@ export default function BaseWorkflowRelay({ expanded, onChange, fragmentRef, + onSelectTask, }: BaseWorkflowRelayProps) { const { workflowName: workflowNameURL } = useParams<{ workflowName: string; }>(); + const navigate = useNavigate(); + const data = useFragment(BaseWorkflowRelayFragment, fragmentRef); + const statusText = data.status?.__typename ?? "Unknown"; + const [selectedTaskIds, setSelectedTaskIds] = useSelectedTaskIds(); const onNavigate = React.useCallback( @@ -57,6 +63,7 @@ export default function BaseWorkflowRelay({ const isCtrl = event?.ctrlKey || event?.metaKey; let updatedTaskIds: string[]; + console.log("TASK CLICKED", taskId); if (isCtrl) { updatedTaskIds = selectedTaskIds.includes(taskId) @@ -65,12 +72,24 @@ export default function BaseWorkflowRelay({ } else { updatedTaskIds = [taskId]; } + setSelectedTaskIds(updatedTaskIds); + if (workflowNameURL !== data.name) { void navigate(`/workflows/${visitToText(data.visit)}/${data.name}`); } - setSelectedTaskIds(updatedTaskIds); + + if (onSelectTask) { + onSelectTask(taskId); + } }, - [navigate, selectedTaskIds, setSelectedTaskIds, workflowNameURL, data], + [ + navigate, + selectedTaskIds, + setSelectedTaskIds, + workflowNameURL, + data, + onSelectTask, + ], ); return ( diff --git a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts index 40bbeaf47..5071cfddd 100644 --- a/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts +++ b/frontend/relay-workflows-lib/lib/components/RelayEnvironment.ts @@ -77,7 +77,7 @@ const fetchFn: FetchFunction = async (request, variables) => { const resp = await fetch(HTTP_ENDPOINT, { method: "POST", headers, - credentials: "include", + // credentials: "include", body: JSON.stringify({ query: request.text, // <-- The GraphQL document composed by Relay variables, @@ -91,9 +91,34 @@ const fetchFn: FetchFunction = async (request, variables) => { return await resp.json(); // eslint-disable-line @typescript-eslint/no-unsafe-return }; - +console.log("HTTP_ENDPOINTXXXXXXXXXXXXXXXXXXXXXXXXXXXX:", HTTP_ENDPOINT); +console.log( + "WS_ENDPOINTYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYYY:", + WS_ENDPOINT, +); export const wsClient = createClient({ url: WS_ENDPOINT, + on: { + connecting: () => { + console.log("WS connecting"); + }, + opened: () => { + console.log("WS opened"); + }, + connected: () => { + console.log("WS connected"); + }, + closed: (event) => { + console.log("WS closed", event); + }, + }, + webSocketImpl: class extends WebSocket { + constructor(url: string | URL, protocols?: string | string[]) { + console.log("Creating browser WebSocket:", url); + super(url, protocols); + } + }, + connectionParams: async () => { if (!USE_AUTH_GATEWAY && !keycloak.authenticated) { await ensureKeycloakInit(); @@ -108,6 +133,8 @@ export const wsClient = createClient({ }); const subscribeFn: SubscribeFunction = (operation, variables) => { + console.log("WS SUBSCRIBE STARTED:", operation.name, variables); + return Observable.create((sink) => { const cleanup = wsClient.subscribe( { @@ -117,20 +144,23 @@ const subscribeFn: SubscribeFunction = (operation, variables) => { }, { next: (response) => { - const data = response.data; - if (data) { - sink.next({ data } as GraphQLResponse); - } else if (data == null) { - console.warn("Data is null:", response); - } else { - console.error("Subscription error response:", response); - sink.error(new Error("Subscription response missing data")); - } + console.log("WS SUBSCRIPTION RESPONSE:", response); + + sink.next(response as GraphQLResponse); + }, + + error: (error) => { + console.error("WS SUBSCRIPTION ERROR:", error); + sink.error(error as Error); + }, + + complete: () => { + console.log("WS SUBSCRIPTION COMPLETE"); + sink.complete(); }, - error: sink.error.bind(sink), - complete: sink.complete.bind(sink), }, ); + return cleanup; }); }; diff --git a/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx b/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx index d83256560..9c9394d37 100644 --- a/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx +++ b/frontend/relay-workflows-lib/lib/components/TasksFlow.tsx @@ -144,30 +144,38 @@ const TasksFlow: React.FC = ({ useEffect(() => { const handleResizeAndOverflow = () => { - if (containerRef.current) { + if (containerRef.current && reactFlowInstance.current) { const { width, height } = containerRef.current.getBoundingClientRect(); - const boundingBox = getNodesBounds(layoutedNodes); + + const nodeLookup = reactFlowInstance.current.getNodes(); + + const boundingBox = getNodesBounds(nodeLookup); + setIsOverflow(boundingBox.width > width || boundingBox.height > height); } }; + const resizeObserver = new ResizeObserver(handleResizeAndOverflow); + const currentContainerRef = containerRef.current; if (currentContainerRef) { resizeObserver.observe(currentContainerRef); } + handleResizeAndOverflow(); + window.addEventListener("resize", handleResizeAndOverflow); return () => { if (currentContainerRef) { resizeObserver.unobserve(currentContainerRef); } + resizeObserver.disconnect(); window.removeEventListener("resize", handleResizeAndOverflow); }; }, [layoutedNodes, layoutedEdges]); - return ( void; +} + +interface TaskTreeNode { + id: string; + children?: TaskTreeNode[]; } export default function BaseSingleWorkflowView({ taskIds, fragmentRef, + selectedTaskId, + onSelectTask, }: BaseSingleWorkflowViewProps) { const data = useFragment(BaseSingleWorkflowViewFragment, fragmentRef); + const fetchedTasks = useFetchedTasks(data ?? null); - const [selectedTaskIds, setSelectedTaskIds] = useSelectedTaskIds(); + const [filledTaskId, setFilledTaskId] = useState(null); + // Resolve task name from id + const selectedTask = useMemo( + () => fetchedTasks.find((task) => task.id === selectedTaskId), + [fetchedTasks, selectedTaskId], + ); + + const [selectedTaskIds, setSelectedTaskIds] = useSelectedTaskIds(); + + // // Artifact hover highlight + // const [ + // filledTaskId, + // setFilledTaskId, + // ] = useState(null); + const taskTree = useMemo(() => buildTaskTree(fetchedTasks), [fetchedTasks]); const outputTaskIds: string[] = useMemo(() => { const newOutputTaskIds: string[] = []; - const traverse = (tasks: TaskNode[]) => { + + const traverse = (tasks: TaskTreeNode[]) => { const sortedTasks = [...tasks].sort((a, b) => a.id.localeCompare(b.id)); + sortedTasks.forEach((taskNode) => { - if ( - taskNode.children && - taskNode.children.length === 0 && - !newOutputTaskIds.includes(taskNode.id) - ) { - newOutputTaskIds.push(taskNode.id); + if (taskNode.children && taskNode.children.length === 0) { + if (!newOutputTaskIds.includes(taskNode.id)) { + newOutputTaskIds.push(taskNode.id); + } } else if (taskNode.children && taskNode.children.length > 0) { traverse(taskNode.children); } }); }; + traverse(taskTree); + return newOutputTaskIds; }, [taskTree]); @@ -70,14 +97,12 @@ export default function BaseSingleWorkflowView({ const handleSelectClear = () => { setSelectedTaskIds([]); + onSelectTask(null); }; - const onArtifactHover = useCallback( - (artifact: Artifact | null) => { - setFilledTaskId(artifact ? artifact.parentTaskId : null); - }, - [setFilledTaskId], - ); + const onArtifactHover = useCallback((artifact: Artifact | null) => { + setFilledTaskId(artifact ? artifact.parentTaskId : null); + }, []); useEffect(() => { setSelectedTaskIds(taskIds ?? []); @@ -89,6 +114,7 @@ export default function BaseSingleWorkflowView({ .map((id) => fetchedTasks.find((task) => task.id === id)) .filter((task): task is Task => !!task) : fetchedTasks; + return filteredTasks.flatMap((task) => task.artifacts); }, [selectedTaskIds, fetchedTasks]); @@ -120,7 +146,10 @@ export default function BaseSingleWorkflowView({ display="flex" flexDirection="column" gap={1} - sx={{ position: "absolute", left: "-100px" }} + sx={{ + position: "absolute", + left: "-100px", + }} > OUTPUT + CLEAR - {fragmentRef && ( - - )} + { + onSelectTask(taskId); + }} + /> + + + {taskIds && ( )} - {} + + ); } diff --git a/frontend/relay-workflows-lib/lib/views/LiveSingleWorkflowView.tsx b/frontend/relay-workflows-lib/lib/views/LiveSingleWorkflowView.tsx index e708f3ff9..e9a43f364 100644 --- a/frontend/relay-workflows-lib/lib/views/LiveSingleWorkflowView.tsx +++ b/frontend/relay-workflows-lib/lib/views/LiveSingleWorkflowView.tsx @@ -26,6 +26,8 @@ const LiveSingleWorkflowViewSubscriptionQuery = graphql` interface LiveWorkflowRelayProps extends SingleWorkflowViewProps { onNullSubscriptionData: () => void; + selectedTaskId: string | null; + onSelectTask: (taskId: string | null) => void; } export default function LiveWorkflowView({ @@ -33,6 +35,8 @@ export default function LiveWorkflowView({ workflowName, taskIds, onNullSubscriptionData, + selectedTaskId, + onSelectTask, }: LiveWorkflowRelayProps) { const [workflowFragmentRef, setWorkflowFragmentRef] = useState(null); @@ -75,6 +79,8 @@ export default function LiveWorkflowView({ ) : null; } diff --git a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx index 3aed782dc..b57b719c7 100644 --- a/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx +++ b/frontend/relay-workflows-lib/lib/views/SingleWorkflowView.tsx @@ -23,6 +23,7 @@ export interface SingleWorkflowViewProps { workflowName: string; taskIds?: string[]; onNullSubscriptionData?: () => void; + onSelectTask?: (taskId: string) => void; } export default function SingleWorkflowView(props: SingleWorkflowViewProps) { @@ -33,22 +34,30 @@ export default function SingleWorkflowView(props: SingleWorkflowViewProps) { name: props.workflowName, }, ); - const finished = - queryData.workflow?.status?.__typename && - finishedStatuses.has(queryData.workflow.status.__typename); + const workflow = queryData.workflow; + + const status = workflow?.status?.__typename; + const finished = status !== undefined && finishedStatuses.has(status); + const [isNull, setIsNull] = useState(false); const onNullSubscriptionData = () => { setIsNull(true); }; + const [selectedTaskId, setSelectedTaskId] = useState(null); + return finished || isNull ? ( ) : ( ); diff --git a/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx new file mode 100644 index 000000000..608119567 --- /dev/null +++ b/frontend/relay-workflows-lib/lib/views/TaskLogViewer.tsx @@ -0,0 +1,262 @@ +import React, { + Dispatch, + SetStateAction, + useEffect, + useMemo, + useRef, + useState, +} from "react"; +import { + Accordion, + AccordionSummary, + AccordionDetails, + Box, + Typography, + CircularProgress, +} from "@mui/material"; +import ArrowDropDownIcon from "@mui/icons-material/ArrowDropDown"; +import { graphql, useSubscription } from "react-relay"; +import { GraphQLSubscriptionConfig } from "relay-runtime"; +import { Visit } from "@diamondlightsource/sci-react-ui"; +import { TaskLogViewerSubscription } from "./__generated__/TaskLogViewerSubscription.graphql"; + +interface TaskLogViewerProps { + visit: Visit; + workflowName: string; + selectedTaskId: string | null; + selectedTaskName?: string; +} + +interface TaskLogSubscriptionProps { + visit: Visit; + workflowName: string; + taskId: string; + setLogLines: Dispatch>; + setTaskCompleted: Dispatch>; + setSubscriptionError: Dispatch>; +} + +const taskLogViewerSubscription = graphql` + subscription TaskLogViewerSubscription( + $visit: VisitInput! + $workflowName: String! + $taskId: String! + ) { + logs(visit: $visit, workflowName: $workflowName, taskId: $taskId) { + content + podName + } + } +`; + +const TaskLogSubscription: React.FC = ({ + visit, + workflowName, + taskId, + setLogLines, + setTaskCompleted, + setSubscriptionError, +}) => { + const subscriptionConfig = useMemo< + GraphQLSubscriptionConfig + >( + () => ({ + subscription: taskLogViewerSubscription, + variables: { + visit, + workflowName, + taskId, + }, + onNext: (payload) => { + const line = payload?.logs.content; + + if (line) { + setLogLines((previousLines) => [...previousLines, line]); + } + }, + onError: (error) => { + console.error("Log subscription error:", error); + + const message = error instanceof Error ? error.message : String(error); + + if ( + message.includes("NoSuchKey") || + message.includes("No logs") || + message.includes("Failed to retrieve archived log artifact") + ) { + setSubscriptionError("No logs available"); + } else { + setSubscriptionError("Unable to retrieve task logs"); + } + + setTaskCompleted(true); + }, + onCompleted: () => { + setTaskCompleted(true); + }, + }), + [ + visit, + workflowName, + taskId, + setLogLines, + setTaskCompleted, + setSubscriptionError, + ], + ); + + useSubscription(subscriptionConfig); + + return null; +}; + +const TaskLogViewerContent: React.FC = ({ + visit, + workflowName, + selectedTaskId, + selectedTaskName, +}) => { + const [logLines, setLogLines] = useState([]); + const [taskCompleted, setTaskCompleted] = useState(false); + const [subscriptionError, setSubscriptionError] = useState( + null, + ); + const [expanded, setExpanded] = useState(Boolean(selectedTaskId)); + + const containerRef = useRef(null); + + useEffect(() => { + if (containerRef.current) { + containerRef.current.scrollTop = containerRef.current.scrollHeight; + } + }, [logLines]); + + return ( + <> + {selectedTaskId && ( + + )} + + { + setExpanded(isExpanded); + }} + sx={{ + mt: 2, + width: "100%", + backgroundColor: "#f3f5f3", + color: "#030303", + }} + > + } + > + + Logs: {selectedTaskName ?? selectedTaskId ?? "No task selected"} + + + {selectedTaskId && !taskCompleted && ( + + )} + + {selectedTaskId && taskCompleted && ( + + ARCHIVED + + )} + + + + + {subscriptionError ? ( + + {subscriptionError} + + ) : logLines.length > 0 ? ( + logLines.map((line, index) => ( + + {line} + {index < logLines.length - 1 && "\n"} + + )) + ) : ( + + {selectedTaskId ? "Waiting for logs..." : "No task selected"} + + )} + + + + + ); +}; + +export const TaskLogViewer: React.FC = ({ + visit, + workflowName, + selectedTaskId, + selectedTaskName, +}) => { + return ( + + ); +}; diff --git a/frontend/relay-workflows-lib/tests/components/TasksDynamic.test.tsx b/frontend/relay-workflows-lib/tests/components/TasksDynamic.test.tsx index 28a41cae0..c477de55e 100644 --- a/frontend/relay-workflows-lib/tests/components/TasksDynamic.test.tsx +++ b/frontend/relay-workflows-lib/tests/components/TasksDynamic.test.tsx @@ -14,6 +14,7 @@ vi.mock("@xyflow/react", () => ({ }) => { const mockInstance = { fitView: vi.fn(), + getNodes: vi.fn(() => []), } as unknown as ReactFlowInstance; onInit(mockInstance); return
; diff --git a/frontend/relay-workflows-lib/tests/views/BaseSingleWorkflowView.test.tsx b/frontend/relay-workflows-lib/tests/views/BaseSingleWorkflowView.test.tsx index 03027d971..09f287cba 100644 --- a/frontend/relay-workflows-lib/tests/views/BaseSingleWorkflowView.test.tsx +++ b/frontend/relay-workflows-lib/tests/views/BaseSingleWorkflowView.test.tsx @@ -39,6 +39,8 @@ const QueryWrappedBaseSingleWorkflowView = () => { ); };