diff --git a/Sources/Services/RuntimeLinux/Server/RuntimeService.swift b/Sources/Services/RuntimeLinux/Server/RuntimeService.swift index cbb36b259..cfc440183 100644 --- a/Sources/Services/RuntimeLinux/Server/RuntimeService.swift +++ b/Sources/Services/RuntimeLinux/Server/RuntimeService.swift @@ -46,11 +46,14 @@ public actor RuntimeService { private let eventLoopGroup: any EventLoopGroup private var waiters: [String: ExitWaiter] = [:] private let lock: AsyncLock = AsyncLock() + private var isCleaning = false private let log: Logging.Logger private var state: State = .created private var processes: [String: ProcessInfo] = [:] private var socketForwarders: [SocketForwarderResult] = [] private var networkSessions: [XPCClientSession] = [] + private let cleanInterval: Duration? + private var periodicCleanTask: Task? = nil private static let sshAuthSocketGuestPath = "/var/host-services/ssh-auth.sock" private static let sshAuthSocketEnvVar = "SSH_AUTH_SOCK" @@ -99,6 +102,7 @@ public actor RuntimeService { interfaceStrategies: [NetworkInterfaceKey: InterfaceStrategy], eventLoopGroup: any EventLoopGroup, connection: xpc_connection_t, + cleanInterval: Duration? = .seconds(3600), log: Logger ) { self.root = root @@ -107,6 +111,7 @@ public actor RuntimeService { self.monitor = ExitMonitor(log: log) self.eventLoopGroup = eventLoopGroup self.connection = connection + self.cleanInterval = cleanInterval } /// Returns an endpoint from an anonymous xpc connection. @@ -335,18 +340,24 @@ public actor RuntimeService { self.log.debug("enter", metadata: ["func": "\(#function)"]) defer { self.log.debug("exit", metadata: ["func": "\(#function)"]) } - return try await self.lock.withLock { lock in + let isInit = try await self.lock.withLock { lock in let id = try message.id() let containerInfo = try await self.getContainer() let containerId = containerInfo.container.id - if id == containerId { - try await self.startInitProcess(lock: lock) - await self.setState(.running) - } else { + guard id == containerId else { try await self.startExecProcess(processId: id, lock: lock) + return false } - return message.reply() + try await self.startInitProcess(lock: lock) + await self.setState(.running) + return true + } + + if isInit { + self.startPeriodicClean() } + + return message.reply() } /// Get statistics for the container. @@ -856,59 +867,167 @@ public actor RuntimeService { /// Clean up unused space in the container filesystem. /// - /// - Parameters: - /// - message: An XPC message with the following parameters: - /// - id: The container ID. + /// - Parameter message: An XPC message with the container ID. /// /// - Returns: An XPC message with no parameters. + /// - Throws: An error if cleaning fails or the container is not running. @Sendable public func clean(_ message: XPCMessage) async throws -> XPCMessage { self.log.info("`clean` xpc handler") - switch self.state { - case .running: - guard let id = message.string(key: RuntimeKeys.id.rawValue) else { - throw ContainerizationError( - .invalidArgument, - message: "no id supplied for clean" + guard message.string(key: RuntimeKeys.id.rawValue) != nil else { + throw ContainerizationError( + .invalidArgument, + message: "no id supplied for clean" + ) + } + + guard self.state == .running else { + throw ContainerizationError( + .invalidState, + message: "cannot clean: container is not running" + ) + } + + guard !self.isCleaning else { + throw ContainerizationError( + .invalidState, + message: "clean already in progress" + ) + } + + self.isCleaning = true + defer { self.isCleaning = false } + + try await self.performClean() + + return message.reply() + } + + /// Internal helper that performs the filesystem trim operations on the root filesystem and + /// any block-backed read-write mounts. + @discardableResult + private func performClean() async throws -> [String] { + guard self.state == .running else { + throw ContainerizationError( + .invalidState, + message: "cannot clean: container is not running" + ) + } + + let ctr = try getContainer() + + var targets: [String] = [] + if !ctr.config.readOnly { + targets.append("/") + } + for mount in ctr.config.mounts where mount.isBlock && !mount.options.readonly { + targets.append(mount.destination) + } + + var failed: [String] = [] + for path in targets { + let clock = ContinuousClock() + let start = clock.now + do { + try await ctr.container.filesystemOperation(operation: .trim, path: path) + let elapsed = clock.now - start + self.log.info( + "cleaned mount", + metadata: [ + "path": "\(path)", + "elapsedSeconds": "\(elapsed.secondsWithMilliseconds)", + ] ) + } catch { + let elapsed = clock.now - start + self.log.error( + "failed to clean mount", + metadata: [ + "path": "\(path)", + "elapsedSeconds": "\(elapsed.secondsWithMilliseconds)", + "error": "\(error)", + ] + ) + failed.append("\(path) (\(error))") } + } - let ctr = try getContainer() + guard failed.isEmpty else { + throw ContainerizationError( + .internalError, + message: "failed to clean mounts in \(ctr.container.id): \(failed.joined(separator: ", "))" + ) + } - var targets: [String] = [] - if !ctr.config.readOnly { - targets.append("/") - } - for mount in ctr.config.mounts where mount.isBlock && !mount.options.readonly { - targets.append(mount.destination) - } + return targets + } + + private func performPeriodicClean() async { + guard self.state == .running else { + return + } + + guard !self.isCleaning else { + self.log.info("skipping periodic container clean: clean already in progress") + return + } + + self.isCleaning = true + defer { self.isCleaning = false } + + self.log.info("running periodic container clean") + let clock = ContinuousClock() + let start = clock.now + do { + try await self.performClean() + let duration = clock.now - start + self.log.info( + "periodic container clean completed", + metadata: ["elapsedSeconds": "\(duration.secondsWithMilliseconds)"] + ) + } catch { + let duration = clock.now - start + self.log.error( + "periodic container clean failed", + metadata: [ + "elapsedSeconds": "\(duration.secondsWithMilliseconds)", + "error": "\(error)", + ] + ) + } + } - var failed: [String] = [] - for path in targets { + private func startPeriodicClean() { + guard self.periodicCleanTask == nil, let cleanInterval = self.cleanInterval, cleanInterval > .zero else { + return + } + + self.periodicCleanTask = Task { [weak self] in + while !Task.isCancelled { do { - try await ctr.container.filesystemOperation(operation: .trim, path: path) + try await Task.sleep(for: cleanInterval) } catch { - self.log.error("failed to clean mount", metadata: ["path": "\(path)", "error": "\(error)"]) - failed.append("\(path) (\(error))") + break } - } - guard failed.isEmpty else { - throw ContainerizationError( - .internalError, - message: "failed to clean mounts in \(id): \(failed.joined(separator: ", "))" - ) - } + guard !Task.isCancelled else { + break + } - return message.reply() - default: - throw ContainerizationError( - .invalidState, - message: "cannot clean: container is not running" - ) + guard let self else { + break + } + + await self.performPeriodicClean() + } } } + private func stopPeriodicClean() { + self.periodicCleanTask?.cancel() + self.periodicCleanTask = nil + } + /// Dial a vsock port on the virtual machine. /// /// - Parameters: @@ -1401,6 +1520,7 @@ public actor RuntimeService { } private func cleanUpContainer(containerInfo: ContainerInfo, exitStatus: ExitStatus? = nil) async throws { + self.stopPeriodicClean() let container = containerInfo.container let id = container.id @@ -1724,3 +1844,12 @@ extension RuntimeService { } } } + +extension Duration { + /// Seconds as a string with millisecond resolution, e.g. "12.345". + fileprivate var secondsWithMilliseconds: String { + let (seconds, attoseconds) = self.components + let ms = seconds * 1000 + attoseconds / 1_000_000_000_000_000 + return String(format: "%d.%03d", ms / 1000, ms % 1000) + } +}