diff --git a/Sources/ContainerBuild/Builder.swift b/Sources/ContainerBuild/Builder.swift index 796ba6b3d..151ee033e 100644 --- a/Sources/ContainerBuild/Builder.swift +++ b/Sources/ContainerBuild/Builder.swift @@ -108,7 +108,7 @@ public struct Builder: Sendable { } if let terminal = config.terminal { - _ = Task { + Task { let winchHandler = AsyncSignalHandler.create(notify: [SIGWINCH]) let setWinch = { (rows: UInt16, cols: UInt16) in var winch = ClientStream() diff --git a/Sources/ContainerK8s/Commands/K8sCreate.swift b/Sources/ContainerK8s/Commands/K8sCreate.swift index c19bc20f5..93ba9fb22 100644 --- a/Sources/ContainerK8s/Commands/K8sCreate.swift +++ b/Sources/ContainerK8s/Commands/K8sCreate.swift @@ -75,112 +75,52 @@ public struct K8sCreate: AsyncParsableCommand { progress.start() let containerSystemConfig: ContainerSystemConfig = try await ConfigurationLoader.load() - 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, - dnsDisabled: false, - entrypoint: nil, - initImage: nil, - kernel: nil, - kernelArgs: [], - labels: [ - "\(ResourceLabelKeys.plugin)=\(K8sHelper.pluginName)", - "\(ResourceLabelKeys.role)=\(K8sHelper.controlPlaneRoleName)", - ], - maskedPaths: [], - mounts: [], - name: name, - networks: [], - os: "linux", - platform: nil, - publishPorts: fqdn == nil ? [try await K8sHelper.clusterPort()] : [], - publishSockets: [], - readOnly: false, - readonlyPaths: [], - remove: remove, - rosetta: true, - runtime: nil, - ssh: false, - shmSize: nil, - tmpFs: [], - useInit: false, - virtualization: false, - volumes: [] - ) - - 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 (config, kernel, initfs) = try await Utility.containerConfigFromFlags( - id: name, - image: nodeImage, - arguments: [], - process: processFlags, - management: management, - resource: updatedResource, - registry: registryFlags, - imageFetch: imageFetchFlags, - containerSystemConfig: containerSystemConfig, - progressUpdate: progress.handler, - log: log - ) - - // Allow the node to modify /proc/sys (e.g. net.ipv4.ip_forward) during setup. - config.maskedPaths = [] - config.readonlyPaths = [] - let client = ContainerClient() - let options = ContainerCreateOptions(autoRemove: remove) - try await client.create( - configuration: config, - options: options, - kernel: kernel, - initImage: initfs + let provisioner = try LinuxNodeProvisioner( + clusterName: name, + roles: [StandardRoles.controlPlane], + nodeImage: nodeImage, + cpus: resourceFlags.cpus, + memory: resourceFlags.memory, + registryScheme: registryFlags.scheme, + maxConcurrentDownloads: imageFetchFlags.maxConcurrentDownloads, + remove: remove, + fqdn: fqdn ) 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]) } + try await provisioner.provision(name: name, log: log) - 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") + let client = ContainerClient() 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) + let vmIP = try await provisioner.address(name: name, log: log) + 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, + schedulable: provisioner.roles.contains(StandardRoles.worker), + 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") + 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") + try? await provisioner.teardown(name: name, log: log) + try? K8sHelper.removeConfig(containerId: name, log: log) + throw error } progress.finish() diff --git a/Sources/ContainerK8s/K8sHelper.swift b/Sources/ContainerK8s/K8sHelper.swift index ffbf79c07..2bc4c9d7f 100644 --- a/Sources/ContainerK8s/K8sHelper.swift +++ b/Sources/ContainerK8s/K8sHelper.swift @@ -15,183 +15,54 @@ //===----------------------------------------------------------------------===// import ContainerAPIClient -import ContainerPersistence -import ContainerPlugin import ContainerResource -import ContainerVersion import ContainerizationError -import ContainerizationOCI import ContainerizationOS -import Darwin import Foundation import Logging -import SystemPackage -import Yams - -// MARK: - ListDisplayable - -protocol ListDisplayable { - static var tableHeader: [String] { get } - var tableRow: [String] { get } - var quietValue: String { get } -} - -// MARK: - TableOutput - -struct TableOutput: Sendable { - private let rows: [[String]] - private let spacing: Int - - init(rows: [[String]], spacing: Int = 2) { - self.rows = rows - self.spacing = spacing - } - - func format() -> String { - var output = "" - let maxLengths = self.maxLength() - - for rowIndex in 0.. [Int: Int] { - var output: [Int: Int] = [:] - for row in self.rows { - for (i, column) in row.enumerated() { - let currentMax = output[i] ?? 0 - output[i] = (column.count > currentMax) ? column.count : currentMax - } - } - return output - } -} // MARK: - K8sHelper -struct K8sHelper { - static let pluginName: String = "k8s" - static let defaultName: String = "k8s-dev" - static let controlPlaneRoleName: String = "control-plane" +public struct K8sHelper { + public static let pluginName: String = "k8s" + public static let defaultName: String = "k8s-dev" + public static let controlPlaneRoleName: String = "control-plane" + public static let workerRoleName: String = "worker" private static var defaultCPUs: Int64 { Int64(max(ProcessInfo.processInfo.processorCount / 4, 2)) } - private static var defaultMemory: String { let gb = Int(ProcessInfo.processInfo.physicalMemory / (1024 * 1024 * 1024)) / 4 return "\(max(gb, 2))g" } - static let nodeImage = "docker.io/kindest/node:v1.35.5@sha256:ce977ae6d65918d0b58a5f8b5e940429c2ce42fa3a5619ec2bbc60b949c0ac95" - private static let kubeconfigPath = "/etc/kubernetes/admin.conf" - private static let kubeconfigEnv = "KUBECONFIG=\(kubeconfigPath)" - private static let kubectlPath = "/bin/kubectl" - private static let kubeadmPath = "/usr/bin/kubeadm" - private static let ignorePreflightErrors = + public static let nodeImage = "docker.io/kindest/node:v1.35.5@sha256:ce977ae6d65918d0b58a5f8b5e940429c2ce42fa3a5619ec2bbc60b949c0ac95" + static let kubeconfigPath = "/etc/kubernetes/admin.conf" + static let kubeconfigEnv = "KUBECONFIG=/etc/kubernetes/admin.conf" + static let kubectlPath = "/bin/kubectl" + public static let kubeadmPath = "/usr/bin/kubeadm" + public static let ignorePreflightErrors = "Swap,SystemVerification,FileContent--proc-sys-net-bridge-bridge-nf-call-iptables" - private static let podSubnet = "10.244.0.0/16" + static let podSubnet = "10.244.0.0/16" // kubeadm default service subnet; must stay in sync if ClusterConfiguration.serviceSubnet is ever set. - private static let serviceSubnet = "10.96.0.0/12" + static let serviceSubnet = "10.96.0.0/12" // Proxy env var names forwarded from the host into the cluster container. static let proxyEnvVars = ["HTTP_PROXY", "HTTPS_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "no_proxy"] - // Returns proxy env vars with NO_PROXY augmented to bypass internal cluster CIDRs. - // Without this, kubelet routes apiserver traffic through the host proxy and times out. - static func nodeProxyEnv() -> [String] { - let bypassCIDRs = "192.168.0.0/16,\(podSubnet),\(serviceSubnet)" - let hostEnv = ProcessInfo.processInfo.environment - return proxyEnvVars.map { name in - guard name.uppercased() == "NO_PROXY" else { return name } - let existing = hostEnv[name] ?? hostEnv[name == "NO_PROXY" ? "no_proxy" : "NO_PROXY"] ?? "" - let augmented = existing.isEmpty ? bypassCIDRs : "\(existing),\(bypassCIDRs)" - return "\(name)=\(augmented)" - } - } - - private static let clusterContainerPort: UInt16 = 6443 - private static let clusterHostPortBase: UInt16 = 6445 - - private static func findAvailableHostPort(excluding: Set = []) throws -> UInt16 { - var port = clusterHostPortBase - while port < UInt16.max { - if excluding.contains(port) { - port += 1 - continue - } - let sock = Darwin.socket(AF_INET, SOCK_STREAM, 0) - guard sock >= 0 else { - throw ContainerizationError(.internalError, message: "socket() failed while probing for available port") - } - var addr = sockaddr_in() - addr.sin_family = sa_family_t(AF_INET) - addr.sin_port = port.bigEndian - addr.sin_addr.s_addr = INADDR_ANY - let available = withUnsafePointer(to: &addr) { ptr in - ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { - Darwin.bind(sock, $0, socklen_t(MemoryLayout.size)) == 0 - } - } - Darwin.close(sock) - if available { return port } - port += 1 - } - throw ContainerizationError(.internalError, message: "no available host port found above \(clusterHostPortBase)") - } - - static func clusterPort() async throws -> String { - let snapshots = try await ContainerClient().list( - filters: ContainerListFilters(labels: [ResourceLabelKeys.plugin: pluginName]) - ) - let reserved = Set(snapshots.flatMap { $0.configuration.publishedPorts.map(\.hostPort) }) - let port = try findAvailableHostPort(excluding: reserved) - return "\(port):\(clusterContainerPort)" - } - - private static let kubeconfigDir: FilePath = FilePath( - FileManager.default.homeDirectoryForCurrentUser.path(percentEncoded: false) - ).appending(".kube") + public static let clusterContainerPort: UInt16 = 6443 // MARK: - Resource defaults - static func defaultedResourceFlags(_ flags: Flags.Resource) -> Flags.Resource { + public static func defaultedResourceFlags(_ flags: Flags.Resource) -> Flags.Resource { var f = flags if f.cpus == nil { f.cpus = defaultCPUs } if f.memory == nil { f.memory = defaultMemory } return f } - // MARK: - Image management - - static func ensureImage(nodeImage: String = K8sHelper.nodeImage, log: Logger, containerSystemConfig: ContainerSystemConfig) async throws { - do { - _ = try await ClientImage.get(reference: nodeImage, containerSystemConfig: containerSystemConfig) - log.debug("k8s node image present", metadata: ["ref": "\(nodeImage)"]) - return - } catch let error as ContainerizationError where error.code == .notFound { - log.info("Pulling k8s node image", metadata: ["ref": "\(nodeImage)"]) - } - let platform = try Platform(from: "linux/\(Arch.hostArchitecture().rawValue)") - _ = try await ClientImage.fetch( - reference: nodeImage, - platform: platform, - scheme: .https, - containerSystemConfig: containerSystemConfig, - progressUpdate: nil) - } - - // MARK: - Node bootstrap - - private static func execCapture( + // Shared exec helper used by bootstrap, readiness, and kubeconfig extensions. + public static func execCapture( containerId: String, executable: String, arguments: [String], client: ContainerClient ) async throws -> (code: Int32, output: String) { @@ -209,418 +80,6 @@ struct K8sHelper { return (code, String(data: data, encoding: .utf8) ?? "") } - static func prepareNode(nodeID: String, client: ContainerClient, log: Logger) async throws { - log.info("Preparing node", metadata: ["id": "\(nodeID)"]) - let result = try await execCapture( - containerId: nodeID, executable: "/bin/sh", - arguments: ["-c", nodePrepScript], client: client) - guard result.code == 0 else { - throw ContainerizationError(.internalError, message: "node prep failed on \(nodeID): \(result.output)") - } - } - - static func bootstrapControlPlane( - nodeID: String, apiServerSANs: [String], advertiseAddress: String, - client: ContainerClient, log: Logger - ) async throws { - let configYAML = initConfigYAML(advertiseAddress: advertiseAddress, certSANs: apiServerSANs) - var r = try await execCapture( - containerId: nodeID, executable: "/bin/sh", - arguments: ["-c", "cat > /etc/kubernetes/kubeadm-config.yaml <<'EOF'\n\(configYAML)\nEOF"], - client: client) - guard r.code == 0 else { - throw ContainerizationError(.internalError, message: "write kubeadm config failed on \(nodeID): \(r.output)") - } - - log.info("Running kubeadm init", metadata: ["node": "\(nodeID)"]) - r = try await execCapture( - containerId: nodeID, executable: kubeadmPath, - arguments: [ - "init", "--config", "/etc/kubernetes/kubeadm-config.yaml", - "--ignore-preflight-errors", ignorePreflightErrors, - ], - client: client) - guard r.code == 0 else { - throw ContainerizationError(.internalError, message: "kubeadm init failed on \(nodeID): \(r.output)") - } - - r = try await execCapture( - containerId: nodeID, executable: "/bin/sh", - arguments: ["-c", "mkdir -p /root/.kube && cp \(kubeconfigPath) /root/.kube/config"], - client: client) - guard r.code == 0 else { - throw ContainerizationError(.internalError, message: "failed to install root kubeconfig on \(nodeID): \(r.output)") - } - - log.info("Removing control-plane taint for single-node scheduling", metadata: ["node": "\(nodeID)"]) - _ = try await runProbe( - client: client, containerId: nodeID, - arguments: ["taint", "nodes", "--all", "node-role.kubernetes.io/control-plane-"]) - - log.info("Applying kindnet CNI", metadata: ["node": "\(nodeID)"]) - let manifest = try await loadKindnetManifest(log: log) - let apply = - "cat > /tmp/kindnet.yaml <<'EOF'\n\(manifest)\nEOF\n" - + "\(kubeconfigEnv) kubectl apply -f /tmp/kindnet.yaml" - r = try await execCapture( - containerId: nodeID, executable: "/bin/sh", - arguments: ["-c", apply], client: client) - guard r.code == 0 else { - throw ContainerizationError(.internalError, message: "apply CNI failed on \(nodeID): \(r.output)") - } - } - - private static func loadKindnetManifest(log: Logger) async throws -> String { - let pluginLoader = try await makePluginLoader(log: log) - guard let plugin = pluginLoader.findPlugin(forExecutable: CommandLine.executablePath), - let resourceURL = plugin.resourceURL - else { - throw ContainerizationError(.internalError, message: "unable to locate k8s plugin installation or resources") - } - let url = resourceURL.appendingPathComponent("kindnet.yaml") - guard let contents = try? String(contentsOf: url, encoding: .utf8) else { - throw ContainerizationError(.internalError, message: "kindnet manifest resource missing at \(url.path)") - } - return contents - } - - /// NOTE: This duplicates `Application.createPluginLoader()` in - /// Sources/ContainerCommands/Application.swift. `ContainerK8s` cannot depend on - /// `ContainerCommands`, so the plugin directory/factory list here is kept in sync - /// by hand — if that logic changes, update this copy too (or factor a shared - /// constructor into `ContainerPlugin`). - private static func makePluginLoader(log: Logger) async throws -> PluginLoader { - let health = try await ClientHealthCheck.ping(timeout: .seconds(10)) - - let installRootPath = FilePath(health.installRoot.path(percentEncoded: false)) - let userPluginsURL = PluginLoader.userPluginsDir(installRoot: health.installRoot) - var directoryExists: ObjCBool = false - _ = FileManager.default.fileExists(atPath: userPluginsURL.path, isDirectory: &directoryExists) - - let appBundlePluginsURL = Bundle.main.resourceURL?.appending(path: "plugins") - let installRootPluginsPath = - installRootPath - .appending(FilePath.Component("libexec")) - .appending(FilePath.Component("container")) - .appending(FilePath.Component("plugins")) - let installRootPluginsURL = URL(fileURLWithPath: installRootPluginsPath.string) - - let pluginDirectories = [ - directoryExists.boolValue ? userPluginsURL : nil, - appBundlePluginsURL, - installRootPluginsURL, - ].compactMap { $0 } - - return try PluginLoader( - appRoot: health.appRoot, - installRoot: health.installRoot, - logRoot: health.logRoot, - pluginDirectories: pluginDirectories, - pluginFactories: [ - DefaultPluginFactory(logger: log), - AppBundlePluginFactory(logger: log), - ], - log: log - ) - } - - private static let nodePrepScript: String = { - """ - set -e - mkdir -p /etc/containerd/conf.d - cat > /etc/containerd/conf.d/native-snapshotter.toml <<'EOF' - [plugins.'io.containerd.cri.v1.images'] - snapshotter = "native" - EOF - sysctl -w net.ipv4.ip_forward=1 2>/dev/null || true - sysctl -w net.bridge.bridge-nf-call-iptables=1 2>/dev/null || true - sysctl -w net.bridge.bridge-nf-call-ip6tables=1 2>/dev/null || true - systemctl restart containerd - ctr -n k8s.io images tag registry.k8s.io/pause:3.10 registry.k8s.io/pause:3.10.1 2>/dev/null || true - /usr/sbin/iptables-nft -t mangle -A OUTPUT -p tcp --tcp-flags SYN,RST SYN -j TCPMSS --set-mss 1220 - /usr/sbin/iptables-nft -t mangle -A FORWARD -p tcp --tcp-flags SYN,RST SYN -j TCPMSS --set-mss 1220 - """ - }() - - private static func initConfigYAML(advertiseAddress: String, certSANs: [String]) -> String { - let sans = certSANs.map { " - \($0)" }.joined(separator: "\n") - return """ - apiVersion: kubeadm.k8s.io/v1beta4 - kind: InitConfiguration - localAPIEndpoint: - advertiseAddress: \(advertiseAddress) - bindPort: 6443 - nodeRegistration: - criSocket: unix:///run/containerd/containerd.sock - --- - apiVersion: kubeadm.k8s.io/v1beta4 - kind: ClusterConfiguration - kubernetesVersion: \(kubernetesVersion()) - networking: - podSubnet: \(podSubnet) - apiServer: - certSANs: - \(sans) - --- - apiVersion: kubelet.config.k8s.io/v1beta1 - kind: KubeletConfiguration - cgroupDriver: systemd - failSwapOn: false - """ - } - - private static func kubernetesVersion() -> String { - let nameAndTag = nodeImage.split(separator: "@").first.map(String.init) ?? nodeImage - guard let ref = try? Reference.parse(nameAndTag), let tag = ref.tag else { return "v1.35" } - return tag - } - - // MARK: - FQDN detection - - static func fqdn(for name: String, domain: String?) -> String? { - if name.contains(".") { return name } - guard let domain, !domain.isEmpty else { return nil } - return "\(name).\(domain)" - } - - static func detectFQDN(name: String) async -> String? { - let domain = try? await ConfigurationLoader.load().dns.domain - return fqdn(for: name, domain: domain) - } - - // MARK: - Readiness - - static func waitForNodeBooted(containerId: String, client: ContainerClient, log: Logger) async throws { - let timeout = 120 - log.info("Waiting for node to boot", metadata: ["node": "\(containerId)"]) - for attempt in 1...timeout { - let result = try await execCapture( - containerId: containerId, executable: "/bin/sh", - arguments: ["-c", "test -S /run/containerd/containerd.sock"], client: client) - if result.code == 0 { return } - if attempt == timeout { - log.info("check container logs with 'container logs \(containerId)'") - throw ContainerizationError( - .timeout, - message: "node \(containerId) did not boot within \(timeout * 2)s: containerd socket not present at /run/containerd/containerd.sock" - ) - } - try await Task.sleep(for: .seconds(2)) - } - } - - private static func runProbe(client: ContainerClient, containerId: String, arguments: [String]) async throws -> Int32 { - let devNull = FileHandle(forWritingAtPath: "/dev/null") - defer { try? devNull?.close() } - let probe = ProcessConfiguration( - executable: kubectlPath, - arguments: arguments, - environment: [kubeconfigEnv], - terminal: false) - let proc = try await client.createProcess( - containerId: containerId, - processId: UUID().uuidString.lowercased(), - configuration: probe, - stdio: [nil, devNull, devNull]) - try await proc.start() - return try await proc.wait() - } - - static func waitForReady(containerId: String, client: ContainerClient, log: Logger) async throws { - let nodeReadyTimeout = 180 - let podReadyTimeout = 300 - - log.info("Waiting for control-plane node to become ready") - for attempt in 1...nodeReadyTimeout { - let code: Int32 - do { - code = try await runProbe( - client: client, containerId: containerId, - arguments: ["wait", "--for=condition=Ready", "node", "--all", "--timeout=2s"]) - } catch { - throw ContainerizationError( - .internalError, message: "k8s cluster \(containerId) stopped unexpectedly during startup: \(error)") - } - if code == 0 { break } - if attempt == nodeReadyTimeout { - log.info("inspect node state with 'container exec \(containerId) kubectl get nodes -o wide'") - throw ContainerizationError( - .timeout, - message: "k8s cluster \(containerId) control-plane node did not become Ready within \(nodeReadyTimeout * 2)s" - ) - } - try await Task.sleep(for: .seconds(2)) - } - - log.info("Waiting for kube-system pods to become ready") - for attempt in 1...podReadyTimeout { - let code: Int32 - do { - code = try await runProbe( - client: client, containerId: containerId, - arguments: ["wait", "--for=condition=Available", "deployment/coredns", "-n", "kube-system", "--timeout=2s"]) - } catch { - throw ContainerizationError( - .internalError, message: "k8s cluster \(containerId) stopped unexpectedly during startup: \(error)") - } - if code == 0 { return } - if attempt < podReadyTimeout { - try await Task.sleep(for: .seconds(2)) - } - } - log.info("inspect pod state with 'container exec \(containerId) kubectl get pods -n kube-system'") - throw ContainerizationError( - .timeout, - message: "k8s cluster \(containerId) kube-system pods did not become available within \(podReadyTimeout * 2)s" - ) - } - - // MARK: - Kubeconfig - - static func fetchConfig(containerId: String, client: ContainerClient, log: Logger) async throws -> KubeConfig { - log.info("Fetching kubeconfig", metadata: ["cluster": "\(containerId)"]) - - let container = try await client.get(id: containerId) - guard container.configuration.labels[ResourceLabelKeys.plugin] == pluginName else { - log.error("container is not a k8s cluster, refusing config fetch", metadata: ["name": "\(containerId)"]) - throw ContainerizationError(.invalidArgument, message: "\(containerId) is not a k8s cluster") - } - - let (exitCode, yaml) = try await execCapture( - containerId: containerId, executable: "/bin/cat", - arguments: [kubeconfigPath], client: client) - guard exitCode == 0 else { - throw ContainerizationError(.internalError, message: "failed to read kubeconfig from \(containerId): exit \(exitCode)") - } - do { - return try YAMLDecoder().decode(KubeConfig.self, from: yaml) - } catch { - throw ContainerizationError(.internalError, message: "failed to decode kubeconfig from \(containerId): \(error)") - } - } - - static func transformConfig(_ config: KubeConfig, containerId: String, fqdn: String?, client: ContainerClient) async throws -> KubeConfig { - var config = config - let serverAddress: String - if let fqdn { - serverAddress = "https://\(fqdn):6443" - } else { - let snapshot = try await client.get(id: containerId) - guard - let hostPort = snapshot.configuration.publishedPorts - .first(where: { $0.containerPort == clusterContainerPort })?.hostPort - else { - throw ContainerizationError(.internalError, message: "no published port for cluster \(containerId)") - } - serverAddress = "https://127.0.0.1:\(hostPort)" - } - for i in config.clusters.indices { - config.clusters[i].cluster.server = serverAddress - } - // Rename all entries to containerId. kubeadm uses fixed names ("kubernetes", - // "kubernetes-admin@kubernetes", etc.) rather than "default", so we rename - // unconditionally and fix up the cross-references in the context. - config.clusters = config.clusters.map { - var c = $0 - c.name = containerId - return c - } - config.users = config.users.map { - var u = $0 - u.name = containerId - return u - } - config.contexts = config.contexts.map { - var nc = $0 - nc.name = containerId - nc.context.cluster = containerId - nc.context.user = containerId - return nc - } - config.currentContext = containerId - return config - } - - static func resolveKubeconfigMergePath() -> FilePath { - let defaultPath = kubeconfigDir.appending("config") - guard let raw = Darwin.getenv("KUBECONFIG") else { return defaultPath } - let env = String(cString: raw) - guard !env.isEmpty else { return defaultPath } - let paths = env.split(separator: ":").map(String.init).filter { !$0.isEmpty } - guard !paths.isEmpty else { return defaultPath } - if paths.count == 1 { return FilePath(paths[0]) } - for p in paths where FileManager.default.fileExists(atPath: p) { - return FilePath(p) - } - return FilePath(paths[paths.count - 1]) - } - - static func mergeConfig(_ config: KubeConfig, containerId: String, targetPath: FilePath? = nil, setCurrentContext: Bool = false, log: Logger) throws { - let path = targetPath ?? resolveKubeconfigMergePath() - log.info("Writing kubeconfig", metadata: ["cluster": "\(containerId)", "path": "\(path)"]) - - let targetDir = path.removingLastComponent() - try FileManager.default.createDirectory(atPath: targetDir.string, withIntermediateDirectories: true) - - var existing: KubeConfig - if FileManager.default.fileExists(atPath: path.string) { - do { - let yaml = try String(contentsOfFile: path.string, encoding: .utf8) - existing = try YAMLDecoder().decode(KubeConfig.self, from: yaml) - } catch { - throw ContainerizationError( - .internalError, - message: "kubeconfig at \(path) could not be parsed: \(error)" - ) - } - } else { - existing = .empty - } - - existing.clusters.removeAll { $0.name == containerId } - existing.contexts.removeAll { $0.name == containerId } - existing.users.removeAll { $0.name == containerId } - - existing.clusters.append(contentsOf: config.clusters) - existing.contexts.append(contentsOf: config.contexts) - existing.users.append(contentsOf: config.users) - if setCurrentContext { - existing.currentContext = containerId - } - - let output = try YAMLEncoder().encode(existing) - try output.write(toFile: path.string, atomically: true, encoding: .utf8) - } - - static func removeConfig(containerId: String, log: Logger) throws { - let path = resolveKubeconfigMergePath() - log.info("Removing kubeconfig", metadata: ["cluster": "\(containerId)", "path": "\(path)"]) - - guard FileManager.default.fileExists(atPath: path.string) else { return } - - let existing: KubeConfig - do { - let yaml = try String(contentsOfFile: path.string, encoding: .utf8) - existing = try YAMLDecoder().decode(KubeConfig.self, from: yaml) - } catch { - log.warning("kubeconfig exists but could not be parsed, skipping removal", metadata: ["path": "\(path)", "error": "\(error)"]) - return - } - - var updated = existing - - updated.clusters.removeAll { $0.name == containerId } - updated.contexts.removeAll { $0.name == containerId } - updated.users.removeAll { $0.name == containerId } - - if updated.currentContext == containerId { - updated.currentContext = nil - } - - let output = try YAMLEncoder().encode(updated) - try output.write(toFile: path.string, atomically: true, encoding: .utf8) - } - // MARK: - List rows static func buildK8sRows(from snapshots: [ContainerSnapshot]) -> [K8sNodeResource] { @@ -669,19 +128,6 @@ struct K8sHelper { } return TableOutput(rows: rows).format() } - - // MARK: - Image reference helpers - - static func isShortName(_ reference: String) -> Bool { - guard let ref = try? Reference.parse(reference) else { return true } - return ref.domain == nil - } - - static func fqReference(_ reference: String) -> String { - guard let ref = try? Reference.parse(reference) else { return reference } - ref.normalize() - return ref.description - } } // MARK: - K8sNodeResource diff --git a/Sources/ContainerK8s/Provisioners/LinuxNodeProvisioner.swift b/Sources/ContainerK8s/Provisioners/LinuxNodeProvisioner.swift new file mode 100644 index 000000000..c50f7d02b --- /dev/null +++ b/Sources/ContainerK8s/Provisioners/LinuxNodeProvisioner.swift @@ -0,0 +1,229 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerPersistence +import ContainerResource +import ContainerizationError +import Logging + +/// A `NodeProvisioner` that runs a k8s node as a Linux container on the same host. +public struct LinuxNodeProvisioner: NodeProvisioner { + public let roles: [String] + + private let clusterName: String + private let nodeImage: String? + private let cpus: Int64? + private let memory: String? + private let registryScheme: String + private let maxConcurrentDownloads: Int + private let remove: Bool + private let fqdn: String? + + public init( + clusterName: String, + roles: [String] = [StandardRoles.controlPlane], + nodeImage: String? = nil, + cpus: Int64? = nil, + memory: String? = nil, + registryScheme: String = "https", + maxConcurrentDownloads: Int = 3, + remove: Bool = false, + fqdn: String? = nil + ) throws { + guard !roles.isEmpty else { + throw ContainerizationError(.invalidArgument, message: "LinuxNode roles must not be empty") + } + let known: Set = [StandardRoles.controlPlane, StandardRoles.worker] + let unrecognized = roles.filter { !known.contains($0) } + guard unrecognized.isEmpty else { + throw ContainerizationError( + .invalidArgument, + message: "LinuxNode roles contains unrecognized values: \(unrecognized.joined(separator: ", "))") + } + self.clusterName = clusterName + self.roles = roles + self.nodeImage = nodeImage + self.cpus = cpus + self.memory = memory + self.registryScheme = registryScheme + self.maxConcurrentDownloads = maxConcurrentDownloads + self.remove = remove + self.fqdn = fqdn + } + + public func provision(name: String, log: Logger) async throws { + let containerSystemConfig: ContainerSystemConfig = try await ConfigurationLoader.load() + let resolvedImage = nodeImage ?? K8sHelper.nodeImage + try await K8sHelper.ensureImage(nodeImage: resolvedImage, log: log, containerSystemConfig: containerSystemConfig) + + let isControlPlane = roles.contains(StandardRoles.controlPlane) + let publishPorts = isControlPlane && fqdn == nil ? [try await K8sHelper.clusterPort()] : [] + + let management = Flags.Management( + arch: Arch.hostArchitecture().rawValue, + capAdd: ["ALL"], + capDrop: [], + cidfile: "", + detach: true, + dns: .init(domain: nil, nameservers: [], options: [], searchDomains: []), + dnsDisabled: false, + entrypoint: nil, + initImage: nil, + kernel: nil, + kernelArgs: [], + labels: [ + "\(ResourceLabelKeys.plugin)=\(K8sHelper.pluginName)", + "\(ResourceLabelKeys.role)=\(roles.joined(separator: ","))", + ], + maskedPaths: [], + mounts: [], + name: name, + networks: [], + os: "linux", + platform: nil, + publishPorts: publishPorts, + publishSockets: [], + readOnly: false, + readonlyPaths: [], + remove: remove, + rosetta: true, + runtime: nil, + ssh: false, + shmSize: nil, + tmpFs: [], + useInit: false, + virtualization: false, + volumes: [] + ) + + let updatedResource = K8sHelper.defaultedResourceFlags(Flags.Resource(cpus: cpus, memory: memory)) + let processFlags = Flags.Process( + cwd: nil, + env: K8sHelper.nodeProxyEnv(), + envFile: [], + gid: nil, + interactive: false, + tty: false, + uid: nil, + ulimits: [], + user: nil + ) + + var (config, kernel, initfs) = try await Utility.containerConfigFromFlags( + id: name, + image: resolvedImage, + arguments: [], + process: processFlags, + management: management, + resource: updatedResource, + registry: Flags.Registry(scheme: registryScheme), + imageFetch: Flags.ImageFetch(maxConcurrentDownloads: maxConcurrentDownloads), + containerSystemConfig: containerSystemConfig, + progressUpdate: { _ in }, + log: log + ) + config.maskedPaths = [] + config.readonlyPaths = [] + + let client = ContainerClient() + try await client.create( + configuration: config, + options: ContainerCreateOptions(autoRemove: remove), + kernel: kernel, + initImage: initfs + ) + + 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() + + try await K8sHelper.waitForNodeBooted(containerId: name, client: client, log: log) + } + + public func address(name: String, log: Logger) async throws -> String { + let client = ContainerClient() + let snapshot = try await client.get(id: name) + guard let ip = snapshot.networks.first?.ipv4Address.address.description else { + throw ContainerizationError(.internalError, message: "no VM IP for node \(name)") + } + return ip + } + + public func join( + name: String, + controlPlaneEndpoint: String, + token: String, + caCertHash: String, + log: Logger + ) async throws { + let client = ContainerClient() + try await K8sHelper.prepareNode(nodeID: name, client: client, log: log) + + let (code, output) = try await K8sHelper.execCapture( + containerId: name, + executable: K8sHelper.kubeadmPath, + arguments: [ + "join", controlPlaneEndpoint, + "--token", token, + "--discovery-token-ca-cert-hash", caCertHash, + "--ignore-preflight-errors", K8sHelper.ignorePreflightErrors, + "--cri-socket", "unix:///run/containerd/containerd.sock", + ], + client: client) + guard code == 0 else { + throw ContainerizationError(.internalError, message: "kubeadm join failed on \(name): \(output)") + } + } + + public func waitForReady(name: String, log: Logger) async throws { + let timeout = 180 + let client = ContainerClient() + log.info("Waiting for node to become ready", metadata: ["node": "\(name)"]) + for attempt in 1...timeout { + let code: Int32 + do { + code = try await K8sHelper.runProbe( + client: client, + containerId: clusterName, + arguments: ["wait", "--for=condition=Ready", "node/\(name)", "--timeout=2s"]) + } catch { + throw ContainerizationError( + .internalError, + message: "node \(name) stopped unexpectedly while waiting for readiness: \(error)") + } + if code == 0 { return } + if attempt == timeout { + throw ContainerizationError( + .timeout, + message: "node \(name) did not become Ready within \(timeout * 2)s") + } + try await Task.sleep(for: .seconds(2)) + } + } + + public func teardown(name: String, log: Logger) async throws { + let client = ContainerClient() + do { + try? await client.stop(id: name) + try await client.delete(id: name) + } catch let error as ContainerizationError where error.code == .notFound { + log.debug("node container not found, skipping delete", metadata: ["name": "\(name)"]) + } + } +} diff --git a/Sources/ContainerK8s/Provisioners/NodeProvisioner.swift b/Sources/ContainerK8s/Provisioners/NodeProvisioner.swift new file mode 100644 index 000000000..706910824 --- /dev/null +++ b/Sources/ContainerK8s/Provisioners/NodeProvisioner.swift @@ -0,0 +1,65 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import Logging + +/// Well-known role identifiers for use in `NodeProvisioner.roles`. +public struct StandardRoles { + public static let controlPlane = "control-plane" + public static let worker = "worker" +} + +/// Manages the lifecycle of a cluster node (control plane or worker). +/// +/// Implement this protocol to provision nodes for a cluster created by `K8sCreate`. +/// +/// For a **control-plane** node, `K8sCreate` calls: +/// 1. `provision` — start the machine before kubeadm init runs +/// 2. `address` — return the node IP, used as the advertise address and cert SAN +/// 3. `teardown` — called on failure to clean up the node +/// +/// For a **worker** node, the caller additionally calls: +/// 4. `join` — run kubeadm join with the bootstrap token and CA cert hash +/// 5. `waitForReady` — poll until the node is registered and Ready in the cluster +/// +/// `K8sDelete` calls `teardown` before removing cluster containers. +/// +/// If any provisioner step throws, `K8sCreate` tears down the cluster before re-throwing. +public protocol NodeProvisioner: Sendable { + /// The roles this node will serve (e.g. `[StandardRoles.controlPlane]`). + var roles: [String] { get } + + /// Start the machine identified by `name` before cluster initialisation. + func provision(name: String, log: Logger) async throws + + /// Return the IP address of the machine identified by `name`. + func address(name: String, log: Logger) async throws -> String + + /// Join a worker node to the cluster using the supplied kubeadm credentials. + /// + /// - Parameters: + /// - name: Node identifier (matches the name passed to `provision`). + /// - controlPlaneEndpoint: `:` of the control-plane API server. + /// - token: kubeadm bootstrap token (format `.`). + /// - caCertHash: Discovery token CA cert hash including the `sha256:` prefix. + func join(name: String, controlPlaneEndpoint: String, token: String, caCertHash: String, log: Logger) async throws + + /// Poll until the node with `name` is registered and Ready in the cluster. + func waitForReady(name: String, log: Logger) async throws + + /// Remove the machine identified by `name`. + func teardown(name: String, log: Logger) async throws +} diff --git a/Sources/ContainerK8s/Support/K8sHelper+Bootstrap.swift b/Sources/ContainerK8s/Support/K8sHelper+Bootstrap.swift new file mode 100644 index 000000000..964f9b72f --- /dev/null +++ b/Sources/ContainerK8s/Support/K8sHelper+Bootstrap.swift @@ -0,0 +1,170 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerPlugin +import ContainerizationError +import ContainerizationOCI +import Foundation +import Logging + +extension K8sHelper { + + public static func prepareNode(nodeID: String, client: ContainerClient, log: Logger) async throws { + log.info("Preparing node", metadata: ["id": "\(nodeID)"]) + let result = try await execCapture( + containerId: nodeID, executable: "/bin/sh", + arguments: ["-c", nodePrepScript], client: client) + guard result.code == 0 else { + throw ContainerizationError(.internalError, message: "node prep failed on \(nodeID): \(result.output)") + } + } + + static func bootstrapControlPlane( + nodeID: String, apiServerSANs: [String], advertiseAddress: String, + schedulable: Bool, client: ContainerClient, log: Logger + ) async throws { + let configYAML = initConfigYAML(advertiseAddress: advertiseAddress, certSANs: apiServerSANs) + var r = try await execCapture( + containerId: nodeID, executable: "/bin/sh", + arguments: ["-c", "cat > /etc/kubernetes/kubeadm-config.yaml <<'EOF'\n\(configYAML)\nEOF"], + client: client) + guard r.code == 0 else { + throw ContainerizationError(.internalError, message: "write kubeadm config failed on \(nodeID): \(r.output)") + } + + log.info("Running kubeadm init", metadata: ["node": "\(nodeID)"]) + r = try await execCapture( + containerId: nodeID, executable: kubeadmPath, + arguments: [ + "init", "--config", "/etc/kubernetes/kubeadm-config.yaml", + "--ignore-preflight-errors", ignorePreflightErrors, + ], + client: client) + guard r.code == 0 else { + throw ContainerizationError(.internalError, message: "kubeadm init failed on \(nodeID): \(r.output)") + } + + r = try await execCapture( + containerId: nodeID, executable: "/bin/sh", + arguments: ["-c", "mkdir -p /root/.kube && cp \(kubeconfigPath) /root/.kube/config"], + client: client) + guard r.code == 0 else { + throw ContainerizationError(.internalError, message: "failed to install root kubeconfig on \(nodeID): \(r.output)") + } + + if schedulable { + log.info("Removing control-plane taint for single-node scheduling", metadata: ["node": "\(nodeID)"]) + _ = try await runProbe( + client: client, containerId: nodeID, + arguments: ["taint", "nodes", "--all", "node-role.kubernetes.io/control-plane-"]) + } + + log.info("Applying kindnet CNI", metadata: ["node": "\(nodeID)"]) + let manifest = try await loadKindnetManifest(log: log) + let apply = + "cat > /tmp/kindnet.yaml <<'EOF'\n\(manifest)\nEOF\n" + + "\(kubeconfigEnv) kubectl apply -f /tmp/kindnet.yaml" + r = try await execCapture( + containerId: nodeID, executable: "/bin/sh", + arguments: ["-c", apply], client: client) + guard r.code == 0 else { + throw ContainerizationError(.internalError, message: "apply CNI failed on \(nodeID): \(r.output)") + } + } + + static func createJoinToken(nodeID: String, client: ContainerClient) async throws -> (token: String, caCertHash: String) { + let (code, output) = try await execCapture( + containerId: nodeID, executable: kubeadmPath, + arguments: ["token", "create", "--print-join-command"], + client: client) + guard code == 0 else { + throw ContainerizationError(.internalError, message: "kubeadm token create failed on \(nodeID): \(output)") + } + let parts = output.trimmingCharacters(in: .whitespacesAndNewlines).components(separatedBy: " ") + guard let tokenIdx = parts.firstIndex(of: "--token"), tokenIdx + 1 < parts.count, + let hashIdx = parts.firstIndex(of: "--discovery-token-ca-cert-hash"), hashIdx + 1 < parts.count + else { + throw ContainerizationError(.internalError, message: "could not parse join command output from kubeadm on \(nodeID)") + } + return (token: parts[tokenIdx + 1], caCertHash: parts[hashIdx + 1]) + } + + private static func loadKindnetManifest(log: Logger) async throws -> String { + let pluginLoader = try await Utility.createPluginLoader(log: log) + guard let plugin = pluginLoader.findPlugin(forExecutable: CommandLine.executablePath), + let resourceURL = plugin.resourceURL + else { + throw ContainerizationError(.internalError, message: "unable to locate k8s plugin installation or resources") + } + let url = resourceURL.appendingPathComponent("kindnet.yaml") + guard let contents = try? String(contentsOf: url, encoding: .utf8) else { + throw ContainerizationError(.internalError, message: "kindnet manifest resource missing at \(url.path)") + } + return contents + } + + private static var nodePrepScript: String { + """ + set -e + mkdir -p /etc/containerd/conf.d + cat > /etc/containerd/conf.d/native-snapshotter.toml <<'EOF' + [plugins.'io.containerd.cri.v1.images'] + snapshotter = "native" + EOF + sysctl -w net.ipv4.ip_forward=1 2>/dev/null || true + sysctl -w net.bridge.bridge-nf-call-iptables=1 2>/dev/null || true + sysctl -w net.bridge.bridge-nf-call-ip6tables=1 2>/dev/null || true + systemctl restart containerd + ctr -n k8s.io images tag registry.k8s.io/pause:3.10 registry.k8s.io/pause:3.10.1 2>/dev/null || true + /usr/sbin/iptables-nft -t mangle -A OUTPUT -p tcp --tcp-flags SYN,RST SYN -j TCPMSS --set-mss 1220 + /usr/sbin/iptables-nft -t mangle -A FORWARD -p tcp --tcp-flags SYN,RST SYN -j TCPMSS --set-mss 1220 + """ + } + + private static func initConfigYAML(advertiseAddress: String, certSANs: [String]) -> String { + let sans = certSANs.map { " - \($0)" }.joined(separator: "\n") + return """ + apiVersion: kubeadm.k8s.io/v1beta4 + kind: InitConfiguration + localAPIEndpoint: + advertiseAddress: \(advertiseAddress) + bindPort: 6443 + nodeRegistration: + criSocket: unix:///run/containerd/containerd.sock + --- + apiVersion: kubeadm.k8s.io/v1beta4 + kind: ClusterConfiguration + kubernetesVersion: \(kubernetesVersion()) + networking: + podSubnet: \(podSubnet) + apiServer: + certSANs: + \(sans) + --- + apiVersion: kubelet.config.k8s.io/v1beta1 + kind: KubeletConfiguration + cgroupDriver: systemd + failSwapOn: false + """ + } + + private static func kubernetesVersion() -> String { + let nameAndTag = nodeImage.split(separator: "@").first.map(String.init) ?? nodeImage + guard let ref = try? Reference.parse(nameAndTag), let tag = ref.tag else { return "v1.35" } + return tag + } +} diff --git a/Sources/ContainerK8s/Support/K8sHelper+Image.swift b/Sources/ContainerK8s/Support/K8sHelper+Image.swift new file mode 100644 index 000000000..3574561c7 --- /dev/null +++ b/Sources/ContainerK8s/Support/K8sHelper+Image.swift @@ -0,0 +1,54 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerPersistence +import ContainerizationError +import ContainerizationOCI +import Logging + +extension K8sHelper { + // MARK: - Image management + + public static func ensureImage(nodeImage: String = K8sHelper.nodeImage, log: Logger, containerSystemConfig: ContainerSystemConfig) async throws { + do { + _ = try await ClientImage.get(reference: nodeImage, containerSystemConfig: containerSystemConfig) + log.debug("k8s node image present", metadata: ["ref": "\(nodeImage)"]) + return + } catch let error as ContainerizationError where error.code == .notFound { + log.info("Pulling k8s node image", metadata: ["ref": "\(nodeImage)"]) + } + let platform = try Platform(from: "linux/\(Arch.hostArchitecture().rawValue)") + _ = try await ClientImage.fetch( + reference: nodeImage, + platform: platform, + containerSystemConfig: containerSystemConfig, + progressUpdate: nil) + } + + // MARK: - Image reference helpers + + static func isShortName(_ reference: String) -> Bool { + guard let ref = try? Reference.parse(reference) else { return true } + return ref.domain == nil + } + + static func fqReference(_ reference: String) -> String { + guard let ref = try? Reference.parse(reference) else { return reference } + ref.normalize() + return ref.description + } +} diff --git a/Sources/ContainerK8s/Support/K8sHelper+Kubeconfig.swift b/Sources/ContainerK8s/Support/K8sHelper+Kubeconfig.swift new file mode 100644 index 000000000..088dd2dbc --- /dev/null +++ b/Sources/ContainerK8s/Support/K8sHelper+Kubeconfig.swift @@ -0,0 +1,177 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerResource +import ContainerizationError +import Darwin +import Foundation +import Logging +import SystemPackage +import Yams + +extension K8sHelper { + // MARK: - Kubeconfig + + private static var kubeconfigDir: FilePath { + FilePath(FileManager.default.homeDirectoryForCurrentUser.path(percentEncoded: false)) + .appending(".kube") + } + + static func fetchConfig(containerId: String, client: ContainerClient, log: Logger) async throws -> KubeConfig { + log.info("Fetching kubeconfig", metadata: ["cluster": "\(containerId)"]) + + let container = try await client.get(id: containerId) + guard container.configuration.labels[ResourceLabelKeys.plugin] == pluginName else { + log.error("container is not a k8s cluster, refusing config fetch", metadata: ["name": "\(containerId)"]) + throw ContainerizationError(.invalidArgument, message: "\(containerId) is not a k8s cluster") + } + + let (exitCode, yaml) = try await execCapture( + containerId: containerId, executable: "/bin/cat", + arguments: [kubeconfigPath], client: client) + guard exitCode == 0 else { + throw ContainerizationError(.internalError, message: "failed to read kubeconfig from \(containerId): exit \(exitCode)") + } + do { + return try YAMLDecoder().decode(KubeConfig.self, from: yaml) + } catch { + throw ContainerizationError(.internalError, message: "failed to decode kubeconfig from \(containerId): \(error)") + } + } + + static func transformConfig(_ config: KubeConfig, containerId: String, fqdn: String?, client: ContainerClient) async throws -> KubeConfig { + var config = config + let serverAddress: String + if let fqdn { + serverAddress = "https://\(fqdn):6443" + } else { + let snapshot = try await client.get(id: containerId) + guard + let hostPort = snapshot.configuration.publishedPorts + .first(where: { $0.containerPort == clusterContainerPort })?.hostPort + else { + throw ContainerizationError(.internalError, message: "no published port for cluster \(containerId)") + } + serverAddress = "https://127.0.0.1:\(hostPort)" + } + for i in config.clusters.indices { + config.clusters[i].cluster.server = serverAddress + } + // Rename all entries to containerId. kubeadm uses fixed names ("kubernetes", + // "kubernetes-admin@kubernetes", etc.) rather than "default", so we rename + // unconditionally and fix up the cross-references in the context. + config.clusters = config.clusters.map { + var c = $0 + c.name = containerId + return c + } + config.users = config.users.map { + var u = $0 + u.name = containerId + return u + } + config.contexts = config.contexts.map { + var nc = $0 + nc.name = containerId + nc.context.cluster = containerId + nc.context.user = containerId + return nc + } + config.currentContext = containerId + return config + } + + static func resolveKubeconfigMergePath() -> FilePath { + let defaultPath = kubeconfigDir.appending("config") + guard let raw = Darwin.getenv("KUBECONFIG") else { return defaultPath } + let env = String(cString: raw) + guard !env.isEmpty else { return defaultPath } + let paths = env.split(separator: ":").map(String.init).filter { !$0.isEmpty } + guard !paths.isEmpty else { return defaultPath } + if paths.count == 1 { return FilePath(paths[0]) } + for p in paths where FileManager.default.fileExists(atPath: p) { + return FilePath(p) + } + return FilePath(paths[paths.count - 1]) + } + + static func mergeConfig(_ config: KubeConfig, containerId: String, targetPath: FilePath? = nil, setCurrentContext: Bool = false, log: Logger) throws { + let path = targetPath ?? resolveKubeconfigMergePath() + log.info("Writing kubeconfig", metadata: ["cluster": "\(containerId)", "path": "\(path)"]) + + let targetDir = path.removingLastComponent() + try FileManager.default.createDirectory(atPath: targetDir.string, withIntermediateDirectories: true) + + var existing: KubeConfig + if FileManager.default.fileExists(atPath: path.string) { + do { + let yaml = try String(contentsOfFile: path.string, encoding: .utf8) + existing = try YAMLDecoder().decode(KubeConfig.self, from: yaml) + } catch { + throw ContainerizationError( + .internalError, + message: "kubeconfig at \(path) could not be parsed: \(error)" + ) + } + } else { + existing = .empty + } + + existing.clusters.removeAll { $0.name == containerId } + existing.contexts.removeAll { $0.name == containerId } + existing.users.removeAll { $0.name == containerId } + + existing.clusters.append(contentsOf: config.clusters) + existing.contexts.append(contentsOf: config.contexts) + existing.users.append(contentsOf: config.users) + if setCurrentContext { + existing.currentContext = containerId + } + + let output = try YAMLEncoder().encode(existing) + try output.write(toFile: path.string, atomically: true, encoding: .utf8) + } + + static func removeConfig(containerId: String, log: Logger) throws { + let path = resolveKubeconfigMergePath() + log.info("Removing kubeconfig", metadata: ["cluster": "\(containerId)", "path": "\(path)"]) + + guard FileManager.default.fileExists(atPath: path.string) else { return } + + let existing: KubeConfig + do { + let yaml = try String(contentsOfFile: path.string, encoding: .utf8) + existing = try YAMLDecoder().decode(KubeConfig.self, from: yaml) + } catch { + log.warning("kubeconfig exists but could not be parsed, skipping removal", metadata: ["path": "\(path)", "error": "\(error)"]) + return + } + + var updated = existing + + updated.clusters.removeAll { $0.name == containerId } + updated.contexts.removeAll { $0.name == containerId } + updated.users.removeAll { $0.name == containerId } + + if updated.currentContext == containerId { + updated.currentContext = nil + } + + let output = try YAMLEncoder().encode(updated) + try output.write(toFile: path.string, atomically: true, encoding: .utf8) + } +} diff --git a/Sources/ContainerK8s/Support/K8sHelper+Networking.swift b/Sources/ContainerK8s/Support/K8sHelper+Networking.swift new file mode 100644 index 000000000..96564f95c --- /dev/null +++ b/Sources/ContainerK8s/Support/K8sHelper+Networking.swift @@ -0,0 +1,91 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerPersistence +import ContainerResource +import ContainerizationError +import Darwin +import Foundation +import Logging + +extension K8sHelper { + // MARK: - Networking + + private static var clusterHostPortBase: UInt16 { 6445 } + + // Returns proxy env vars with NO_PROXY augmented to bypass internal cluster CIDRs. + // Without this, kubelet routes apiserver traffic through the host proxy and times out. + public static func nodeProxyEnv() -> [String] { + let bypassCIDRs = "192.168.0.0/16,\(podSubnet),\(serviceSubnet)" + let hostEnv = ProcessInfo.processInfo.environment + return proxyEnvVars.map { name in + guard name.uppercased() == "NO_PROXY" else { return name } + let existing = hostEnv[name] ?? hostEnv[name == "NO_PROXY" ? "no_proxy" : "NO_PROXY"] ?? "" + let augmented = existing.isEmpty ? bypassCIDRs : "\(existing),\(bypassCIDRs)" + return "\(name)=\(augmented)" + } + } + + private static func findAvailableHostPort(excluding: Set = []) throws -> UInt16 { + var port = clusterHostPortBase + while port < UInt16.max { + if excluding.contains(port) { + port += 1 + continue + } + let sock = Darwin.socket(AF_INET, SOCK_STREAM, 0) + guard sock >= 0 else { + throw ContainerizationError(.internalError, message: "socket() failed while probing for available port") + } + var addr = sockaddr_in() + addr.sin_family = sa_family_t(AF_INET) + addr.sin_port = port.bigEndian + addr.sin_addr.s_addr = INADDR_ANY + let available = withUnsafePointer(to: &addr) { ptr in + ptr.withMemoryRebound(to: sockaddr.self, capacity: 1) { + Darwin.bind(sock, $0, socklen_t(MemoryLayout.size)) == 0 + } + } + Darwin.close(sock) + if available { return port } + port += 1 + } + throw ContainerizationError(.internalError, message: "no available host port found above \(clusterHostPortBase)") + } + + public static func clusterPort() async throws -> String { + let snapshots = try await ContainerClient().list( + filters: ContainerListFilters(labels: [ResourceLabelKeys.plugin: pluginName]) + ) + let reserved = Set(snapshots.flatMap { $0.configuration.publishedPorts.map(\.hostPort) }) + let port = try findAvailableHostPort(excluding: reserved) + return "\(port):\(clusterContainerPort)" + } + + // MARK: - FQDN detection + + static func fqdn(for name: String, domain: String?) -> String? { + if name.contains(".") { return name } + guard let domain, !domain.isEmpty else { return nil } + return "\(name).\(domain)" + } + + static func detectFQDN(name: String) async -> String? { + let domain = try? await ConfigurationLoader.load().dns.domain + return fqdn(for: name, domain: domain) + } +} diff --git a/Sources/ContainerK8s/Support/K8sHelper+Readiness.swift b/Sources/ContainerK8s/Support/K8sHelper+Readiness.swift new file mode 100644 index 000000000..c64865440 --- /dev/null +++ b/Sources/ContainerK8s/Support/K8sHelper+Readiness.swift @@ -0,0 +1,111 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import ContainerAPIClient +import ContainerResource +import ContainerizationError +import ContainerizationOS +import Foundation +import Logging + +extension K8sHelper { + // MARK: - Readiness + + public static func runProbe(client: ContainerClient, containerId: String, arguments: [String]) async throws -> Int32 { + let devNull = FileHandle(forWritingAtPath: "/dev/null") + defer { try? devNull?.close() } + let probe = ProcessConfiguration( + executable: kubectlPath, + arguments: arguments, + environment: [kubeconfigEnv], + terminal: false) + let proc = try await client.createProcess( + containerId: containerId, + processId: UUID().uuidString.lowercased(), + configuration: probe, + stdio: [nil, devNull, devNull]) + try await proc.start() + return try await proc.wait() + } + + public static func waitForNodeBooted(containerId: String, client: ContainerClient, log: Logger) async throws { + let timeout = 120 + log.info("Waiting for node to boot", metadata: ["node": "\(containerId)"]) + for attempt in 1...timeout { + let result = try await execCapture( + containerId: containerId, executable: "/bin/sh", + arguments: ["-c", "test -S /run/containerd/containerd.sock"], client: client) + if result.code == 0 { return } + if attempt == timeout { + log.info("check container logs with 'container logs \(containerId)'") + throw ContainerizationError( + .timeout, + message: "node \(containerId) did not boot within \(timeout * 2)s: containerd socket not present at /run/containerd/containerd.sock" + ) + } + try await Task.sleep(for: .seconds(2)) + } + } + + static func waitForReady(containerId: String, client: ContainerClient, log: Logger) async throws { + let nodeReadyTimeout = 180 + let podReadyTimeout = 300 + + log.info("Waiting for control-plane node to become ready") + for attempt in 1...nodeReadyTimeout { + let code: Int32 + do { + code = try await runProbe( + client: client, containerId: containerId, + arguments: ["wait", "--for=condition=Ready", "node", "--all", "--timeout=2s"]) + } catch { + throw ContainerizationError( + .internalError, message: "k8s cluster \(containerId) stopped unexpectedly during startup: \(error)") + } + if code == 0 { break } + if attempt == nodeReadyTimeout { + log.info("inspect node state with 'container exec \(containerId) kubectl get nodes -o wide'") + throw ContainerizationError( + .timeout, + message: "k8s cluster \(containerId) control-plane node did not become Ready within \(nodeReadyTimeout * 2)s" + ) + } + try await Task.sleep(for: .seconds(2)) + } + + log.info("Waiting for kube-system pods to become ready") + for attempt in 1...podReadyTimeout { + let code: Int32 + do { + code = try await runProbe( + client: client, containerId: containerId, + arguments: ["wait", "--for=condition=Available", "deployment/coredns", "-n", "kube-system", "--timeout=2s"]) + } catch { + throw ContainerizationError( + .internalError, message: "k8s cluster \(containerId) stopped unexpectedly during startup: \(error)") + } + if code == 0 { return } + if attempt < podReadyTimeout { + try await Task.sleep(for: .seconds(2)) + } + } + log.info("inspect pod state with 'container exec \(containerId) kubectl get pods -n kube-system'") + throw ContainerizationError( + .timeout, + message: "k8s cluster \(containerId) kube-system pods did not become available within \(podReadyTimeout * 2)s" + ) + } +} diff --git a/Sources/ContainerK8s/Support/TableOutput.swift b/Sources/ContainerK8s/Support/TableOutput.swift new file mode 100644 index 000000000..d5256befb --- /dev/null +++ b/Sources/ContainerK8s/Support/TableOutput.swift @@ -0,0 +1,65 @@ +//===----------------------------------------------------------------------===// +// Copyright © 2026 Apple Inc. and the container project authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +//===----------------------------------------------------------------------===// + +import Foundation + +// MARK: - ListDisplayable + +protocol ListDisplayable { + static var tableHeader: [String] { get } + var tableRow: [String] { get } + var quietValue: String { get } +} + +// MARK: - TableOutput + +struct TableOutput: Sendable { + private let rows: [[String]] + private let spacing: Int + + init(rows: [[String]], spacing: Int = 2) { + self.rows = rows + self.spacing = spacing + } + + func format() -> String { + var output = "" + let maxLengths = self.maxLength() + + for rowIndex in 0.. [Int: Int] { + var output: [Int: Int] = [:] + for row in self.rows { + for (i, column) in row.enumerated() { + let currentMax = output[i] ?? 0 + output[i] = (column.count > currentMax) ? column.count : currentMax + } + } + return output + } +} diff --git a/Sources/Plugins/NetworkVmnet/NetworkVmnetHelper+Start.swift b/Sources/Plugins/NetworkVmnet/NetworkVmnetHelper+Start.swift index e447bcce7..7d67f1f32 100644 --- a/Sources/Plugins/NetworkVmnet/NetworkVmnetHelper+Start.swift +++ b/Sources/Plugins/NetworkVmnet/NetworkVmnetHelper+Start.swift @@ -26,7 +26,6 @@ import ContainerizationError import ContainerizationExtras import Foundation import Logging -import SystemPackage enum Variant: String, ExpressibleByArgument { case reserved diff --git a/Sources/Plugins/RuntimeLinux/RuntimeLinuxHelper+Start.swift b/Sources/Plugins/RuntimeLinux/RuntimeLinuxHelper+Start.swift index 7fd0aabdf..d4c049b4b 100644 --- a/Sources/Plugins/RuntimeLinux/RuntimeLinuxHelper+Start.swift +++ b/Sources/Plugins/RuntimeLinux/RuntimeLinuxHelper+Start.swift @@ -24,7 +24,6 @@ import ContainerXPC import Foundation import Logging import NIO -import SystemPackage extension RuntimeLinuxHelper { struct Start: AsyncParsableCommand { diff --git a/Sources/Services/ContainerAPIService/Client/Utility+PluginLoader.swift b/Sources/Services/ContainerAPIService/Client/Utility+PluginLoader.swift index d631e67c9..f5747b160 100644 --- a/Sources/Services/ContainerAPIService/Client/Utility+PluginLoader.swift +++ b/Sources/Services/ContainerAPIService/Client/Utility+PluginLoader.swift @@ -22,15 +22,12 @@ import SystemPackage extension Utility { public static func createPluginLoader(log: Logger) async throws -> PluginLoader { - let installRootPath = CommandLine.executablePath - .removingLastComponent() - .removingLastComponent() + let health = try await ClientHealthCheck.ping(timeout: .seconds(10)) // TODO: Remove when we convert PluginLoader to FilePath. - let installRootURL = URL(fileURLWithPath: installRootPath.string) - let pluginsURL = PluginLoader.userPluginsDir(installRoot: installRootURL) + let installRootPath = FilePath(health.installRoot.path(percentEncoded: false)) + let userPluginsURL = PluginLoader.userPluginsDir(installRoot: health.installRoot) var directoryExists: ObjCBool = false - _ = FileManager.default.fileExists(atPath: pluginsURL.path, isDirectory: &directoryExists) - let userPluginsURL = directoryExists.boolValue ? pluginsURL : nil + _ = FileManager.default.fileExists(atPath: userPluginsURL.path, isDirectory: &directoryExists) // plugins built into the application installed as a macOS app bundle let appBundlePluginsURL = Bundle.main.resourceURL?.appending(path: "plugins") @@ -43,7 +40,7 @@ extension Utility { .appending(FilePath.Component("plugins")) let installRootPluginsURL = URL(fileURLWithPath: installRootPluginsPath.string) let pluginDirectories = [ - userPluginsURL, + directoryExists.boolValue ? userPluginsURL : nil, appBundlePluginsURL, installRootPluginsURL, ].compactMap { $0 }