ManagedProcess: Capture vmexec stderr and convert to ContainerizationError

Signed-off-by: Rahul Thennarasu <[email protected]>

Author: RahulThennarasu

vminitd/Sources/vmexec/ExecCommand.swift

@@ -34,16 +34,21 @@ struct ExecCommand: ParsableCommand {
     var parentPid: Int
 
     func run() throws {
-        LoggingSystem.bootstrap(App.standardError)
-        let log = Logger(label: "vmexec")
-
-        let src = URL(fileURLWithPath: processPath)
-        let processBytes = try Data(contentsOf: src)
-        let process = try JSONDecoder().decode(
-            ContainerizationOCI.Process.self,
-            from: processBytes
-        )
-        try execInNamespaces(process: process, log: log)
+        do {
+            LoggingSystem.bootstrap(App.standardError)
+            let log = Logger(label: "vmexec")
+
+            let src = URL(fileURLWithPath: processPath)
+            let processBytes = try Data(contentsOf: src)
+            let process = try JSONDecoder().decode(
+                ContainerizationOCI.Process.self,
+                from: processBytes
+            )
+            try execInNamespaces(process: process, log: log)
+        } catch {
+            App.writeError(error)
+            throw error
+        }
     }
 
     static func enterNS(pidFd: Int32, nsType: Int32) throws {

vminitd/Sources/vmexec/RunCommand.swift

@@ -32,12 +32,17 @@ struct RunCommand: ParsableCommand {
     var bundlePath: String
 
     mutating func run() throws {
-        LoggingSystem.bootstrap(App.standardError)
-        let log = Logger(label: "vmexec")
-
-        let bundle = try ContainerizationOCI.Bundle.load(path: URL(filePath: bundlePath))
-        let ociSpec = try bundle.loadConfig()
-        try execInNamespace(spec: ociSpec, log: log)
+        do {
+            LoggingSystem.bootstrap(App.standardError)
+            let log = Logger(label: "vmexec")
+
+            let bundle = try ContainerizationOCI.Bundle.load(path: URL(filePath: bundlePath))
+            let ociSpec = try bundle.loadConfig()
+            try execInNamespace(spec: ociSpec, log: log)
+        } catch {
+            App.writeError(error)
+            throw error
+        }
     }
 
     private func childRootSetup(rootfs: ContainerizationOCI.Root, mounts: [ContainerizationOCI.Mount], log: Logger) throws {

vminitd/Sources/vmexec/vmexec.swift

@@ -182,4 +182,20 @@ extension App {
             message: message
         )
     }
+
+    static func writeError(_ error: Error) {
+        let errorPipe = FileHandle(fileDescriptor: 5)
+
+        let errorMessage: String
+        if let czError = error as? ContainerizationError {
+            errorMessage = czError.description
+        } else {
+            errorMessage = String(describing: error)
+        }
+
+        if let data = errorMessage.data(using: .utf8) {
+            try? errorPipe.write(contentsOf: data)
+        }
+        try? errorPipe.close()
+    }
 }

vminitd/Sources/vminitd/ManagedProcess.swift

@@ -50,6 +50,7 @@ final class ManagedProcess: Sendable {
         var waiters: [CheckedContinuation<ExitStatus, Never>] = []
         var exitStatus: ExitStatus? = nil
         var pid: Int32?
+        var errorPipe: Pipe?
     }
 
     private static let ackPid = "AckPid"
@@ -95,6 +96,9 @@ final class ManagedProcess: Sendable {
         try ackPipe.setCloexec()
         self.ackPipe = ackPipe
 
+        let errorPipe = Pipe()
+        try errorPipe.setCloexec()
+
         let args: [String]
         if let owningPid {
             args = [
@@ -114,6 +118,7 @@ final class ManagedProcess: Sendable {
             extraFiles: [
                 syncPipe.fileHandleForWriting,
                 ackPipe.fileHandleForReading,
+                errorPipe.fileHandleForWriting,
             ]
         )
 
@@ -144,130 +149,153 @@ final class ManagedProcess: Sendable {
         self.terminal = stdio.terminal
         self.bundle = bundle
         self.state = Mutex(State(io: io))
+        self.state.withLock { state in
+            state.errorPipe = errorPipe
+        }
     }
 }
 
 extension ManagedProcess {
     func start() throws -> Int32 {
-        try self.state.withLock {
-            log.info(
-                "starting managed process",
-                metadata: [
-                    "id": "\(id)"
-                ])
-
-            // Start the underlying process.
-            try command.start()
-            defer {
-                try? self.ackPipe.fileHandleForWriting.close()
-                try? self.syncPipe.fileHandleForReading.close()
-                try? self.ackPipe.fileHandleForReading.close()
-                try? self.syncPipe.fileHandleForWriting.close()
-            }
-
-            // Close our side of any pipes.
-            try $0.io.closeAfterExec()
-            try self.ackPipe.fileHandleForReading.close()
-            try self.syncPipe.fileHandleForWriting.close()
+        do {
+            try self.state.withLock {
+                log.info(
+                    "starting managed process",
+                    metadata: [
+                        "id": "\(id)"
+                    ])
 
-            let size = MemoryLayout<Int32>.size
-            guard let piddata = try syncPipe.fileHandleForReading.read(upToCount: size) else {
-                throw ContainerizationError(.internalError, message: "no PID data from sync pipe")
-            }
+                // Start the underlying process.
+                try command.start()
 
-            guard piddata.count == size else {
-                throw ContainerizationError(.internalError, message: "invalid payload")
-            }
+                defer {
+                    try? self.ackPipe.fileHandleForWriting.close()
+                    try? self.syncPipe.fileHandleForReading.close()
+                    try? self.ackPipe.fileHandleForReading.close()
+                    try? self.syncPipe.fileHandleForWriting.close()
+                }
 
-            let pid = piddata.withUnsafeBytes { ptr in
-                ptr.load(as: Int32.self)
-            }
+                // Close our side of any pipes.
+                try $0.io.closeAfterExec()
+                try self.ackPipe.fileHandleForReading.close()
+                try self.syncPipe.fileHandleForWriting.close()
 
-            log.info(
-                "got back pid data",
-                metadata: [
-                    "pid": "\(pid)"
-                ])
-            $0.pid = pid
+                let size = MemoryLayout<Int32>.size
+                guard let piddata = try syncPipe.fileHandleForReading.read(upToCount: size) else {
+                    throw ContainerizationError(.internalError, message: "no PID data from sync pipe")
+                }
 
-            // This should probably happen in vmexec, but we don't need to set any cgroup
-            // toggles so the problem is much simpler to just do it here.
-            if let owningPid {
-                let cgManager = try Cgroup2Manager.loadFromPid(pid: owningPid)
-                try cgManager.addProcess(pid: pid)
-            }
+                guard piddata.count == size else {
+                    throw ContainerizationError(.internalError, message: "invalid payload")
+                }
 
-            log.info(
-                "sending pid acknowledgement",
-                metadata: [
-                    "pid": "\(pid)"
-                ])
-            try self.ackPipe.fileHandleForWriting.write(contentsOf: Self.ackPid.data(using: .utf8)!)
+                let pid = piddata.withUnsafeBytes { ptr in
+                    ptr.load(as: Int32.self)
+                }
 
-            if self.terminal {
                 log.info(
-                    "wait for PTY FD",
+                    "got back pid data",
                     metadata: [
-                        "id": "\(id)"
+                        "pid": "\(pid)"
                     ])
+                $0.pid = pid
 
-                // Wait for a new write that will contain the pty fd if we asked for one.
-                guard let ptyFd = try self.syncPipe.fileHandleForReading.read(upToCount: size) else {
-                    throw ContainerizationError(
-                        .internalError,
-                        message: "no PTY data from sync pipe"
-                    )
-                }
-                let fd = ptyFd.withUnsafeBytes { ptr in
-                    ptr.load(as: Int32.self)
+                // This should probably happen in vmexec, but we don't need to set any cgroup
+                // toggles so the problem is much simpler to just do it here.
+                if let owningPid {
+                    let cgManager = try Cgroup2Manager.loadFromPid(pid: owningPid)
+                    try cgManager.addProcess(pid: pid)
                 }
+
                 log.info(
-                    "received PTY FD from container, attaching",
+                    "sending pid acknowledgement",
                     metadata: [
-                        "id": "\(id)"
+                        "pid": "\(pid)"
                     ])
+                try self.ackPipe.fileHandleForWriting.write(contentsOf: Self.ackPid.data(using: .utf8)!)
+
+                if self.terminal {
+                    log.info(
+                        "wait for PTY FD",
+                        metadata: [
+                            "id": "\(id)"
+                        ])
+
+                    // Wait for a new write that will contain the pty fd if we asked for one.
+                    guard let ptyFd = try self.syncPipe.fileHandleForReading.read(upToCount: size) else {
+                        throw ContainerizationError(
+                            .internalError,
+                            message: "no PTY data from sync pipe"
+                        )
+                    }
+                    let fd = ptyFd.withUnsafeBytes { ptr in
+                        ptr.load(as: Int32.self)
+                    }
+                    log.info(
+                        "received PTY FD from container, attaching",
+                        metadata: [
+                            "id": "\(id)"
+                        ])
+
+                    try $0.io.attach(pid: pid, fd: fd)
+                    try self.ackPipe.fileHandleForWriting.write(contentsOf: Self.ackConsole.data(using: .utf8)!)
+                }
 
-                try $0.io.attach(pid: pid, fd: fd)
-                try self.ackPipe.fileHandleForWriting.write(contentsOf: Self.ackConsole.data(using: .utf8)!)
-            }
-
-            // Wait for the syncPipe to close (after exec).
-            _ = try self.syncPipe.fileHandleForReading.readToEnd()
+                // Wait for the syncPipe to close (after exec).
+                _ = try self.syncPipe.fileHandleForReading.readToEnd()
 
-            log.info(
-                "started managed process",
-                metadata: [
-                    "pid": "\(pid)",
-                    "id": "\(id)",
-                ])
+                log.info(
+                    "started managed process",
+                    metadata: [
+                        "pid": "\(pid)",
+                        "id": "\(id)",
+                    ])
 
-            return pid
+                return pid
+            }
+        } catch {
+            if let errorPipe = self.state.withLock({ $0.errorPipe }) {
+                if let errorData = try? errorPipe.fileHandleForReading.readToEnd(),
+                    let errorString = String(data: errorData, encoding: .utf8),
+                    !errorString.isEmpty
+                {
+                    throw ContainerizationError(
+                        .internalError,
+                        message: "vmexec error: \(errorString.trimmingCharacters(in: .whitespacesAndNewlines))"
+                    )
+                }
+            }
+            throw error
         }
     }
 
     func setExit(_ status: Int32) {
-        self.state.withLock {
+        self.state.withLock { state in
             self.log.info(
                 "managed process exit",
                 metadata: [
                     "status": "\(status)"
                 ])
 
             let exitStatus = ExitStatus(exitStatus: status, exitedAt: Date.now)
-            $0.exitStatus = exitStatus
+            state.exitStatus = exitStatus
+
+            try? state.errorPipe?.fileHandleForReading.close()
+            try? state.errorPipe?.fileHandleForWriting.close()
+            state.errorPipe = nil
 
             do {
-                try $0.io.close()
+                try state.io.close()
             } catch {
                 self.log.error("failed to close I/O for process: \(error)")
             }
 
-            for waiter in $0.waiters {
+            for waiter in state.waiters {
                 waiter.resume(returning: exitStatus)
             }
 
-            self.log.debug("\($0.waiters.count) managed process waiters signaled")
-            $0.waiters.removeAll()
+            self.log.debug("\(state.waiters.count) managed process waiters signaled")
+            state.waiters.removeAll()
         }
     }