Split K8sHelper, add WorkerProvisioner protocol, and add LinuxWorker implementation

Author: jshi991

Sources/ContainerK8s/Commands/K8sCreate.swift

@@ -51,6 +51,12 @@ public struct K8sCreate: AsyncParsableCommand {
     @Option(help: "Node image reference (default: \(K8sHelper.nodeImage))")
     var nodeImage: String = K8sHelper.nodeImage
 
+    public var worker: (any WorkerProvisioner)?
+
+    private enum CodingKeys: String, CodingKey {
+        case name, remove, resourceFlags, registryFlags, imageFetchFlags, nodeImage
+    }
+
     public func run() async throws {
         LoggingSystem.bootstrap { _ in StderrLogHandler() }
         let log = Logger(label: K8sHelper.pluginName)
@@ -78,15 +84,14 @@ public struct K8sCreate: AsyncParsableCommand {
         try await K8sHelper.ensureImage(nodeImage: nodeImage, log: log, containerSystemConfig: containerSystemConfig)
 
         let fqdn = K8sHelper.fqdn(for: name, domain: containerSystemConfig.dns.domain)
-        let dns = Flags.DNS(domain: nil, nameservers: [], options: [], searchDomains: [])
 
         let management = Flags.Management(
             arch: Arch.hostArchitecture().rawValue,
             capAdd: ["ALL"],
             capDrop: [],
             cidfile: "",
             detach: true,
-            dns: dns,
+            dns: Flags.DNS(),
             dnsDisabled: false,
             entrypoint: nil,
             initImage: nil,
@@ -118,7 +123,8 @@ public struct K8sCreate: AsyncParsableCommand {
         )
 
         let updatedResource = K8sHelper.defaultedResourceFlags(resourceFlags)
-        let processFlags = Flags.Process(cwd: nil, env: K8sHelper.nodeProxyEnv(), envFile: [], gid: nil, interactive: false, tty: false, uid: nil, ulimits: [], user: nil)
+        var processFlags = Flags.Process()
+        processFlags.env = K8sHelper.nodeProxyEnv()
 
         var (config, kernel, initfs) = try await Utility.containerConfigFromFlags(
             id: name,
@@ -147,40 +153,74 @@ public struct K8sCreate: AsyncParsableCommand {
             initImage: initfs
         )
 
-        progress.set(description: "Starting cluster")
-        let io = try ProcessIO.create(tty: false, interactive: false, detach: true)
-        defer { try? io.close() }
-        let process = try await client.bootstrap(id: name, stdio: io.stdio)
-        try await process.start()
-        try io.closeAfterStart()
-
-        progress.set(description: "Waiting for node to boot")
-        try await K8sHelper.waitForNodeBooted(containerId: name, client: client, log: log)
-
-        let snapshot = try await client.get(id: name)
-        guard let vmIP = snapshot.networks.first?.ipv4Address.address.description else {
-            throw ContainerizationError(.internalError, message: "no VM IP for control plane \(name)")
-        }
-        var sans = ["127.0.0.1"]
-        if let fqdn { sans.append(contentsOf: [vmIP, fqdn]) }
-
-        progress.set(description: "Running kubeadm init")
-        try await K8sHelper.prepareNode(nodeID: name, client: client, log: log)
-        try await K8sHelper.bootstrapControlPlane(
-            nodeID: name, apiServerSANs: sans, advertiseAddress: vmIP,
-            client: client, log: log)
-
-        progress.set(description: "Waiting for cluster to be ready")
-        try await K8sHelper.waitForReady(containerId: name, client: client, log: log)
-
-        progress.set(description: "Writing kubeconfig")
+        // From here on, clean up the cluster container if a worker step fails.
         do {
-            let rawConfig = try await K8sHelper.fetchConfig(containerId: name, client: client, log: log)
-            let kubeConfig = try await K8sHelper.transformConfig(rawConfig, containerId: name, fqdn: fqdn, client: client)
-            try K8sHelper.mergeConfig(kubeConfig, containerId: name, setCurrentContext: true, log: log)
+            progress.set(description: "Starting cluster")
+            let io = try ProcessIO.create(tty: false, interactive: false, detach: true)
+            defer { try? io.close() }
+            let process = try await client.bootstrap(id: name, stdio: io.stdio)
+            try await process.start()
+            try io.closeAfterStart()
+
+            progress.set(description: "Waiting for node to boot")
+            try await K8sHelper.waitForNodeBooted(containerId: name, client: client, log: log)
+
+            // Provision the worker before cluster init so its address can be added as a cert SAN.
+            let workerName = "\(name)-worker-0"
+            if let worker {
+                progress.set(description: "Provisioning worker node")
+                try await worker.provision(name: workerName, log: log)
+            }
+
+            let snapshot = try await client.get(id: name)
+            guard let vmIP = snapshot.networks.first?.ipv4Address.address.description else {
+                throw ContainerizationError(.internalError, message: "no VM IP for control plane \(name)")
+            }
+            var sans = ["127.0.0.1"]
+            if let fqdn { sans.append(contentsOf: [vmIP, fqdn]) }
+            if let worker {
+                let workerAddr = try await worker.address(name: workerName, log: log)
+                sans.append(workerAddr)
+            }
+
+            progress.set(description: "Running kubeadm init")
+            try await K8sHelper.prepareNode(nodeID: name, client: client, log: log)
+            try await K8sHelper.bootstrapControlPlane(
+                nodeID: name, apiServerSANs: sans, advertiseAddress: vmIP,
+                client: client, log: log)
+
+            progress.set(description: "Waiting for cluster to be ready")
+            try await K8sHelper.waitForReady(containerId: name, client: client, log: log)
+
+            if let worker {
+                let (token, caCertHash) = try await K8sHelper.createJoinToken(nodeID: name, client: client)
+                progress.set(description: "Joining worker node")
+                try await worker.join(
+                    name: workerName,
+                    controlPlaneEndpoint: "\(vmIP):6443",
+                    token: token,
+                    caCertHash: caCertHash,
+                    log: log)
+                progress.set(description: "Waiting for worker node to be ready")
+                try await worker.waitForReady(name: workerName, log: log)
+            }
+
+            progress.set(description: "Writing kubeconfig")
+            do {
+                let rawConfig = try await K8sHelper.fetchConfig(containerId: name, client: client, log: log)
+                let kubeConfig = try await K8sHelper.transformConfig(rawConfig, containerId: name, fqdn: fqdn, client: client)
+                try K8sHelper.mergeConfig(kubeConfig, containerId: name, setCurrentContext: true, log: log)
+            } catch {
+                log.warning("failed to write kubeconfig", metadata: ["name": "\(name)", "error": "\(error)"])
+                log.info("cluster is running; use 'container k8s write-config --name \(name)' to write the kubeconfig")
+            }
         } catch {
-            log.warning("failed to write kubeconfig", metadata: ["name": "\(name)", "error": "\(error)"])
-            log.info("cluster is running; use 'container k8s write-config --name \(name)' to write the kubeconfig")
+            if worker != nil {
+                try? await client.stop(id: name)
+                try? await client.delete(id: name)
+                try? K8sHelper.removeConfig(containerId: name, log: log)
+            }
+            throw error
         }
 
         progress.finish()

Sources/ContainerK8s/Commands/K8sDelete.swift

@@ -33,6 +33,12 @@ public struct K8sDelete: AsyncParsableCommand {
     @Option(name: .long, help: "Cluster name (default: \(K8sHelper.defaultName))")
     var name: String = K8sHelper.defaultName
 
+    public var worker: (any WorkerProvisioner)?
+
+    private enum CodingKeys: String, CodingKey {
+        case name
+    }
+
     public func run() async throws {
         LoggingSystem.bootstrap { _ in StderrLogHandler() }
         let log = Logger(label: K8sHelper.pluginName)
@@ -46,6 +52,11 @@ public struct K8sDelete: AsyncParsableCommand {
             }
         }
 
+        if let worker {
+            let workerName = "\(name)-worker-0"
+            try await worker.teardown(name: workerName, log: log)
+        }
+
         do {
             try? await client.stop(id: name)
             try await client.delete(id: name)