diff --git a/README.md b/README.md index dd761ea..2f661df 100644 --- a/README.md +++ b/README.md @@ -151,6 +151,22 @@ Workers advertise matching capabilities with labels such as also register default OpenJD capabilities for OS family, CPU architecture, and slot count, so OpenJD `hostRequirements` work out of the box for common cases. +Operators can define shared limits for scarce licenses or farm-wide resources, +then tasks can declare the named limits they need: + +```text +POST /v1/limits +GET /v1/limits +``` + +```json +{ + "name": "maya-render", + "command": { "executable": "mayapy", "args": ["render.py"] }, + "limits": ["maya"] +} +``` + Common queue operations are available as stable action endpoints: ```text diff --git a/crates/farm-controller/src/main.rs b/crates/farm-controller/src/main.rs index b504d7d..d991908 100644 --- a/crates/farm-controller/src/main.rs +++ b/crates/farm-controller/src/main.rs @@ -10,8 +10,9 @@ use axum::{Json, Router}; use clap::{Parser, ValueEnum}; use farm_core::{ DashboardSnapshot, FarmError, FarmLogEntry, FarmStats, InMemoryScheduler, Job, JobId, - JobPriorityUpdate, JobSubmit, SchedulerConfig, SqliteScheduler, Task, TaskComplete, TaskId, - TaskLease, TaskLeaseRenewal, TaskStarted, WorkerId, WorkerInfo, WorkerLogBatch, WorkerRegister, + JobPriorityUpdate, JobSubmit, ResourceLimitDefinition, ResourceLimitSnapshot, SchedulerConfig, + SqliteScheduler, Task, TaskComplete, TaskId, TaskLease, TaskLeaseRenewal, TaskStarted, + WorkerId, WorkerInfo, WorkerLogBatch, WorkerRegister, }; use serde_json::json; use tower_http::cors::CorsLayer; @@ -102,6 +103,23 @@ impl AppScheduler { } } + fn define_limit( + &self, + definition: ResourceLimitDefinition, + ) -> Result { + match self { + Self::Memory(scheduler) => scheduler.define_limit(definition), + Self::Sqlite(scheduler) => scheduler.define_limit(definition), + } + } + + fn list_limits(&self) -> Result, FarmError> { + match self { + Self::Memory(scheduler) => scheduler.list_limits(), + Self::Sqlite(scheduler) => scheduler.list_limits(), + } + } + fn pause_job(&self, job_id: JobId) -> Result { match self { Self::Memory(scheduler) => scheduler.pause_job(job_id), @@ -234,6 +252,7 @@ fn app(scheduler: AppScheduler) -> Router { .route("/v1/dashboard", get(get_dashboard)) .route("/v1/logs", get(list_logs)) .route("/v1/stats", get(get_stats)) + .route("/v1/limits", get(list_limits).post(define_limit)) .route("/v1/jobs", get(list_jobs).post(submit_job)) .route("/v1/jobs/{job_id}", get(get_job)) .route("/v1/jobs/{job_id}/pause", post(pause_job)) @@ -337,6 +356,19 @@ async fn list_logs( Ok(Json(scheduler.list_logs()?)) } +async fn define_limit( + State(scheduler): State, + Json(definition): Json, +) -> Result, ApiError> { + Ok(Json(scheduler.define_limit(definition)?)) +} + +async fn list_limits( + State(scheduler): State, +) -> Result>, ApiError> { + Ok(Json(scheduler.list_limits()?)) +} + async fn register_worker( State(scheduler): State, Json(registration): Json, diff --git a/crates/farm-core/src/models.rs b/crates/farm-core/src/models.rs index 30323b6..0cbfd20 100644 --- a/crates/farm-core/src/models.rs +++ b/crates/farm-core/src/models.rs @@ -37,6 +37,8 @@ pub struct TaskSubmit { #[serde(default)] pub requirements: TaskRequirements, #[serde(default)] + pub limits: Vec, + #[serde(default)] pub max_retries: Option, #[serde(default)] pub artifact_paths: Vec, @@ -431,6 +433,8 @@ pub struct DashboardSnapshot { pub jobs: Vec, pub workers: Vec, pub logs: Vec, + #[serde(default)] + pub limits: Vec, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -468,6 +472,8 @@ pub struct Task { pub dependencies: Vec, #[serde(default)] pub requirements: TaskRequirements, + #[serde(default)] + pub limits: Vec, pub attempts: u32, pub max_retries: u32, pub lease: Option, @@ -610,6 +616,20 @@ pub enum LogSource { Task, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResourceLimitDefinition { + pub name: String, + pub max_count: u32, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResourceLimitSnapshot { + pub name: String, + pub max_count: u32, + pub used: u32, + pub available: u32, +} + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] pub enum WorkerState { diff --git a/crates/farm-core/src/openjd.rs b/crates/farm-core/src/openjd.rs index 8ff26dd..d0c37a3 100644 --- a/crates/farm-core/src/openjd.rs +++ b/crates/farm-core/src/openjd.rs @@ -216,6 +216,7 @@ fn expand_openjd_tasks( .as_ref() .map(openjd_host_requirements) .unwrap_or_default(), + limits: Vec::new(), max_retries: None, artifact_paths: infer_artifact_paths(&job_parameters), openjd: Some(openjd), diff --git a/crates/farm-core/src/scheduler.rs b/crates/farm-core/src/scheduler.rs index c88139d..c09aaf2 100644 --- a/crates/farm-core/src/scheduler.rs +++ b/crates/farm-core/src/scheduler.rs @@ -10,9 +10,10 @@ use uuid::Uuid; use crate::models::{ DashboardSnapshot, FarmLogEntry, FarmStats, Job, JobId, JobState, JobSubmit, LogLevel, - LogSource, OpenJdAmountRequirement, OpenJdAttributeRequirement, Task, TaskArtifact, - TaskComplete, TaskId, TaskLease, TaskLeaseInfo, TaskLeaseRenewal, TaskStarted, TaskState, - TaskSubmit, WorkerCapacity, WorkerId, WorkerInfo, WorkerLogBatch, WorkerRegister, WorkerState, + LogSource, OpenJdAmountRequirement, OpenJdAttributeRequirement, ResourceLimitDefinition, + ResourceLimitSnapshot, Task, TaskArtifact, TaskComplete, TaskId, TaskLease, TaskLeaseInfo, + TaskLeaseRenewal, TaskStarted, TaskState, TaskSubmit, WorkerCapacity, WorkerId, WorkerInfo, + WorkerLogBatch, WorkerRegister, WorkerState, }; use crate::openjd::openjd_to_tasks; @@ -65,6 +66,7 @@ struct SchedulerState { jobs: HashMap, workers: HashMap, logs: Vec, + limits: HashMap, } const MAX_LOG_ENTRIES: usize = 500; @@ -195,6 +197,7 @@ impl InMemoryScheduler { jobs, workers, logs: state.logs.clone(), + limits: limit_snapshots(&state), }) } @@ -272,6 +275,29 @@ impl InMemoryScheduler { Ok(recorded) } + pub fn define_limit( + &self, + definition: ResourceLimitDefinition, + ) -> Result { + if definition.name.trim().is_empty() || definition.max_count == 0 { + return Err(FarmError::InvalidSubmission( + "resource limits require a non-empty name and max_count greater than zero" + .to_string(), + )); + } + + let mut state = self.inner.lock().map_err(|_| FarmError::LockPoisoned)?; + state + .limits + .insert(definition.name.clone(), definition.clone()); + Ok(limit_snapshot(&state, &definition)) + } + + pub fn list_limits(&self) -> Result, FarmError> { + let state = self.inner.lock().map_err(|_| FarmError::LockPoisoned)?; + Ok(limit_snapshots(&state)) + } + pub fn heartbeat_worker(&self, worker_id: WorkerId) -> Result { let mut state = self.inner.lock().map_err(|_| FarmError::LockPoisoned)?; let worker = state @@ -405,6 +431,7 @@ impl InMemoryScheduler { .filter(|(job_id, task_id, _, _)| { task_matches_worker(&state, *job_id, *task_id, &worker) }) + .filter(|(job_id, task_id, _, _)| limits_are_available(&state, *job_id, *task_id)) .max_by_key(|(_, _, priority, created_at)| (*priority, std::cmp::Reverse(*created_at))); let Some((job_id, task_id, _, _)) = selected else { @@ -741,6 +768,17 @@ impl SqliteScheduler { self.write_durably(|inner| inner.record_worker_logs(worker_id, batch)) } + pub fn define_limit( + &self, + definition: ResourceLimitDefinition, + ) -> Result { + self.write_durably(|inner| inner.define_limit(definition)) + } + + pub fn list_limits(&self) -> Result, FarmError> { + self.inner.list_limits() + } + pub fn pause_job(&self, job_id: JobId) -> Result { self.write_durably(|inner| inner.pause_job(job_id)) } @@ -919,6 +957,7 @@ fn task_from_submit( openjd: submission.openjd, dependencies, requirements: submission.requirements, + limits: submission.limits, attempts: 0, max_retries: submission.max_retries.unwrap_or(job_max_retries), lease: None, @@ -1072,6 +1111,55 @@ fn worker_label_value<'a>(worker: &'a WorkerInfo, name: &str) -> Option<&'a str> }) } +fn limits_are_available(state: &SchedulerState, job_id: JobId, task_id: TaskId) -> bool { + let Some(job) = state.jobs.get(&job_id) else { + return false; + }; + let Some(task) = job.tasks.iter().find(|task| task.id == task_id) else { + return false; + }; + + task.limits.iter().all(|limit_name| { + state + .limits + .get(limit_name) + .is_some_and(|definition| active_limit_usage(state, limit_name) < definition.max_count) + }) +} + +fn active_limit_usage(state: &SchedulerState, limit_name: &str) -> u32 { + state + .jobs + .values() + .flat_map(|job| &job.tasks) + .filter(|task| matches!(task.state, TaskState::Leased | TaskState::Running)) + .filter(|task| task.limits.iter().any(|candidate| candidate == limit_name)) + .count() as u32 +} + +fn limit_snapshots(state: &SchedulerState) -> Vec { + let mut snapshots = state + .limits + .values() + .map(|definition| limit_snapshot(state, definition)) + .collect::>(); + snapshots.sort_by(|left, right| left.name.cmp(&right.name)); + snapshots +} + +fn limit_snapshot( + state: &SchedulerState, + definition: &ResourceLimitDefinition, +) -> ResourceLimitSnapshot { + let used = active_limit_usage(state, &definition.name); + ResourceLimitSnapshot { + name: definition.name.clone(), + max_count: definition.max_count, + used, + available: definition.max_count.saturating_sub(used), + } +} + fn dependencies_are_satisfied(state: &SchedulerState, job_id: JobId, task_id: TaskId) -> bool { let Some(job) = state.jobs.get(&job_id) else { return false; @@ -1236,8 +1324,8 @@ fn compute_stats(state: &SchedulerState) -> FarmStats { #[cfg(test)] mod tests { use crate::models::{ - CommandSpec, OpenJdAmountRequirement, OpenJdAttributeRequirement, TaskLease, - TaskLeaseRenewal, TaskRequirements, WorkerCapacity, WorkerId, WorkerInfo, + CommandSpec, OpenJdAmountRequirement, OpenJdAttributeRequirement, ResourceLimitDefinition, + TaskLease, TaskLeaseRenewal, TaskRequirements, WorkerCapacity, WorkerId, WorkerInfo, }; use super::*; @@ -1257,6 +1345,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1266,6 +1355,7 @@ mod tests { command: command("echo"), dependencies: vec!["prepare".to_string()], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1320,6 +1410,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1415,6 +1506,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1471,6 +1563,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1527,6 +1620,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1568,6 +1662,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1577,6 +1672,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1643,6 +1739,7 @@ mod tests { pools: vec!["lighting".to_string()], ..Default::default() }, + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1656,6 +1753,7 @@ mod tests { pools: vec!["sim".to_string()], ..Default::default() }, + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1696,6 +1794,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1740,6 +1839,7 @@ mod tests { ..Default::default() }, max_retries: None, + limits: Vec::new(), artifact_paths: Vec::new(), openjd: None, }], @@ -1786,6 +1886,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1823,6 +1924,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1858,6 +1960,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1898,6 +2001,111 @@ mod tests { 10 ); } + + #[test] + fn resource_limits_block_until_usage_is_released() { + let scheduler = InMemoryScheduler::default(); + scheduler + .define_limit(ResourceLimitDefinition { + name: "maya".to_string(), + max_count: 1, + }) + .expect("limit should be defined"); + scheduler + .submit_job(JobSubmit { + name: "limited".to_string(), + priority: 0, + max_retries: 0, + tasks: vec![ + limited_task("first", "maya"), + limited_task("second", "maya"), + ], + openjd: None, + }) + .expect("job should submit"); + let worker = register_worker(&scheduler); + + let first = scheduler.lease_task(worker.id).unwrap().unwrap(); + assert!(scheduler.lease_task(worker.id).unwrap().is_none()); + let snapshot = scheduler.list_limits().unwrap(); + assert_eq!(snapshot[0].used, 1); + assert_eq!(snapshot[0].available, 0); + + scheduler + .complete_task( + first.task.id, + TaskComplete { + worker_id: worker.id, + lease_token: first.lease_token, + exit_code: 0, + stdout_tail: None, + stderr_tail: None, + artifacts: Vec::new(), + }, + ) + .expect("completion should release limit"); + + let second = scheduler.lease_task(worker.id).unwrap().unwrap(); + assert_ne!(first.task.id, second.task.id); + } + + #[test] + fn expired_leases_release_limit_usage_for_recovery() { + let scheduler = InMemoryScheduler::default(); + scheduler + .define_limit(ResourceLimitDefinition { + name: "gpu".to_string(), + max_count: 1, + }) + .expect("limit should be defined"); + let job = scheduler + .submit_job(JobSubmit { + name: "gpu-job".to_string(), + priority: 0, + max_retries: 0, + tasks: vec![limited_task("main", "gpu")], + openjd: None, + }) + .expect("job should submit"); + let worker = register_worker(&scheduler); + let lease = scheduler.lease_task(worker.id).unwrap().unwrap(); + + { + let mut state = scheduler.inner.lock().unwrap(); + let task = state + .jobs + .get_mut(&job.id) + .unwrap() + .tasks + .iter_mut() + .find(|task| task.id == lease.task.id) + .unwrap(); + task.state = TaskState::Running; + task.lease.as_mut().unwrap().expires_at = Utc::now() - Duration::seconds(1); + } + + let recovered = scheduler.lease_task(worker.id).unwrap().unwrap(); + assert_eq!(recovered.task.id, lease.task.id); + assert_eq!(scheduler.list_limits().unwrap()[0].used, 1); + } + + #[test] + fn undefined_limit_blocks_task() { + let scheduler = InMemoryScheduler::default(); + scheduler + .submit_job(JobSubmit { + name: "missing-limit".to_string(), + priority: 0, + max_retries: 0, + tasks: vec![limited_task("main", "houdini")], + openjd: None, + }) + .expect("job should submit"); + let worker = register_worker(&scheduler); + + assert!(scheduler.lease_task(worker.id).unwrap().is_none()); + } + fn command(executable: &str) -> CommandSpec { CommandSpec { executable: executable.to_string(), @@ -1929,6 +2137,7 @@ mod tests { command: command("echo"), dependencies: vec![], requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, @@ -1942,6 +2151,19 @@ mod tests { .expect("a task should be leased") } + fn limited_task(name: &str, limit: &str) -> TaskSubmit { + TaskSubmit { + name: name.to_string(), + command: command("echo"), + dependencies: Vec::new(), + requirements: Default::default(), + limits: vec![limit.to_string()], + max_retries: None, + artifact_paths: Vec::new(), + openjd: None, + } + } + fn temp_database_path() -> std::path::PathBuf { std::env::temp_dir().join(format!("renderacre-{}.sqlite3", Uuid::new_v4())) } diff --git a/crates/farm-python/src/lib.rs b/crates/farm-python/src/lib.rs index febcab7..0d95f93 100644 --- a/crates/farm-python/src/lib.rs +++ b/crates/farm-python/src/lib.rs @@ -23,6 +23,7 @@ fn command_job(name: String, executable: String, args: Option>) -> P }, dependencies: Vec::new(), requirements: Default::default(), + limits: Vec::new(), max_retries: None, artifact_paths: Vec::new(), openjd: None, diff --git a/dashboard/src/App.tsx b/dashboard/src/App.tsx index 8938501..ae3d7f2 100644 --- a/dashboard/src/App.tsx +++ b/dashboard/src/App.tsx @@ -145,7 +145,7 @@ export function App() { ) : null} {activeView === "logs" ? : null} {activeView === "openjd" ? : null} - {activeView === "settings" ? : null} + {activeView === "settings" ? : null} ); diff --git a/dashboard/src/components/SecondaryViews.tsx b/dashboard/src/components/SecondaryViews.tsx index c1b8fcc..00f4168 100644 --- a/dashboard/src/components/SecondaryViews.tsx +++ b/dashboard/src/components/SecondaryViews.tsx @@ -1,6 +1,6 @@ import { Activity, ListChecks, Server } from "lucide-react"; import { inferApplication, inferStep, progress } from "../lib/jobs"; -import type { ApiJob, ApiStats } from "../types"; +import type { ApiJob, ApiLimit, ApiStats } from "../types"; import { Progress, StateBadge } from "./Common"; export function OpenJdView({ jobs }: { jobs: ApiJob[] }) { @@ -26,7 +26,7 @@ export function OpenJdView({ jobs }: { jobs: ApiJob[] }) { ); } -export function SettingsView({ source, stats }: { source: "live" | "sample"; stats: ApiStats }) { +export function SettingsView({ limits, source, stats }: { limits: ApiLimit[]; source: "live" | "sample"; stats: ApiStats }) { return (
@@ -34,6 +34,21 @@ export function SettingsView({ source, stats }: { source: "live" | "sample"; sta
Workers{stats.workers_online}/{stats.workers_total} online
Tasks{stats.tasks_running + stats.tasks_leased} active
+
+

Shared Limits

{limits.length} limits
+
+ {limits.map((limit) => ( +
+
+ {limit.name} + {limit.used}/{limit.max_count} in use +
+ + {limit.available} available +
+ ))} +
+
); } diff --git a/dashboard/src/data/sampleData.ts b/dashboard/src/data/sampleData.ts index 47ba938..86b1d6f 100644 --- a/dashboard/src/data/sampleData.ts +++ b/dashboard/src/data/sampleData.ts @@ -45,6 +45,10 @@ export const sampleSnapshot: DashboardSnapshot = { 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") ], + limits: [ + { name: "maya", max_count: 8, used: 3, available: 5 }, + { name: "gpu", max_count: 24, used: 11, available: 13 } + ], jobs: [ sampleJob("job-01", "Shot_010_Lighting_v003", "running", 75, "Blender 4.1", "Lighting", 78), sampleJob("job-02", "Shot_010_Comp_v002", "running", 50, "Nuke 15.2", "Comp", 42), diff --git a/dashboard/src/lib/jobs.ts b/dashboard/src/lib/jobs.ts index ef68e43..f3182f2 100644 --- a/dashboard/src/lib/jobs.ts +++ b/dashboard/src/lib/jobs.ts @@ -69,7 +69,8 @@ export function normalizeSnapshot(data: Partial): DashboardSn stats: data.stats ?? sampleSnapshot.stats, jobs: data.jobs ?? [], workers: data.workers ?? [], - logs: data.logs ?? [] + logs: data.logs ?? [], + limits: data.limits ?? [] }; } diff --git a/dashboard/src/types.ts b/dashboard/src/types.ts index 35c5f4f..e3829e8 100644 --- a/dashboard/src/types.ts +++ b/dashboard/src/types.ts @@ -94,6 +94,13 @@ export interface FarmLog { worker_id?: string | null; } +export interface ApiLimit { + name: string; + max_count: number; + used: number; + available: number; +} + export interface ApiStats { jobs_total: number; jobs_queued: number; @@ -118,6 +125,7 @@ export interface DashboardSnapshot { jobs: ApiJob[]; workers: ApiWorker[]; logs: FarmLog[]; + limits: ApiLimit[]; } export interface WorkflowNode { diff --git a/docs/architecture.md b/docs/architecture.md index 929b501..b089803 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,6 +29,8 @@ Core endpoints: - `POST /v1/jobs/{job_id}/priority`: update queue priority. - `POST /v1/tasks/{task_id}/cancel`: cancel a selected task. - `POST /v1/tasks/{task_id}/requeue`: retry or requeue a selected task. +- `GET /v1/limits`: inspect shared resource and license usage. +- `POST /v1/limits`: define a named shared resource limit. - `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. @@ -53,6 +55,13 @@ standard attribute capabilities are read from worker labels. Lifecycle actions are idempotent where repeating the same request is safe, and invalid state transitions return a conflict response with a descriptive error. +Shared resource limits are named scheduler constraints for scarce farm-wide +capacity such as floating DCC licenses, GPU partitions, or heavy caches. Tasks +can request limits by name; undefined or exhausted limits block otherwise +runnable tasks until usage is released by completion, cancellation, requeue, or +lease expiry. Limit snapshots are exposed in the dashboard payload and the +dedicated `/v1/limits` API. + ## OpenJD path The controller accepts an OpenJD template bundle under `openjd.template_yaml`. It uses the official OpenJD Rust crates instead of a hand-written parser: