diff --git a/crates/farm-controller/src/main.rs b/crates/farm-controller/src/main.rs index d991908..1bcbbff 100644 --- a/crates/farm-controller/src/main.rs +++ b/crates/farm-controller/src/main.rs @@ -92,6 +92,17 @@ impl AppScheduler { } } + fn list_task_attempt_logs( + &self, + task_id: TaskId, + attempt_id: uuid::Uuid, + ) -> Result, FarmError> { + match self { + Self::Memory(scheduler) => scheduler.list_task_attempt_logs(task_id, attempt_id), + Self::Sqlite(scheduler) => scheduler.list_task_attempt_logs(task_id, attempt_id), + } + } + fn record_worker_logs( &self, worker_id: WorkerId, @@ -272,6 +283,10 @@ fn app(scheduler: AppScheduler) -> Router { .route("/v1/tasks/{task_id}/started", post(mark_task_started)) .route("/v1/tasks/{task_id}/renew", post(renew_task_lease)) .route("/v1/tasks/{task_id}/complete", post(complete_task)) + .route( + "/v1/tasks/{task_id}/attempts/{attempt_id}/logs", + get(list_task_attempt_logs), + ) .route( "/v1/tasks/{task_id}/artifacts/{artifact_index}", get(download_task_artifact), @@ -383,6 +398,13 @@ async fn list_worker_logs( Ok(Json(scheduler.list_worker_logs(worker_id)?)) } +async fn list_task_attempt_logs( + State(scheduler): State, + Path((task_id, attempt_id)): Path<(TaskId, uuid::Uuid)>, +) -> Result>, ApiError> { + Ok(Json(scheduler.list_task_attempt_logs(task_id, attempt_id)?)) +} + async fn record_worker_logs( State(scheduler): State, Path(worker_id): Path, @@ -486,6 +508,7 @@ impl From for ApiError { | FarmError::TaskNotFound(_) | FarmError::WorkerNotFound(_) | FarmError::ArtifactNotFound { .. } => StatusCode::NOT_FOUND, + FarmError::AttemptNotFound { .. } => StatusCode::NOT_FOUND, FarmError::InvalidSubmission(_) => StatusCode::BAD_REQUEST, FarmError::InvalidLease | FarmError::InvalidState(_) => StatusCode::CONFLICT, FarmError::LockPoisoned | FarmError::Storage(_) => StatusCode::INTERNAL_SERVER_ERROR, diff --git a/crates/farm-core/src/models.rs b/crates/farm-core/src/models.rs index 0cbfd20..fdef237 100644 --- a/crates/farm-core/src/models.rs +++ b/crates/farm-core/src/models.rs @@ -485,6 +485,8 @@ pub struct Task { pub stdout_tail: Option, pub stderr_tail: Option, #[serde(default)] + pub attempt_records: Vec, + #[serde(default)] pub artifact_paths: Vec, #[serde(default)] pub artifacts: Vec, @@ -502,6 +504,41 @@ pub struct TaskArtifact { pub modified_at: Option>, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskAttempt { + pub id: Uuid, + pub number: u32, + pub worker_id: WorkerId, + pub state: TaskAttemptState, + pub leased_at: DateTime, + #[serde(default)] + pub started_at: Option>, + #[serde(default)] + pub completed_at: Option>, + #[serde(default)] + pub exit_code: Option, + #[serde(default)] + pub stdout_tail: Option, + #[serde(default)] + pub stderr_tail: Option, + #[serde(default)] + pub failure_summary: Option, + pub log_ref: String, + #[serde(default)] + pub artifacts: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum TaskAttemptState { + Leased, + Running, + Succeeded, + Failed, + Cancelled, + Expired, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum ArtifactKind { @@ -580,6 +617,8 @@ pub struct WorkerLogInput { pub job_id: Option, #[serde(default)] pub task_id: Option, + #[serde(default)] + pub attempt_id: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -596,6 +635,8 @@ pub struct FarmLogEntry { #[serde(default)] pub task_id: Option, #[serde(default)] + pub attempt_id: Option, + #[serde(default)] pub worker_id: Option, } @@ -641,6 +682,8 @@ pub enum WorkerState { pub struct TaskLeaseInfo { pub token: String, pub worker_id: WorkerId, + #[serde(default)] + pub attempt_id: Uuid, pub leased_at: DateTime, pub expires_at: DateTime, } diff --git a/crates/farm-core/src/scheduler.rs b/crates/farm-core/src/scheduler.rs index c09aaf2..59d7037 100644 --- a/crates/farm-core/src/scheduler.rs +++ b/crates/farm-core/src/scheduler.rs @@ -11,9 +11,9 @@ use uuid::Uuid; use crate::models::{ DashboardSnapshot, FarmLogEntry, FarmStats, Job, JobId, JobState, JobSubmit, LogLevel, LogSource, OpenJdAmountRequirement, OpenJdAttributeRequirement, ResourceLimitDefinition, - ResourceLimitSnapshot, Task, TaskArtifact, TaskComplete, TaskId, TaskLease, TaskLeaseInfo, - TaskLeaseRenewal, TaskStarted, TaskState, TaskSubmit, WorkerCapacity, WorkerId, WorkerInfo, - WorkerLogBatch, WorkerRegister, WorkerState, + ResourceLimitSnapshot, Task, TaskArtifact, TaskAttempt, TaskAttemptState, TaskComplete, TaskId, + TaskLease, TaskLeaseInfo, TaskLeaseRenewal, TaskStarted, TaskState, TaskSubmit, WorkerCapacity, + WorkerId, WorkerInfo, WorkerLogBatch, WorkerRegister, WorkerState, }; use crate::openjd::openjd_to_tasks; @@ -30,6 +30,8 @@ pub enum FarmError { task_id: TaskId, artifact_index: usize, }, + #[error("attempt was not found for task {task_id}: {attempt_id}")] + AttemptNotFound { task_id: TaskId, attempt_id: Uuid }, #[error("invalid submission: {0}")] InvalidSubmission(String), #[error("invalid state transition: {0}")] @@ -156,6 +158,7 @@ impl InMemoryScheduler { message: format!("job submitted: {}", job.name), job_id: Some(job.id), task_id: None, + attempt_id: None, worker_id: None, stream: None, }, @@ -240,6 +243,7 @@ impl InMemoryScheduler { message: format!("worker registered: {}", worker.name), job_id: None, task_id: None, + attempt_id: None, worker_id: Some(worker.id), stream: None, }, @@ -258,6 +262,11 @@ impl InMemoryScheduler { } let mut recorded = Vec::new(); for input in batch.entries { + let attempt_id = input.attempt_id.or_else(|| { + input + .task_id + .and_then(|task_id| active_attempt_id(&state, task_id)) + }); let entry = push_log( &mut state, NewLog { @@ -266,6 +275,7 @@ impl InMemoryScheduler { message: input.message, job_id: input.job_id, task_id: input.task_id, + attempt_id, worker_id: Some(worker_id), stream: input.stream, }, @@ -275,6 +285,37 @@ impl InMemoryScheduler { Ok(recorded) } + pub fn list_task_attempt_logs( + &self, + task_id: TaskId, + attempt_id: Uuid, + ) -> Result, FarmError> { + let state = self.inner.lock().map_err(|_| FarmError::LockPoisoned)?; + let (job_id, task_index) = find_task_location(&state, task_id)?; + let task = state + .jobs + .get(&job_id) + .and_then(|job| job.tasks.get(task_index)) + .ok_or(FarmError::TaskNotFound(task_id))?; + if !task + .attempt_records + .iter() + .any(|attempt| attempt.id == attempt_id) + { + return Err(FarmError::AttemptNotFound { + task_id, + attempt_id, + }); + } + Ok(state + .logs + .iter() + .filter(|entry| entry.task_id == Some(task_id)) + .filter(|entry| entry.attempt_id == Some(attempt_id)) + .cloned() + .collect()) + } + pub fn define_limit( &self, definition: ResourceLimitDefinition, @@ -376,6 +417,18 @@ impl InMemoryScheduler { task.state = TaskState::Cancelled; task.updated_at = now; task.completed_at = Some(now); + finish_current_attempt( + task, + AttemptFinish { + state: TaskAttemptState::Cancelled, + completed_at: now, + exit_code: None, + stdout_tail: None, + stderr_tail: None, + artifacts: Vec::new(), + failure_summary: Some("job cancelled".to_string()), + }, + ); if task.lease.is_none() { task.started_at = None; } @@ -443,6 +496,7 @@ impl InMemoryScheduler { .get_mut(&job_id) .ok_or(FarmError::JobNotFound(job_id))?; let lease_token = Uuid::new_v4().to_string(); + let attempt_id = Uuid::new_v4(); let leased_task = { let task = job .tasks @@ -453,9 +507,26 @@ impl InMemoryScheduler { task.state = TaskState::Leased; task.attempts += 1; task.updated_at = now; + let attempt_number = next_attempt_number(task); + task.attempt_records.push(TaskAttempt { + id: attempt_id, + number: attempt_number, + worker_id, + state: TaskAttemptState::Leased, + leased_at: now, + started_at: None, + completed_at: None, + exit_code: None, + stdout_tail: None, + stderr_tail: None, + failure_summary: None, + log_ref: attempt_log_ref(task.id, attempt_id), + artifacts: Vec::new(), + }); task.lease = Some(TaskLeaseInfo { token: lease_token.clone(), worker_id, + attempt_id, leased_at: now, expires_at: now + self.lease_ttl(), }); @@ -471,6 +542,7 @@ impl InMemoryScheduler { message: format!("leased task: {task_name}"), job_id: Some(job_id), task_id: Some(task_id), + attempt_id: Some(attempt_id), worker_id: Some(worker_id), stream: None, }, @@ -495,9 +567,22 @@ impl InMemoryScheduler { .ok_or(FarmError::TaskNotFound(task_id))?; match task.state { TaskState::Pending | TaskState::Leased | TaskState::Running => { + let now = Utc::now(); task.state = TaskState::Cancelled; - task.updated_at = Utc::now(); - task.completed_at = Some(Utc::now()); + task.updated_at = now; + task.completed_at = Some(now); + finish_current_attempt( + task, + AttemptFinish { + state: TaskAttemptState::Cancelled, + completed_at: now, + exit_code: None, + stdout_tail: None, + stderr_tail: None, + artifacts: Vec::new(), + failure_summary: Some("task cancelled".to_string()), + }, + ); if task.lease.is_none() { task.started_at = None; } @@ -564,9 +649,12 @@ impl InMemoryScheduler { .get_mut(task_index) .ok_or(FarmError::TaskNotFound(task_id))?; validate_lease(task, started.worker_id, &started.lease_token)?; + let now = Utc::now(); task.state = TaskState::Running; - task.started_at = Some(Utc::now()); - task.updated_at = Utc::now(); + task.started_at = Some(now); + task.updated_at = now; + start_current_attempt(task, now); + let attempt_id = task.lease.as_ref().map(|lease| lease.attempt_id); let task = task.clone(); update_job_state(job); push_log( @@ -577,6 +665,7 @@ impl InMemoryScheduler { message: format!("task started: {}", task.name), job_id: Some(job_id), task_id: Some(task.id), + attempt_id, worker_id: Some(started.worker_id), stream: None, }, @@ -626,21 +715,36 @@ impl InMemoryScheduler { .ok_or(FarmError::TaskNotFound(task_id))?; validate_lease(task, completion.worker_id, &completion.lease_token)?; let job_was_paused = job.state == JobState::Paused; + let attempt_id = task.lease.as_ref().map(|lease| lease.attempt_id); + let completed_at = Utc::now(); if task.state == TaskState::Cancelled { - task.completed_at = Some(Utc::now()); - task.updated_at = Utc::now(); + task.completed_at = Some(completed_at); + task.updated_at = completed_at; + finish_current_attempt( + task, + AttemptFinish { + state: TaskAttemptState::Cancelled, + completed_at, + exit_code: Some(completion.exit_code), + stdout_tail: completion.stdout_tail.clone(), + stderr_tail: completion.stderr_tail.clone(), + artifacts: completion.artifacts.clone(), + failure_summary: Some( + "task cancelled before completion was accepted".to_string(), + ), + }, + ); task.lease = None; return Ok(task.clone()); } - task.completed_at = Some(Utc::now()); - task.updated_at = Utc::now(); + task.completed_at = Some(completed_at); + task.updated_at = completed_at; task.last_exit_code = Some(completion.exit_code); - task.stdout_tail = completion.stdout_tail; - task.stderr_tail = completion.stderr_tail; - task.artifacts = completion.artifacts; - task.lease = None; + task.stdout_tail = completion.stdout_tail.clone(); + task.stderr_tail = completion.stderr_tail.clone(); + task.artifacts = completion.artifacts.clone(); task.state = if completion.exit_code == 0 { TaskState::Succeeded } else if task.attempts <= task.max_retries { @@ -650,6 +754,27 @@ impl InMemoryScheduler { } else { TaskState::Failed }; + let attempt_state = if completion.exit_code == 0 { + TaskAttemptState::Succeeded + } else { + TaskAttemptState::Failed + }; + finish_current_attempt( + task, + AttemptFinish { + state: attempt_state, + completed_at, + exit_code: Some(completion.exit_code), + stdout_tail: completion.stdout_tail.clone(), + stderr_tail: completion.stderr_tail.clone(), + artifacts: completion.artifacts.clone(), + failure_summary: failure_summary( + completion.exit_code, + completion.stderr_tail.as_deref(), + ), + }, + ); + task.lease = None; let task = task.clone(); update_job_state(job); @@ -672,6 +797,7 @@ impl InMemoryScheduler { ), job_id: Some(job_id), task_id: Some(task.id), + attempt_id, worker_id: Some(completion.worker_id), stream: None, }, @@ -760,6 +886,14 @@ impl SqliteScheduler { self.inner.list_worker_logs(worker_id) } + pub fn list_task_attempt_logs( + &self, + task_id: TaskId, + attempt_id: Uuid, + ) -> Result, FarmError> { + self.inner.list_task_attempt_logs(task_id, attempt_id) + } + pub fn record_worker_logs( &self, worker_id: WorkerId, @@ -970,6 +1104,7 @@ fn task_from_submit( last_exit_code: None, stdout_tail: None, stderr_tail: None, + attempt_records: Vec::new(), }) } @@ -987,6 +1122,75 @@ fn active_worker_slots(state: &SchedulerState, worker_id: WorkerId) -> u32 { .count() as u32 } +fn start_current_attempt(task: &mut Task, started_at: chrono::DateTime) { + let Some(attempt_id) = task.lease.as_ref().map(|lease| lease.attempt_id) else { + return; + }; + if let Some(attempt) = task + .attempt_records + .iter_mut() + .rev() + .find(|attempt| attempt.id == attempt_id) + { + attempt.state = TaskAttemptState::Running; + attempt.started_at = Some(started_at); + } +} + +fn next_attempt_number(task: &Task) -> u32 { + task.attempt_records + .iter() + .map(|attempt| attempt.number) + .max() + .unwrap_or(0) + + 1 +} + +struct AttemptFinish { + state: TaskAttemptState, + completed_at: chrono::DateTime, + exit_code: Option, + stdout_tail: Option, + stderr_tail: Option, + artifacts: Vec, + failure_summary: Option, +} + +fn finish_current_attempt(task: &mut Task, finish: AttemptFinish) { + let Some(attempt_id) = task.lease.as_ref().map(|lease| lease.attempt_id) else { + return; + }; + if let Some(attempt) = task + .attempt_records + .iter_mut() + .rev() + .find(|attempt| attempt.id == attempt_id) + { + attempt.state = finish.state; + attempt.completed_at = Some(finish.completed_at); + attempt.exit_code = finish.exit_code; + attempt.stdout_tail = finish.stdout_tail; + attempt.stderr_tail = finish.stderr_tail; + attempt.failure_summary = finish.failure_summary; + attempt.artifacts = finish.artifacts; + } +} + +fn failure_summary(exit_code: i32, stderr_tail: Option<&str>) -> Option { + if exit_code == 0 { + return None; + } + let detail = stderr_tail + .and_then(|tail| tail.lines().rev().find(|line| !line.trim().is_empty())) + .map(|line| line.trim().chars().take(240).collect::()) + .unwrap_or_else(|| format!("task exited with code {exit_code}")); + Some(detail) +} + +fn attempt_log_ref(task_id: TaskId, attempt_id: Uuid) -> String { + format!("/v1/tasks/{task_id}/attempts/{attempt_id}/logs") +} + fn task_matches_worker( state: &SchedulerState, job_id: JobId, @@ -1185,6 +1389,18 @@ fn expire_old_leases(state: &mut SchedulerState) { .as_ref() .is_some_and(|lease| lease.expires_at < now) { + finish_current_attempt( + task, + AttemptFinish { + state: TaskAttemptState::Expired, + completed_at: now, + exit_code: None, + stdout_tail: None, + stderr_tail: None, + artifacts: Vec::new(), + failure_summary: Some("lease expired before completion".to_string()), + }, + ); task.state = TaskState::Pending; task.lease = None; task.started_at = None; @@ -1204,6 +1420,7 @@ struct NewLog { message: String, job_id: Option, task_id: Option, + attempt_id: Option, worker_id: Option, stream: Option, } @@ -1218,6 +1435,7 @@ fn push_log(state: &mut SchedulerState, log: NewLog) -> FarmLogEntry { message: log.message, job_id: log.job_id, task_id: log.task_id, + attempt_id: log.attempt_id, worker_id: log.worker_id, }; state.logs.push(entry.clone()); @@ -1240,6 +1458,16 @@ fn find_task_location( Err(FarmError::TaskNotFound(task_id)) } +fn active_attempt_id(state: &SchedulerState, task_id: TaskId) -> Option { + state + .jobs + .values() + .flat_map(|job| &job.tasks) + .find(|task| task.id == task_id) + .and_then(|task| task.lease.as_ref()) + .map(|lease| lease.attempt_id) +} + fn validate_lease(task: &Task, worker_id: WorkerId, lease_token: &str) -> Result<(), FarmError> { let Some(lease) = &task.lease else { return Err(FarmError::InvalidLease); @@ -1324,8 +1552,10 @@ fn compute_stats(state: &SchedulerState) -> FarmStats { #[cfg(test)] mod tests { use crate::models::{ - CommandSpec, OpenJdAmountRequirement, OpenJdAttributeRequirement, ResourceLimitDefinition, - TaskLease, TaskLeaseRenewal, TaskRequirements, WorkerCapacity, WorkerId, WorkerInfo, + ArtifactKind, CommandSpec, LogLevel, OpenJdAmountRequirement, OpenJdAttributeRequirement, + ResourceLimitDefinition, TaskArtifact, TaskAttemptState, TaskLease, TaskLeaseRenewal, + TaskRequirements, TaskStarted, WorkerCapacity, WorkerId, WorkerInfo, WorkerLogBatch, + WorkerLogInput, }; use super::*; @@ -1549,6 +1779,137 @@ mod tests { let _ = std::fs::remove_file(database_path); } + #[test] + fn task_attempts_keep_failure_details_and_stable_logs() { + let scheduler = InMemoryScheduler::default(); + let worker = register_worker(&scheduler); + let lease = lease_single_task(&scheduler, worker.id); + let attempt_id = lease.task.lease.as_ref().unwrap().attempt_id; + + scheduler + .mark_task_started( + lease.task.id, + TaskStarted { + worker_id: worker.id, + lease_token: lease.lease_token.clone(), + }, + ) + .expect("task should start"); + scheduler + .record_worker_logs( + worker.id, + WorkerLogBatch { + entries: vec![WorkerLogInput { + level: LogLevel::Error, + stream: Some("stderr".to_string()), + message: "renderer crashed".to_string(), + job_id: Some(lease.task.job_id), + task_id: Some(lease.task.id), + attempt_id: None, + }], + }, + ) + .expect("logs should record"); + scheduler + .complete_task( + lease.task.id, + TaskComplete { + worker_id: worker.id, + lease_token: lease.lease_token, + exit_code: 42, + stdout_tail: Some("started".to_string()), + stderr_tail: Some("renderer crashed".to_string()), + artifacts: vec![TaskArtifact { + name: "failure.log".to_string(), + path: "failure.log".into(), + size_bytes: 12, + kind: ArtifactKind::Log, + modified_at: None, + }], + }, + ) + .expect("task should complete"); + + let task = &scheduler.get_job(lease.task.job_id).unwrap().tasks[0]; + let attempt = task + .attempt_records + .first() + .expect("attempt should persist"); + assert_eq!(attempt.id, attempt_id); + assert_eq!(attempt.state, TaskAttemptState::Failed); + assert_eq!(attempt.exit_code, Some(42)); + assert_eq!(attempt.worker_id, worker.id); + assert_eq!(attempt.failure_summary.as_deref(), Some("renderer crashed")); + assert!(attempt.log_ref.ends_with(&format!("/{attempt_id}/logs"))); + assert_eq!(attempt.artifacts[0].name, "failure.log"); + + let attempt_logs = scheduler + .list_task_attempt_logs(lease.task.id, attempt_id) + .expect("attempt logs should load"); + assert_eq!(attempt_logs.len(), 4); + assert!(attempt_logs + .iter() + .all(|entry| entry.attempt_id == Some(attempt_id))); + } + + #[test] + fn sqlite_scheduler_restores_attempt_history() { + let database_path = temp_database_path(); + let scheduler = SqliteScheduler::open(&database_path).expect("sqlite should open"); + let worker = scheduler + .register_worker(WorkerRegister { + name: "durable-worker".to_string(), + labels: HashMap::new(), + capacity: WorkerCapacity::default(), + }) + .expect("worker should register"); + let job = scheduler + .submit_job(JobSubmit { + name: "attempt-history".to_string(), + priority: 0, + max_retries: 0, + tasks: vec![TaskSubmit { + name: "main".to_string(), + command: command("echo"), + dependencies: vec![], + requirements: Default::default(), + limits: Vec::new(), + max_retries: None, + artifact_paths: Vec::new(), + openjd: None, + }], + openjd: None, + }) + .expect("job should submit"); + let lease = scheduler.lease_task(worker.id).unwrap().unwrap(); + let attempt_id = lease.task.lease.as_ref().unwrap().attempt_id; + scheduler + .complete_task( + lease.task.id, + TaskComplete { + worker_id: worker.id, + lease_token: lease.lease_token, + exit_code: 0, + stdout_tail: Some("done".to_string()), + stderr_tail: None, + artifacts: Vec::new(), + }, + ) + .expect("task should complete"); + + let reopened = SqliteScheduler::open(&database_path).expect("sqlite should reopen"); + let restored = reopened.get_job(job.id).expect("job should be durable"); + let attempt = &restored.tasks[0].attempt_records[0]; + assert_eq!(attempt.id, attempt_id); + assert_eq!(attempt.state, TaskAttemptState::Succeeded); + assert_eq!(attempt.stdout_tail.as_deref(), Some("done")); + assert_eq!( + attempt.log_ref, + format!("/v1/tasks/{}/attempts/{attempt_id}/logs", lease.task.id) + ); + let _ = std::fs::remove_file(database_path); + } + #[test] fn sqlite_scheduler_recovers_expired_leases_after_restart() { let database_path = temp_database_path(); @@ -1996,6 +2357,9 @@ mod tests { TaskState::Pending ); assert_eq!(scheduler.get_job(job.id).unwrap().state, JobState::Queued); + let second_lease = scheduler.lease_task(worker.id).unwrap().unwrap(); + assert_eq!(second_lease.task.attempts, 1); + assert_eq!(second_lease.task.attempt_records[1].number, 2); assert_eq!( scheduler.update_job_priority(job.id, 10).unwrap().priority, 10 diff --git a/crates/farm-worker/src/main.rs b/crates/farm-worker/src/main.rs index 937fe3c..603401c 100644 --- a/crates/farm-worker/src/main.rs +++ b/crates/farm-worker/src/main.rs @@ -70,6 +70,7 @@ impl WorkerLogSink { message, job_id, task_id, + attempt_id: None, }]) .await; } @@ -91,6 +92,7 @@ impl WorkerLogSink { message: line.to_string(), job_id: Some(job_id), task_id: Some(task_id), + attempt_id: None, }) .collect::>(); self.post_batch(entries).await; diff --git a/dashboard/src/components/Common.tsx b/dashboard/src/components/Common.tsx index 3dabe96..cd96d08 100644 --- a/dashboard/src/components/Common.tsx +++ b/dashboard/src/components/Common.tsx @@ -1,9 +1,9 @@ import { AlertCircle } from "lucide-react"; import type { ReactNode } from "react"; import { formatTime } from "../lib/format"; -import type { FarmLog, JobState, TaskState } from "../types"; +import type { FarmLog, JobState, TaskAttemptState, TaskState } from "../types"; -export function StateBadge({ state }: { state: JobState | TaskState }) { +export function StateBadge({ state }: { state: JobState | TaskState | TaskAttemptState }) { return {state}; } diff --git a/dashboard/src/components/Queue.tsx b/dashboard/src/components/Queue.tsx index 43ac54b..e9d1f3a 100644 --- a/dashboard/src/components/Queue.tsx +++ b/dashboard/src/components/Queue.tsx @@ -68,7 +68,7 @@ export function Inspector(props: {
- {(["overview", "workflow", "tasks", "artifacts", "logs"] as const).map((tab) => ( + {(["overview", "workflow", "tasks", "attempts", "artifacts", "logs"] as const).map((tab) => ( @@ -77,6 +77,7 @@ export function Inspector(props: { {props.tab === "overview" ? : null} {props.tab === "workflow" ? : null} {props.tab === "tasks" ? : null} + {props.tab === "attempts" ? : null} {props.tab === "artifacts" ? : null} {props.tab === "logs" ? : null} @@ -119,6 +120,7 @@ function TasksTab({ job, onTaskAction }: { job: ApiJob; onTaskAction: (action: T {upstream.length ? `${upstream.length} upstream` : "root"} {downstream.length ? `${downstream.length} downstream` : "terminal"} {requirementSummary(task)} + {task.attempt_records?.length ?? task.attempts} attempts {task.artifacts?.length ?? 0} artifacts {missingDependencies.length ? missing {missingDependencies.length} : null}
@@ -131,6 +133,33 @@ function TasksTab({ job, onTaskAction }: { job: ApiJob; onTaskAction: (action: T ); } +function AttemptsTab({ job }: { job: ApiJob }) { + const attempts = job.tasks.flatMap((task) => + (task.attempt_records ?? []).map((attempt) => ({ task, attempt })) + ); + if (!attempts.length) { + return } title="No attempts recorded" />; + } + return ( +
+ {attempts.map(({ task, attempt }) => ( +
+
+ {task.name} #{attempt.number} + {attempt.log_ref} +
+ + {attempt.exit_code ?? "-"} + {formatDate(attempt.started_at ?? attempt.leased_at)} + {attempt.completed_at ? formatDate(attempt.completed_at) : "active"} + {attempt.artifacts?.length ?? 0} artifacts + {attempt.failure_summary ? {attempt.failure_summary} : null} +
+ ))} +
+ ); +} + function JobActions({ job, onJobAction }: { job: ApiJob; onJobAction: (action: JobAction, job: ApiJob) => void }) { return (
diff --git a/dashboard/src/data/sampleData.ts b/dashboard/src/data/sampleData.ts index 86b1d6f..06e1835 100644 --- a/dashboard/src/data/sampleData.ts +++ b/dashboard/src/data/sampleData.ts @@ -41,9 +41,9 @@ export const sampleSnapshot: DashboardSnapshot = { ], logs: [ sampleLog("log-01", "info", "controller", "job submitted: Shot_010_Lighting_v003", "job-01"), - sampleLog("log-02", "info", "worker", "starting task Lighting_006", "job-01", "job-01-task-5", "worker-07", "worker"), - sampleLog("log-03", "info", "worker", "[INFO] Rendering tiles on GPU", "job-01", "job-01-task-5", "worker-07", "stdout"), - sampleLog("log-04", "error", "worker", "[ERROR] renderer returned exit code 1", "job-07", "job-07-task-0", "worker-12", "stderr") + sampleLog("log-02", "info", "worker", "starting task Lighting_006", "job-01", "job-01-task-5", "job-01-task-5-attempt-1", "worker-07", "worker"), + sampleLog("log-03", "info", "worker", "[INFO] Rendering tiles on GPU", "job-01", "job-01-task-5", "job-01-task-5-attempt-1", "worker-07", "stdout"), + sampleLog("log-04", "error", "worker", "[ERROR] renderer returned exit code 1", "job-07", "job-07-task-0", "job-07-task-0-attempt-1", "worker-12", "stderr") ], limits: [ { name: "maya", max_count: 8, used: 3, available: 5 }, @@ -90,6 +90,8 @@ function sampleJob( updated_at: "2026-05-27T09:25:10Z", tasks: taskIds.map((taskId, index) => { const done = index < succeeded; + const attemptId = `${taskId}-attempt-1`; + const hasAttempt = done || (state === "failed" && index === succeeded) || (state === "running" && index === succeeded); return { id: taskId, name: `${step}_${String(index + 1).padStart(3, "0")}`, @@ -117,7 +119,41 @@ function sampleJob( } ] : [], - lease: state === "running" && index === succeeded ? { worker_id: "worker-07", leased_at: "2026-05-27T09:18:01Z", expires_at: "2026-05-27T09:28:01Z" } : null, + lease: state === "running" && index === succeeded ? { worker_id: "worker-07", attempt_id: attemptId, leased_at: "2026-05-27T09:18:01Z", expires_at: "2026-05-27T09:28:01Z" } : null, + attempt_records: hasAttempt + ? [ + { + id: attemptId, + number: 1, + worker_id: state === "failed" ? "worker-12" : "worker-07", + state: + state === "failed" && index === succeeded + ? "failed" + : state === "running" && index === succeeded + ? "running" + : "succeeded", + leased_at: "2026-05-27T09:18:01Z", + started_at: "2026-05-27T09:18:04Z", + completed_at: done ? "2026-05-27T09:24:10Z" : null, + exit_code: state === "failed" && index === succeeded ? 1 : done ? 0 : null, + stdout_tail: done ? "RENDERACRE_ARTIFACT=//show/render/frame.png\n[INFO] Render complete" : null, + stderr_tail: state === "failed" && index === succeeded ? "[ERROR] renderer returned exit code 1" : null, + failure_summary: state === "failed" && index === succeeded ? "[ERROR] renderer returned exit code 1" : null, + log_ref: `/v1/tasks/${taskId}/attempts/${attemptId}/logs`, + artifacts: done + ? [ + { + name: `${name}_${1001 + index}.png`, + path: `//show/render/${name}_${1001 + index}.png`, + size_bytes: 182044, + kind: "image", + modified_at: "2026-05-27T09:24:10Z" + } + ] + : [] + } + ] + : [], openjd: { step_name: step, task_parameters: { @@ -144,6 +180,7 @@ function sampleLog( message: string, jobId?: string, taskId?: string, + attemptId?: string, workerId?: string, stream?: string ): FarmLog { @@ -156,6 +193,7 @@ function sampleLog( message, job_id: jobId, task_id: taskId, + attempt_id: attemptId, worker_id: workerId }; } diff --git a/dashboard/src/lib/jobs.ts b/dashboard/src/lib/jobs.ts index f3182f2..908ba2e 100644 --- a/dashboard/src/lib/jobs.ts +++ b/dashboard/src/lib/jobs.ts @@ -79,7 +79,7 @@ export function isView(value: string): value is View { } function isInspectorTab(value: string): value is InspectorTab { - return ["overview", "workflow", "tasks", "artifacts", "logs"].includes(value); + return ["overview", "workflow", "tasks", "attempts", "artifacts", "logs"].includes(value); } export function tabLabel(tab: InspectorTab) { @@ -87,6 +87,7 @@ export function tabLabel(tab: InspectorTab) { overview: "Overview", workflow: "Workflow", tasks: "Tasks", + attempts: "Attempts", artifacts: "Artifacts", logs: "Logs" }[tab]; diff --git a/dashboard/src/types.ts b/dashboard/src/types.ts index e3829e8..6aac1ea 100644 --- a/dashboard/src/types.ts +++ b/dashboard/src/types.ts @@ -2,9 +2,10 @@ import type { ReactNode } from "react"; import type { Node } from "@xyflow/react"; export type View = "queue" | "workers" | "openjd" | "logs" | "settings"; -export type InspectorTab = "overview" | "workflow" | "tasks" | "artifacts" | "logs"; +export type InspectorTab = "overview" | "workflow" | "tasks" | "attempts" | "artifacts" | "logs"; export type JobState = "queued" | "running" | "paused" | "succeeded" | "failed" | "cancelled"; export type TaskState = "pending" | "leased" | "running" | "succeeded" | "failed" | "cancelled"; +export type TaskAttemptState = "leased" | "running" | "succeeded" | "failed" | "cancelled" | "expired"; export type JobAction = "pause" | "resume" | "cancel" | "priority"; export type TaskAction = "cancel" | "requeue"; export type LogLevel = "debug" | "info" | "warn" | "error"; @@ -18,6 +19,22 @@ export interface ApiArtifact { modified_at?: string | null; } +export interface ApiTaskAttempt { + id: string; + number: number; + worker_id: string; + state: TaskAttemptState; + leased_at: string; + started_at?: string | null; + completed_at?: string | null; + exit_code?: number | null; + stdout_tail?: string | null; + stderr_tail?: string | null; + failure_summary?: string | null; + log_ref: string; + artifacts?: ApiArtifact[]; +} + export interface ApiAmountRequirement { name: string; min?: number | null; @@ -50,9 +67,11 @@ export interface ApiTask { artifacts?: ApiArtifact[]; lease?: { worker_id: string; + attempt_id?: string; leased_at: string; expires_at: string; } | null; + attempt_records?: ApiTaskAttempt[]; openjd?: { step_name: string; task_parameters?: Record; @@ -91,6 +110,7 @@ export interface FarmLog { message: string; job_id?: string | null; task_id?: string | null; + attempt_id?: string | null; worker_id?: string | null; } diff --git a/docs/architecture.md b/docs/architecture.md index b089803..c0d48d8 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -34,6 +34,8 @@ Core endpoints: - `POST /v1/tasks/{task_id}/started`: mark a leased task as running. - `POST /v1/tasks/{task_id}/renew`: extend a healthy in-flight task lease. - `POST /v1/tasks/{task_id}/complete`: report result and trigger retry/final state. +- `GET /v1/tasks/{task_id}/attempts/{attempt_id}/logs`: read logs for one durable attempt record. +- `GET /v1/tasks/{task_id}/artifacts/{artifact_index}`: download a captured task artifact. Task leases have a configurable controller-side TTL. Workers renew leases while direct command and OpenJD tasks are still executing, so long-running work is not @@ -103,9 +105,25 @@ SQLite persists submitted jobs, task attempts, leases, worker registrations, and dashboard history needed for normal queue recovery. Expired leases are recovered when workers resume leasing after a controller restart. +## Logs and artifacts + +The scheduler owns durable attempt metadata: each leased execution receives a +stable attempt id, worker identity, timing, exit code, concise failure summary, +stdout/stderr tails, artifact metadata, and a `log_ref` that resolves through +the controller API. Worker output is accepted as structured log batches and +attached to the active attempt when a task id is present, so running tasks can +show near-live stdout/stderr while completed attempts keep their own record. + +The first artifact implementation records filesystem paths and serves readable +files through the controller. Workers can discover artifacts from submitted +artifact paths, modified files under output directories, and +`RENDERACRE_ARTIFACT=...` stdout/stderr directives. Future object storage +backends should preserve the same task artifact contract while replacing only +the path resolution and byte-serving implementation. + ## Dashboard path -`dashboard/` is a Vite React application for queue operations. It reads `/v1/dashboard`, `/v1/jobs`, `/v1/workers`, and `/v1/stats`, then renders a Deadline-style queue table, worker assignment panel, OpenJD step detail, dependency mini graph, and stdout tail. During local development Vite proxies `/v1` to the controller; in deployment, serve the built static assets from `dashboard/dist` beside the controller API. +`dashboard/` is a Vite React application for queue operations. It reads `/v1/dashboard`, `/v1/jobs`, `/v1/workers`, and `/v1/stats`, then renders a Deadline-style queue table, worker assignment panel, OpenJD step detail, dependency mini graph, attempt history, artifacts, and stdout/stderr tails. During local development Vite proxies `/v1` to the controller; in deployment, serve the built static assets from `dashboard/dist` beside the controller API. ## Release assets