Swap to APIServer for all communications

Today, we have a sort of odd model all things considered. We
talk to the APIServer to do the initial container creation, which
mostly all of the work is just registering the runtime helper with
launchd. After this registration, almost all communication from a client
is talking directly to that runtime helper. This has a couple rather annoying
issues in that we now need a sort of event/notification channel that the helper
will establish with the APIServer for errors/start/exited cases.

If instead of talking to the runtime helper directly, we instead took a detour
through the APIServer, the APIServer is clued into exactly what order of operations
is occurring. This makes "did starting the container fail? Okay we should clean
up" scenarios much simpler, and it also simplifies the clients quite a bit
as they don't need this split brained client model, everyone just talks to the
APIServer. This change is in pursuit of that. I have reworked our clients, the
ContainerService and some of our XPC types to accomplish it.

The biggest "contract" changes are in the SandboxService. The first is today we have
on the flag when we register any runtime helper that makes any xpc messages wake up the
registered process. This isn't great in scenarios where the process may have crashed, or
it exited normally and we're just trying to invoke an RPC on it. Today the helper would
spawn again and try and answer our request. It'd be much nicer if we have a connection
object that will become invalid if the process that vended it to us is gone. To accomplish
this, now the runtime helpers will listen on an anonymous xpc connection and vend endpoints
from this via only one handler exposed by the SandboxService (createEndpoint). From that point
onwards all communication will be through the endpoint the service vended a client.

The second change is the runtime helper will not exit on its own when the container exits, and
the event mechanism has been removed. Now the APIServer simply calls wait() to listen for container
exit in the background, and once we get an exit we will explicitly tell the helper to shutdown.
The rationale is if shutdown is driven by the APIServer now, we can be certain we received everything
we need from the helpers before they power down.

Author: dcantah

Sources/CLI/Builder/BuilderStart.swift

@@ -249,7 +249,7 @@ extension ClientContainer {
             defer { try? io.close() }
 
             let process = try await bootstrap(stdio: io.stdio)
-            _ = try await process.start()
+            try await process.start()
             await taskManager?.finish()
             try io.closeAfterStart()
 

Sources/CLI/Container/ContainerStop.swift

@@ -68,7 +68,10 @@ extension Application {
             )
             let failed = try await Self.stopContainers(containers: containers, stopOptions: opts)
             if failed.count > 0 {
-                throw ContainerizationError(.internalError, message: "stop failed for one or more containers \(failed.joined(separator: ","))")
+                throw ContainerizationError(
+                    .internalError,
+                    message: "stop failed for one or more containers \(failed.joined(separator: ","))"
+                )
             }
         }
 

Sources/CLI/RunCommand.swift

@@ -109,7 +109,6 @@ extension Application {
             )
 
             let detach = self.managementFlags.detach
-
             do {
                 let io = try ProcessIO.create(
                     tty: self.processFlags.tty,

Sources/ContainerClient/ContainerEvents.swift

@@ -15,6 +15,5 @@
 //===----------------------------------------------------------------------===//
 
 public enum ContainerEvent: Sendable, Codable {
-    case containerStart(id: String)
     case containerExit(id: String, exitCode: Int64)
 }

Sources/ContainerClient/Core/ClientContainer.swift

@@ -25,10 +25,6 @@ import TerminalProgress
 public struct ClientContainer: Sendable, Codable {
     static let serviceIdentifier = "com.apple.container.apiserver"
 
-    private var sandboxClient: SandboxClient {
-        SandboxClient(id: configuration.id, runtime: configuration.runtimeHandler)
-    }
-
     /// Identifier of the container.
     public var id: String {
         configuration.id
@@ -58,10 +54,6 @@ public struct ClientContainer: Sendable, Codable {
         self.status = snapshot.status
         self.networks = snapshot.networks
     }
-
-    public var initProcess: ClientProcess {
-        ClientProcessImpl(containerId: self.id, client: self.sandboxClient)
-    }
 }
 
 extension ClientContainer {
@@ -85,7 +77,7 @@ extension ClientContainer {
     ) async throws -> ClientContainer {
         do {
             let client = Self.newClient()
-            let request = XPCMessage(route: .createContainer)
+            let request = XPCMessage(route: .containerCreate)
 
             let data = try JSONEncoder().encode(configuration)
             let kdata = try JSONEncoder().encode(kernel)
@@ -108,7 +100,7 @@ extension ClientContainer {
     public static func list() async throws -> [ClientContainer] {
         do {
             let client = Self.newClient()
-            let request = XPCMessage(route: .listContainer)
+            let request = XPCMessage(route: .containerList)
 
             let response = try await xpcSend(
                 client: client,
@@ -145,16 +137,67 @@ extension ClientContainer {
 
 extension ClientContainer {
     public func bootstrap(stdio: [FileHandle?]) async throws -> ClientProcess {
-        let client = self.sandboxClient
-        try await client.bootstrap(stdio: stdio)
-        return ClientProcessImpl(containerId: self.id, client: self.sandboxClient)
+        let request = XPCMessage(route: .containerBootstrap)
+        let client = Self.newClient()
+
+        for (i, h) in stdio.enumerated() {
+            let key: XPCKeys = {
+                switch i {
+                case 0: .stdin
+                case 1: .stdout
+                case 2: .stderr
+                default:
+                    fatalError("invalid fd \(i)")
+                }
+            }()
+
+            if let h {
+                request.set(key: key, value: h)
+            }
+        }
+
+        do {
+            request.set(key: .id, value: self.id)
+            try await client.send(request)
+            return ClientProcessImpl(containerId: self.id, client: client)
+        } catch {
+            throw ContainerizationError(
+                .internalError,
+                message: "failed to bootstrap container",
+                cause: error
+            )
+        }
+    }
+
+    public func kill(_ signal: Int32) async throws {
+        do {
+            let request = XPCMessage(route: .containerKill)
+            request.set(key: .id, value: id)
+            request.set(key: .processIdentifier, value: id)
+            request.set(key: .signal, value: Int64(signal))
+
+            let client = Self.newClient()
+            try await client.send(request)
+        } catch {
+            throw ContainerizationError(
+                .internalError,
+                message: "failed to kill container",
+                cause: error
+            )
+        }
     }
 
     /// Stop the container and all processes currently executing inside.
     public func stop(opts: ContainerStopOptions = ContainerStopOptions.default) async throws {
         do {
-            let client = self.sandboxClient
-            try await client.stop(options: opts)
+            let client = Self.newClient()
+            let request = XPCMessage(route: .containerStop)
+            let data = try JSONEncoder().encode(opts)
+            request.set(key: .id, value: self.id)
+            request.set(key: .stopOptions, value: data)
+
+            let responseTimeout = Duration(.seconds(Int64(opts.timeoutInSeconds + 1)))
+            try await client.send(request, responseTimeout: responseTimeout)
         } catch {
             throw ContainerizationError(
                 .internalError,
@@ -167,8 +210,8 @@ extension ClientContainer {
     /// Delete the container along with any resources.
     public func delete(force: Bool = false) async throws {
         do {
-            let client = XPCClient(service: Self.serviceIdentifier)
-            let request = XPCMessage(route: .deleteContainer)
+            let client = Self.newClient()
+            let request = XPCMessage(route: .containerDelete)
             request.set(key: .id, value: self.id)
             request.set(key: .forceDelete, value: force)
             try await client.send(request)
@@ -180,46 +223,53 @@ extension ClientContainer {
             )
         }
     }
-}
 
-extension ClientContainer {
-    /// Execute a new process inside a running container.
+    /// Create a new process inside a running container. The process is in a
+    /// created state and must still be started.
     public func createProcess(
         id: String,
         configuration: ProcessConfiguration,
         stdio: [FileHandle?]
     ) async throws -> ClientProcess {
         do {
-            let client = self.sandboxClient
-            try await client.createProcess(id, config: configuration, stdio: stdio)
-            return ClientProcessImpl(containerId: self.id, processId: id, client: client)
-        } catch {
-            throw ContainerizationError(
-                .internalError,
-                message: "failed to exec in container",
-                cause: error
-            )
-        }
-    }
+            let request = XPCMessage(route: .containerCreateProcess)
+            request.set(key: .id, value: self.id)
+            request.set(key: .processIdentifier, value: id)
 
-    /// Send or "kill" a signal to the initial process of the container.
-    /// Kill does not wait for the process to exit, it only delivers the signal.
-    public func kill(_ signal: Int32) async throws {
-        do {
-            let client = self.sandboxClient
-            try await client.kill(self.id, signal: Int64(signal))
+            let data = try JSONEncoder().encode(configuration)
+            request.set(key: .processConfig, value: data)
+
+            for (i, h) in stdio.enumerated() {
+                let key: XPCKeys = {
+                    switch i {
+                    case 0: .stdin
+                    case 1: .stdout
+                    case 2: .stderr
+                    default:
+                        fatalError("invalid fd \(i)")
+                    }
+                }()
+
+                if let h {
+                    request.set(key: key, value: h)
+                }
+            }
+
+            let client = Self.newClient()
+            try await client.send(request)
+            return ClientProcessImpl(containerId: self.id, processId: id, client: client)
         } catch {
             throw ContainerizationError(
                 .internalError,
-                message: "failed to kill container \(self.id)",
+                message: "failed to create process in container",
                 cause: error
             )
         }
     }
 
     public func logs() async throws -> [FileHandle] {
         do {
-            let client = XPCClient(service: Self.serviceIdentifier)
+            let client = Self.newClient()
             let request = XPCMessage(route: .containerLogs)
             request.set(key: .id, value: self.id)
 
@@ -242,15 +292,27 @@ extension ClientContainer {
     }
 
     public func dial(_ port: UInt32) async throws -> FileHandle {
+        let request = XPCMessage(route: .containerDial)
+        request.set(key: .id, value: self.id)
+        request.set(key: .port, value: UInt64(port))
+
+        let client = Self.newClient()
+        let response: XPCMessage
         do {
-            let client = self.sandboxClient
-            return try await client.dial(port)
+            response = try await client.send(request)
         } catch {
             throw ContainerizationError(
                 .internalError,
-                message: "failed to dial \(port) in container \(self.id)",
+                message: "failed to dial port \(port) on container",
                 cause: error
             )
         }
+        guard let fh = response.fileHandle(key: .fd) else {
+            throw ContainerizationError(
+                .internalError,
+                message: "failed to get fd for vsock port \(port)"
+            )
+        }
+        return fh
     }
 }

Sources/ContainerClient/Core/ClientProcess.swift

@@ -35,7 +35,7 @@ public protocol ClientProcess: Sendable {
     func start() async throws
     /// Send a terminal resize request to the process `id`.
     func resize(_ size: Terminal.Size) async throws
-    /// Send or "kill" a signal to the process `id`.
+    /// Send a signal to the process `id`.
     /// Kill does not wait for the process to exit, it only delivers the signal.
     func kill(_ signal: Int32) async throws
     ///  Wait for the process `id` to complete and return its exit code.
@@ -45,79 +45,66 @@ public protocol ClientProcess: Sendable {
 
 struct ClientProcessImpl: ClientProcess, Sendable {
     static let serviceIdentifier = "com.apple.container.apiserver"
+
+    /// ID of the process.
+    public var id: String {
+        processId ?? containerId
+    }
+
     /// Identifier of the container.
     public let containerId: String
 
-    private let client: SandboxClient
-
     /// Identifier of a process. That is running inside of a container.
     /// This field is nil if the process this objects refers to is the
     /// init process of the container.
     public let processId: String?
 
-    public var id: String {
-        processId ?? containerId
-    }
+    private let client: XPCClient
 
-    init(containerId: String, processId: String? = nil, client: SandboxClient) {
+    init(containerId: String, processId: String? = nil, client: XPCClient) {
         self.containerId = containerId
         self.processId = processId
         self.client = client
     }
 
-    /// Start the container and return the initial process.
+    /// Start the process.
     public func start() async throws {
-        do {
-            let client = self.client
-            try await client.startProcess(self.id)
-        } catch {
-            throw ContainerizationError(
-                .internalError,
-                message: "failed to start container",
-                cause: error
-            )
-        }
+        let request = XPCMessage(route: .containerStartProcess)
+        request.set(key: .id, value: containerId)
+        request.set(key: .processIdentifier, value: id)
+
+        try await client.send(request)
     }
 
+    /// Send a signal to the process.
     public func kill(_ signal: Int32) async throws {
-        do {
-
-            let client = self.client
-            try await client.kill(self.id, signal: Int64(signal))
-        } catch {
-            throw ContainerizationError(
-                .internalError,
-                message: "failed to kill process",
-                cause: error
-            )
-        }
-    }
+        let request = XPCMessage(route: .containerKill)
+        request.set(key: .id, value: containerId)
+        request.set(key: .processIdentifier, value: id)
+        request.set(key: .signal, value: Int64(signal))
 
-    public func resize(_ size: ContainerizationOS.Terminal.Size) async throws {
-        do {
+        try await client.send(request)
+    }
 
-            let client = self.client
-            try await client.resize(self.id, size: size)
+    /// Resize the processes PTY if it has one.
+    public func resize(_ size: Terminal.Size) async throws {
+        let request = XPCMessage(route: .containerResize)
+        request.set(key: .id, value: containerId)
+        request.set(key: .processIdentifier, value: id)
+        request.set(key: .width, value: UInt64(size.width))
+        request.set(key: .height, value: UInt64(size.height))
 
-        } catch {
-            throw ContainerizationError(
-                .internalError,
-                message: "failed to resize process",
-                cause: error
-            )
-        }
+        try await client.send(request)
     }
 
+    /// Wait for the process to exit.
     public func wait() async throws -> Int32 {
-        do {
-            let client = self.client
-            return try await client.wait(self.id)
-        } catch {
-            throw ContainerizationError(
-                .internalError,
-                message: "failed to wait on process",
-                cause: error
-            )
-        }
+        let request = XPCMessage(route: .containerWait)
+        request.set(key: .id, value: containerId)
+        request.set(key: .processIdentifier, value: id)
+
+        let response = try await client.send(request)
+        let code = response.int64(key: .exitCode)
+        return Int32(code)
     }
 }