Resource-aware executor scheduling

Packages/ExecutorApp/Package.swift

@@ -40,6 +40,12 @@ let package = Package(
             .product(name: "FlyingFox", package: "FlyingFox")
         ], exclude: [
             "Resources/openapi.yaml"
+        ]),
+        .testTarget(name: "ExecutorAppTests", dependencies: [
+            "ExecutorApp",
+            .product(name: "TartCommon", package: "TartCommon"),
+            .product(name: "LoggingDomain", package: "Logging"),
+            .product(name: "VirtualMachineDomain", package: "VirtualMachine")
         ])
     ]
 )

Packages/ExecutorApp/Sources/ExecutorApp/ExecutorServer.swift

@@ -75,7 +75,9 @@ final class ExecutorServer {
             routerUrl: settings.routerUrl,
             virtualMachineProvider: virtualMachineProvider,
             logger: logger,
-            numberOfMachines: settings.numberOfMachines
+            numberOfMachines: settings.numberOfMachines,
+            cpuLimit: settings.cpuLimit,
+            totalMemory: settings.totalMemory
         )
 
         Task {
@@ -181,7 +183,9 @@ tart_executor_memory_used\(labels) \(jobStatus.memoryUsed)
                 cpuLimit: settings.cpuLimit,
                 cpuUsed: jobStatus.cpuUsed,
                 totalMemory: settings.totalMemory,
-                memoryUsed: jobStatus.memoryUsed
+                memoryUsed: jobStatus.memoryUsed,
+                defaultCpu: settings.defaultCpu,
+                defaultMemory: settings.defaultMemory
             )
 
             let body = try encoder.encode(status)

Packages/ExecutorApp/Sources/ExecutorApp/Internal/ExecutorAPIRoutes.swift

@@ -39,6 +39,8 @@ extension ExecutorServer {
                 cpuUsed: jobStatus.cpuUsed,
                 totalMemory: settings.totalMemory,
                 memoryUsed: jobStatus.memoryUsed,
+                defaultCpu: settings.defaultCpu,
+                defaultMemory: settings.defaultMemory,
                 diskTotalBytes: disk?.totalBytes,
                 diskFreeBytes: disk?.freeBytes,
                 diskUsedBytes: disk?.usedBytes

Packages/ExecutorApp/Sources/ExecutorApp/Internal/ExecutorJobHandler+Capacity.swift

@@ -0,0 +1,54 @@
+import TartCommon
+
+// Shared by webhook handling and the management dispatch path so every non-forced VM start
+// observes the same VM-count and aggregate-resource limits.
+extension ExecutorJobHandler {
+    var hasVirtualMachineSlot: Bool {
+        activeJobs.count < numberOfMachines
+    }
+
+    var resourceCapacity: ResourceCapacity {
+        let status = jobStatus
+        return ResourceCapacity(
+            cpuLimit: cpuLimit,
+            cpuUsed: status.cpuUsed,
+            memoryLimit: totalMemory,
+            memoryUsed: status.memoryUsed
+        )
+    }
+
+    func hasResources(for pendingJob: ExecutorPendingJob) -> Bool {
+        resourceCapacity.canFit(.init(cpu: pendingJob.cpu, memory: pendingJob.memory))
+    }
+
+    func startIfCapacityAllows(_ pendingJob: ExecutorPendingJob) {
+        guard hasVirtualMachineSlot else {
+            logVirtualMachineLimitReached(pendingJob)
+            return
+        }
+        guard hasResources(for: pendingJob) else {
+            logResourceCapacityReached(pendingJob)
+            return
+        }
+        start(pendingJob: pendingJob)
+    }
+
+    func logVirtualMachineLimitReached(_ pendingJob: ExecutorPendingJob) {
+        logger.info("Job Executor Virtual Machine Limit Reached", pendingJob: pendingJob, [
+            LogParameterKey.activeVirtualMachines: "\(activeJobs.count)",
+            LogParameterKey.virtualMachineLimit: "\(numberOfMachines)"
+        ])
+    }
+
+    func logResourceCapacityReached(_ pendingJob: ExecutorPendingJob) {
+        let status = jobStatus
+        logger.info("Job Executor Aggregate Resource Capacity Reached", pendingJob: pendingJob, [
+            LogParameterKey.cpu: pendingJob.cpu.map(String.init) ?? "unknown",
+            LogParameterKey.cpuUsed: "\(status.cpuUsed)",
+            LogParameterKey.cpuLimit: "\(cpuLimit)",
+            LogParameterKey.memory: pendingJob.memory.map(String.init) ?? "unknown",
+            LogParameterKey.memoryUsed: "\(status.memoryUsed)",
+            LogParameterKey.memoryLimit: "\(totalMemory)"
+        ])
+    }
+}

Packages/ExecutorApp/Sources/ExecutorApp/Internal/ExecutorJobHandler+Dispatch.swift

@@ -10,6 +10,7 @@ enum ExecutorDispatchOutcome: Sendable {
     case skippedVirtualMachineExists(vmName: String)
     case skippedAlreadyTracked(state: ExecutorJobState)
     case skippedAtCapacity(active: Int, limit: Int)
+    case skippedInsufficientResources(cpuUsed: Int, cpuLimit: Int, memoryUsed: Int, memoryLimit: Int)
 }
 
 extension ExecutorJobHandler {
@@ -47,11 +48,29 @@ extension ExecutorJobHandler {
                 ])
                 return .skippedAtCapacity(active: activeJobs.count, limit: numberOfMachines)
             }
+            guard hasResources(for: pendingJob) else {
+                let status = jobStatus
+                logResourceCapacityReached(pendingJob)
+                return .skippedInsufficientResources(
+                    cpuUsed: status.cpuUsed,
+                    cpuLimit: cpuLimit,
+                    memoryUsed: status.memoryUsed,
+                    memoryLimit: totalMemory
+                )
+            }
         } else if activeJobs.count >= numberOfMachines {
             logger.error("Job Dispatch Forced Over Capacity", pendingJob: pendingJob, [
                 LogParameterKey.activeJobs: "\(activeJobs.count)",
                 LogParameterKey.maxMachines: "\(numberOfMachines)"
             ])
+        } else if !hasResources(for: pendingJob) {
+            let status = jobStatus
+            logger.error("Job Dispatch Forced Over Aggregate Resource Capacity", pendingJob: pendingJob, [
+                LogParameterKey.cpuUsed: "\(status.cpuUsed)",
+                LogParameterKey.cpuLimit: "\(cpuLimit)",
+                LogParameterKey.memoryUsed: "\(status.memoryUsed)",
+                LogParameterKey.memoryLimit: "\(totalMemory)"
+            ])
         }
 
         // Track it as pending so the webhooks that follow behave: a real "queued" delivery is

Packages/ExecutorApp/Sources/ExecutorApp/Internal/ExecutorJobHandler.swift

@@ -17,6 +17,8 @@ actor ExecutorJobHandler {
     // Module-internal rather than private so `ExecutorJobHandler+Dispatch` can reach them; the
     // actor itself is internal, so this widens nothing outside ExecutorApp.
     let numberOfMachines: Int
+    let cpuLimit: Int
+    let totalMemory: Int
     var activeJobs = [UUID: ActiveJob]()
     var inProgressJobs = [Int: ExecutorPendingJob]()
     var pendingJobs = [Int: ExecutorPendingJob]()
@@ -35,11 +37,20 @@ actor ExecutorJobHandler {
         )
     }
 
-    init(routerUrl: String?, virtualMachineProvider: VirtualMachineProvider, logger: Logger, numberOfMachines: Int) {
+    init(
+        routerUrl: String?,
+        virtualMachineProvider: VirtualMachineProvider,
+        logger: Logger,
+        numberOfMachines: Int,
+        cpuLimit: Int,
+        totalMemory: Int
+    ) {
         self.routerUrl = routerUrl
         self.virtualMachineProvider = virtualMachineProvider
         self.logger = logger
         self.numberOfMachines = numberOfMachines
+        self.cpuLimit = cpuLimit
+        self.totalMemory = totalMemory
 
         Task {
            await activeJobEnded()
@@ -163,20 +174,24 @@ actor ExecutorJobHandler {
 private extension ExecutorJobHandler {
     func handleRouterStart(pendingJob: ExecutorPendingJob) -> Bool {
         let isAlreadyRunning = activeJobs.values.contains { $0.jobId == pendingJob.id }
-        let willStart = !isAlreadyRunning && activeJobs.count < numberOfMachines
-        logger.info("Job Handle Router Started", pendingJob: pendingJob, [
-            LogParameterKey.willStart: "\(willStart)",
-            LogParameterKey.activeJobs: "\(activeJobs.count)",
-            LogParameterKey.maxMachines: "\(numberOfMachines)"
-        ])
         if isAlreadyRunning {
             // The router sent a job this executor already has a virtual machine for. Report
             // success so the router does not send it elsewhere as well.
             return true
         }
-        guard willStart else {
+        guard hasVirtualMachineSlot else {
+            logVirtualMachineLimitReached(pendingJob)
+            return false
+        }
+        guard hasResources(for: pendingJob) else {
+            logResourceCapacityReached(pendingJob)
             return false
         }
+        logger.info("Job Handle Router Started", pendingJob: pendingJob, [
+            LogParameterKey.willStart: "true",
+            LogParameterKey.activeJobs: "\(activeJobs.count)",
+            LogParameterKey.maxMachines: "\(numberOfMachines)"
+        ])
         start(pendingJob: pendingJob)
         return true
     }
@@ -191,9 +206,7 @@ private extension ExecutorJobHandler {
         logger.info("Job Handle Pending", pendingJob: pendingJob)
         pendingJobs[pendingJob.id] = pendingJob
 
-        if activeJobs.count < numberOfMachines {
-            start(pendingJob: pendingJob)
-        }
+        startIfCapacityAllows(pendingJob)
     }
 
     func handleInProgress(pendingJob: ExecutorPendingJob) {
@@ -257,8 +270,10 @@ private extension ExecutorJobHandler {
     }
 
     func startNextPendingJob() {
-        guard activeJobs.count < numberOfMachines else { return }
-        guard let pendingJob = pendingJobs.first(where: { !$0.value.didStart })?.value else { return }
+        guard hasVirtualMachineSlot else { return }
+        guard let pendingJob = pendingJobs.values
+            .filter({ !$0.didStart && hasResources(for: $0) })
+            .min(by: { $0.queuedAt < $1.queuedAt }) else { return }
         start(pendingJob: pendingJob)
     }
 

Packages/ExecutorApp/Sources/ExecutorApp/Internal/ExecutorServer+Jobs.swift

@@ -128,6 +128,21 @@ extension ExecutorServer {
                 statusCode: .conflict,
                 encoder: apiEncoder
             )
+        case let .skippedInsufficientResources(cpuUsed, cpuLimit, memoryUsed, memoryLimit):
+            return .json(
+                ExecutorDispatchResponse(
+                    started: false,
+                    jobId: workflowJob.id,
+                    hostname: settings.hostname,
+                    cpu: pendingJob.cpu,
+                    memory: pendingJob.memory,
+                    forced: force,
+                    reason: .atCapacity,
+                    message: "Insufficient aggregate resources: CPU \(cpuUsed)/\(cpuLimit), memory \(memoryUsed)/\(memoryLimit) MB."
+                ),
+                statusCode: .conflict,
+                encoder: apiEncoder
+            )
         }
     }
 

Packages/ExecutorApp/Sources/ExecutorApp/Internal/OpenAPISpec.swift

@@ -382,10 +382,20 @@ components:
         startedPendingJobs: { type: integer }
         activeVirtualMachines: { type: integer }
         virtualMachineLimit: { type: integer }
-        cpuLimit: { type: integer }
+        cpuLimit:
+          type: integer
+          description: Aggregate CPU capacity available to virtual machines.
         cpuUsed: { type: integer }
-        totalMemory: { type: integer }
+        totalMemory:
+          type: integer
+          description: Aggregate memory capacity available to virtual machines, in binary MB.
         memoryUsed: { type: integer }
+        defaultCpu:
+          type: [integer, "null"]
+          description: Per-VM CPU used when a job omits the cpu label.
+        defaultMemory:
+          type: [integer, "null"]
+          description: Per-VM memory used when a job omits the memory label, in binary MB.
         diskTotalBytes:
           type: [integer, "null"]
           description: Total capacity of the volume backing the Tart home directory, in bytes.
@@ -414,8 +424,12 @@ components:
         netBridgedAdapter: { type: [string, "null"] }
         defaultCpu: { type: [integer, "null"] }
         defaultMemory: { type: [integer, "null"] }
-        cpuLimit: { type: integer }
-        totalMemory: { type: integer }
+        cpuLimit:
+          type: integer
+          description: Aggregate CPU capacity available to virtual machines.
+        totalMemory:
+          type: integer
+          description: Aggregate memory capacity available to virtual machines, in binary MB.
         loggingEndpoint: { type: [string, "null"] }
         authEnabled:
           type: boolean

Packages/ExecutorApp/Sources/ExecutorApp/Resources/openapi.yaml

@@ -377,10 +377,20 @@ components:
         startedPendingJobs: { type: integer }
         activeVirtualMachines: { type: integer }
         virtualMachineLimit: { type: integer }
-        cpuLimit: { type: integer }
+        cpuLimit:
+          type: integer
+          description: Aggregate CPU capacity available to virtual machines.
         cpuUsed: { type: integer }
-        totalMemory: { type: integer }
+        totalMemory:
+          type: integer
+          description: Aggregate memory capacity available to virtual machines, in binary MB.
         memoryUsed: { type: integer }
+        defaultCpu:
+          type: [integer, "null"]
+          description: Per-VM CPU used when a job omits the cpu label.
+        defaultMemory:
+          type: [integer, "null"]
+          description: Per-VM memory used when a job omits the memory label, in binary MB.
         diskTotalBytes:
           type: [integer, "null"]
           description: Total capacity of the volume backing the Tart home directory, in bytes.
@@ -409,8 +419,12 @@ components:
         netBridgedAdapter: { type: [string, "null"] }
         defaultCpu: { type: [integer, "null"] }
         defaultMemory: { type: [integer, "null"] }
-        cpuLimit: { type: integer }
-        totalMemory: { type: integer }
+        cpuLimit:
+          type: integer
+          description: Aggregate CPU capacity available to virtual machines.
+        totalMemory:
+          type: integer
+          description: Aggregate memory capacity available to virtual machines, in binary MB.
         loggingEndpoint: { type: [string, "null"] }
         authEnabled:
           type: boolean