mirror of
https://github.com/apple/container.git
synced 2026-08-24 10:05:43 -05:00
Split K8sHelper, add WorkerProvisioner protocol, and add LinuxWorker … (#2110)
- Closes #2102. - LinuxWorkerProvisioner sets up a kind-based node. - Includes some test cleanups. Co-authored-by: John Logan <john_logan@apple.com>
This commit is contained in:
jshi991
John Logan
+2
-3
@@ -166,13 +166,12 @@ let package = Package(
|
||||
]
|
||||
),
|
||||
.testTarget(
|
||||
name: "K8sTests",
|
||||
name: "K8sPluginTests",
|
||||
dependencies: [
|
||||
"ContainerK8s",
|
||||
"ContainerResource",
|
||||
"Yams",
|
||||
],
|
||||
path: "Tests/K8sPluginTests"
|
||||
]
|
||||
),
|
||||
.target(
|
||||
name: "ContainerK8s",
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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: [],
|
||||
let provisioner = try LinuxNodeProvisioner(
|
||||
clusterName: name,
|
||||
roles: [StandardRoles.controlPlane, StandardRoles.worker],
|
||||
nodeImage: nodeImage,
|
||||
cpus: resourceFlags.cpus,
|
||||
memory: resourceFlags.memory,
|
||||
registryScheme: registryFlags.scheme,
|
||||
maxConcurrentDownloads: imageFetchFlags.maxConcurrentDownloads,
|
||||
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
|
||||
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()
|
||||
try await provisioner.provision(name: name, log: log)
|
||||
|
||||
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")
|
||||
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()
|
||||
|
||||
@@ -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..<self.rows.count {
|
||||
let row = self.rows[rowIndex]
|
||||
for columnIndex in 0..<row.count - 1 {
|
||||
let currentLength = (maxLengths[columnIndex] ?? 0) + self.spacing
|
||||
let padded = row[columnIndex].padding(toLength: currentLength, withPad: " ", startingAt: 0)
|
||||
output += padded
|
||||
}
|
||||
output += row.last ?? ""
|
||||
output += (rowIndex == self.rows.count - 1) ? "" : "\n"
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
private func maxLength() -> [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<UInt16> = []) 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<sockaddr_in>.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
|
||||
|
||||
@@ -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<String> = [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)"])
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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: `<ip>:<port>` of the control-plane API server.
|
||||
/// - token: kubeadm bootstrap token (format `<id>.<secret>`).
|
||||
/// - 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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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<UInt16> = []) 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<sockaddr_in>.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)
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
)
|
||||
}
|
||||
}
|
||||
@@ -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..<self.rows.count {
|
||||
let row = self.rows[rowIndex]
|
||||
for columnIndex in 0..<row.count - 1 {
|
||||
let currentLength = (maxLengths[columnIndex] ?? 0) + self.spacing
|
||||
let padded = row[columnIndex].padding(toLength: currentLength, withPad: " ", startingAt: 0)
|
||||
output += padded
|
||||
}
|
||||
output += row.last ?? ""
|
||||
output += (rowIndex == self.rows.count - 1) ? "" : "\n"
|
||||
}
|
||||
return output
|
||||
}
|
||||
|
||||
private func maxLength() -> [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
|
||||
}
|
||||
}
|
||||
@@ -26,7 +26,6 @@ import ContainerizationError
|
||||
import ContainerizationExtras
|
||||
import Foundation
|
||||
import Logging
|
||||
import SystemPackage
|
||||
|
||||
enum Variant: String, ExpressibleByArgument {
|
||||
case reserved
|
||||
|
||||
@@ -24,7 +24,6 @@ import ContainerXPC
|
||||
import Foundation
|
||||
import Logging
|
||||
import NIO
|
||||
import SystemPackage
|
||||
|
||||
extension RuntimeLinuxHelper {
|
||||
struct Start: AsyncParsableCommand {
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -0,0 +1,41 @@
|
||||
//===----------------------------------------------------------------------===//
|
||||
// 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 ContainerTestSupport
|
||||
import Foundation
|
||||
|
||||
extension ContainerFixture {
|
||||
func dumpNodeDiagnostics(node: String) {
|
||||
print("=== NODE DIAGNOSTICS [\(node)] ===")
|
||||
let cmds: [(label: String, args: [String])] = [
|
||||
("ip-link", ["ip", "link", "show"]),
|
||||
("iptables-mss", ["iptables", "-t", "mangle", "-L", "-n", "-v"]),
|
||||
("containerd", ["systemctl", "status", "containerd", "--no-pager", "-l"]),
|
||||
("kubelet-log", ["journalctl", "-u", "kubelet", "--no-pager", "-n", "60"]),
|
||||
("crictl-images", ["crictl", "images"]),
|
||||
]
|
||||
for (label, args) in cmds {
|
||||
if let r = try? self.run(["exec", node] + args) {
|
||||
let out = r.output.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let err = r.error.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
print("[\(label)] exit=\(r.status)")
|
||||
if !out.isEmpty { print(out) }
|
||||
if !err.isEmpty { print("stderr: \(err)") }
|
||||
}
|
||||
}
|
||||
print("=== END NODE DIAGNOSTICS ===")
|
||||
}
|
||||
}
|
||||
@@ -23,27 +23,6 @@ struct TestK8sLoadImageSerial {
|
||||
|
||||
private static let testImage = WarmupImage.alpine320.rawValue
|
||||
|
||||
private func dumpNodeDiagnostics(_ f: ContainerFixture, node: String) {
|
||||
print("=== NODE DIAGNOSTICS [\(node)] ===")
|
||||
let cmds: [(label: String, args: [String])] = [
|
||||
("ip-link", ["ip", "link", "show"]),
|
||||
("iptables-mss", ["iptables", "-t", "mangle", "-L", "-n", "-v"]),
|
||||
("containerd", ["systemctl", "status", "containerd", "--no-pager", "-l"]),
|
||||
("kubelet-log", ["journalctl", "-u", "kubelet", "--no-pager", "-n", "60"]),
|
||||
("crictl-images", ["crictl", "images"]),
|
||||
]
|
||||
for (label, args) in cmds {
|
||||
if let r = try? f.run(["exec", node] + args) {
|
||||
let out = r.output.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let err = r.error.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
print("[\(label)] exit=\(r.status)")
|
||||
if !out.isEmpty { print(out) }
|
||||
if !err.isEmpty { print("stderr: \(err)") }
|
||||
}
|
||||
}
|
||||
print("=== END NODE DIAGNOSTICS ===")
|
||||
}
|
||||
|
||||
private func imageExistsInNode(_ f: ContainerFixture, node: String, image: String) throws -> Bool {
|
||||
print("[k8s-load] ctr images list (node: \(node), checking: \(image))")
|
||||
let result = try f.run(["exec", node, "ctr", "--namespace", "k8s.io", "images", "list"])
|
||||
@@ -81,7 +60,7 @@ struct TestK8sLoadImageSerial {
|
||||
if result.status != 0 {
|
||||
print("[k8s-load] k8s create stderr: \(result.error)")
|
||||
dumpEnv(f, clusterName: name)
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
|
||||
|
||||
@@ -21,28 +21,7 @@ import Testing
|
||||
@Suite(.serialized)
|
||||
struct TestK8sNetworkingSerial {
|
||||
|
||||
private static let testImage = WarmupImage.alpine320.rawValue
|
||||
|
||||
private func dumpNodeDiagnostics(_ f: ContainerFixture, node: String) {
|
||||
print("=== NODE DIAGNOSTICS [\(node)] ===")
|
||||
let cmds: [(label: String, args: [String])] = [
|
||||
("ip-link", ["ip", "link", "show"]),
|
||||
("iptables-mss", ["iptables", "-t", "mangle", "-L", "-n", "-v"]),
|
||||
("containerd", ["systemctl", "status", "containerd", "--no-pager", "-l"]),
|
||||
("kubelet-log", ["journalctl", "-u", "kubelet", "--no-pager", "-n", "60"]),
|
||||
("crictl-images", ["crictl", "images"]),
|
||||
]
|
||||
for (label, args) in cmds {
|
||||
if let r = try? f.run(["exec", node] + args) {
|
||||
let out = r.output.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let err = r.error.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
print("[\(label)] exit=\(r.status)")
|
||||
if !out.isEmpty { print(out) }
|
||||
if !err.isEmpty { print("stderr: \(err)") }
|
||||
}
|
||||
}
|
||||
print("=== END NODE DIAGNOSTICS ===")
|
||||
}
|
||||
private static let testImage = WarmupImage.alpine320
|
||||
|
||||
private func dumpEnv(_ f: ContainerFixture, clusterName: String) {
|
||||
print("=== ENV DUMP [\(clusterName)] ===")
|
||||
@@ -72,19 +51,19 @@ struct TestK8sNetworkingSerial {
|
||||
return (result.output, result.status)
|
||||
}
|
||||
|
||||
private func waitForPod(_ f: ContainerFixture, node: String, podName: String, timeout: Int = 300) throws {
|
||||
print("[k8s-net] waitForPod \(podName) on \(node) (timeout=\(timeout)s)")
|
||||
private func waitForPod(_ f: ContainerFixture, node: String, podName: String, timeoutSeconds: Int) throws {
|
||||
print("[k8s-net] waitForPod \(podName) on \(node) (timeout=\(timeoutSeconds)s)")
|
||||
let (_, status) = try kubectl(
|
||||
f, node: node,
|
||||
args: [
|
||||
"wait", "--for=condition=Ready", "pod/\(podName)", "--timeout=\(timeout)s",
|
||||
"wait", "--for=condition=Ready", "pod/\(podName)", "--timeout=\(timeoutSeconds)s",
|
||||
])
|
||||
guard status == 0 else {
|
||||
let (podStatus, _) = try kubectl(f, node: node, args: ["get", "pod", podName, "--no-headers"])
|
||||
print("[k8s-net] pod \(podName) status: \(podStatus.trimmingCharacters(in: .whitespacesAndNewlines))")
|
||||
let (podDesc, _) = try kubectl(f, node: node, args: ["describe", "pod", podName])
|
||||
print("[k8s-net] pod \(podName) describe:\n\(podDesc.prefix(1000))")
|
||||
throw CommandError.executionFailed("pod \(podName) did not become ready within \(timeout)s")
|
||||
throw CommandError.executionFailed("pod \(podName) did not become ready within \(timeoutSeconds)s")
|
||||
}
|
||||
print("[k8s-net] pod \(podName) is Ready")
|
||||
}
|
||||
@@ -102,15 +81,15 @@ struct TestK8sNetworkingSerial {
|
||||
if result.status != 0 {
|
||||
print("[k8s-net] k8s create stderr: \(result.error)")
|
||||
dumpEnv(f, clusterName: name)
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
|
||||
print("[k8s-net] pulling \(Self.testImage)")
|
||||
try f.restoreWarmupImage(.alpine320)
|
||||
print("[k8s-net] pulling \(Self.testImage.rawValue)")
|
||||
try f.restoreWarmupImage(Self.testImage)
|
||||
|
||||
print("[k8s-net] k8s load-image --name \(name) \(Self.testImage)")
|
||||
let loadResult = try f.run(["k8s", "load-image", "--name", name, Self.testImage])
|
||||
print("[k8s-net] k8s load-image --name \(name) \(Self.testImage.rawValue)")
|
||||
let loadResult = try f.run(["k8s", "load-image", "--name", name, Self.testImage.rawValue])
|
||||
print("[k8s-net] k8s load-image exit=\(loadResult.status)")
|
||||
if loadResult.status != 0 { print("[k8s-net] k8s load-image stderr: \(loadResult.error)") }
|
||||
#expect(loadResult.status == 0)
|
||||
@@ -119,14 +98,14 @@ struct TestK8sNetworkingSerial {
|
||||
f, node: name,
|
||||
args: [
|
||||
"run", "test-pod",
|
||||
"--image=\(Self.testImage)",
|
||||
"--image=\(Self.testImage.rawValue)",
|
||||
"--image-pull-policy=Never",
|
||||
"--restart=Never",
|
||||
"--", "sleep", "300",
|
||||
])
|
||||
#expect(createStatus == 0)
|
||||
|
||||
try waitForPod(f, node: name, podName: "test-pod")
|
||||
try waitForPod(f, node: name, podName: "test-pod", timeoutSeconds: 300)
|
||||
|
||||
let (output, execStatus) = try kubectl(
|
||||
f, node: name,
|
||||
@@ -154,11 +133,11 @@ struct TestK8sNetworkingSerial {
|
||||
}
|
||||
try result.check()
|
||||
|
||||
print("[k8s-net] pulling \(Self.testImage)")
|
||||
try f.restoreWarmupImage(.alpine320)
|
||||
print("[k8s-net] pulling \(Self.testImage.rawValue)")
|
||||
try f.restoreWarmupImage(Self.testImage)
|
||||
|
||||
print("[k8s-net] k8s load-image --name \(name) \(Self.testImage)")
|
||||
let loadResult = try f.run(["k8s", "load-image", "--name", name, Self.testImage])
|
||||
print("[k8s-net] k8s load-image --name \(name) \(Self.testImage.rawValue)")
|
||||
let loadResult = try f.run(["k8s", "load-image", "--name", name, Self.testImage.rawValue])
|
||||
print("[k8s-net] k8s load-image exit=\(loadResult.status)")
|
||||
if loadResult.status != 0 { print("[k8s-net] k8s load-image stderr: \(loadResult.error)") }
|
||||
#expect(loadResult.status == 0)
|
||||
@@ -168,7 +147,7 @@ struct TestK8sNetworkingSerial {
|
||||
f, node: name,
|
||||
args: [
|
||||
"run", "server",
|
||||
"--image=\(Self.testImage)",
|
||||
"--image=\(Self.testImage.rawValue)",
|
||||
"--image-pull-policy=Never",
|
||||
"--restart=Never",
|
||||
"--port=8080",
|
||||
@@ -176,7 +155,7 @@ struct TestK8sNetworkingSerial {
|
||||
"while true; do printf 'HTTP/1.1 200 OK\\r\\nContent-Length: 2\\r\\nConnection: close\\r\\n\\r\\nok' | nc -l -p 8080; done",
|
||||
])
|
||||
#expect(serverStatus == 0)
|
||||
try waitForPod(f, node: name, podName: "server")
|
||||
try waitForPod(f, node: name, podName: "server", timeoutSeconds: 300)
|
||||
|
||||
// Expose server as a ClusterIP service.
|
||||
let (_, exposeStatus) = try kubectl(
|
||||
@@ -191,13 +170,13 @@ struct TestK8sNetworkingSerial {
|
||||
f, node: name,
|
||||
args: [
|
||||
"run", "client",
|
||||
"--image=\(Self.testImage)",
|
||||
"--image=\(Self.testImage.rawValue)",
|
||||
"--image-pull-policy=Never",
|
||||
"--restart=Never",
|
||||
"--", "sleep", "300",
|
||||
])
|
||||
#expect(clientStatus == 0)
|
||||
try waitForPod(f, node: name, podName: "client")
|
||||
try waitForPod(f, node: name, podName: "client", timeoutSeconds: 300)
|
||||
|
||||
// Reach server via the service DNS name — exercises CoreDNS + kube-proxy.
|
||||
print("[k8s-net] wget from client to server-svc:8080")
|
||||
|
||||
@@ -23,27 +23,6 @@ import Yams
|
||||
@Suite(.serialized)
|
||||
struct TestK8sRunSerial {
|
||||
|
||||
private func dumpNodeDiagnostics(_ f: ContainerFixture, node: String) {
|
||||
print("=== NODE DIAGNOSTICS [\(node)] ===")
|
||||
let cmds: [(label: String, args: [String])] = [
|
||||
("ip-link", ["ip", "link", "show"]),
|
||||
("iptables-mss", ["iptables", "-t", "mangle", "-L", "-n", "-v"]),
|
||||
("containerd", ["systemctl", "status", "containerd", "--no-pager", "-l"]),
|
||||
("kubelet-log", ["journalctl", "-u", "kubelet", "--no-pager", "-n", "60"]),
|
||||
("crictl-images", ["crictl", "images"]),
|
||||
]
|
||||
for (label, args) in cmds {
|
||||
if let r = try? f.run(["exec", node] + args) {
|
||||
let out = r.output.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let err = r.error.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
print("[\(label)] exit=\(r.status)")
|
||||
if !out.isEmpty { print(out) }
|
||||
if !err.isEmpty { print("stderr: \(err)") }
|
||||
}
|
||||
}
|
||||
print("=== END NODE DIAGNOSTICS ===")
|
||||
}
|
||||
|
||||
private func loadKubeconfig() throws -> [String: Any] {
|
||||
let path = FilePath(FileManager.default.homeDirectoryForCurrentUser.path)
|
||||
.appending(".kube")
|
||||
@@ -76,7 +55,7 @@ struct TestK8sRunSerial {
|
||||
print("[k8s-run] k8s create exit=\(result.status)")
|
||||
if result.status != 0 {
|
||||
print("[k8s-run] k8s create stderr: \(result.error)")
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
#expect(result.output.contains(name))
|
||||
@@ -111,7 +90,7 @@ struct TestK8sRunSerial {
|
||||
print("[k8s-run] k8s create exit=\(result1.status)")
|
||||
if result1.status != 0 {
|
||||
print("[k8s-run] k8s create stderr: \(result1.error)")
|
||||
dumpNodeDiagnostics(f, node: name1)
|
||||
f.dumpNodeDiagnostics(node: name1)
|
||||
}
|
||||
#expect(result1.status == 0)
|
||||
|
||||
@@ -120,7 +99,7 @@ struct TestK8sRunSerial {
|
||||
print("[k8s-run] k8s create exit=\(result2.status)")
|
||||
if result2.status != 0 {
|
||||
print("[k8s-run] k8s create stderr: \(result2.error)")
|
||||
dumpNodeDiagnostics(f, node: name2)
|
||||
f.dumpNodeDiagnostics(node: name2)
|
||||
}
|
||||
#expect(result2.status == 0)
|
||||
|
||||
|
||||
@@ -23,27 +23,6 @@ import Yams
|
||||
@Suite(.serialized)
|
||||
struct TestK8sWriteConfigSerial {
|
||||
|
||||
private func dumpNodeDiagnostics(_ f: ContainerFixture, node: String) {
|
||||
print("=== NODE DIAGNOSTICS [\(node)] ===")
|
||||
let cmds: [(label: String, args: [String])] = [
|
||||
("ip-link", ["ip", "link", "show"]),
|
||||
("iptables-mss", ["iptables", "-t", "mangle", "-L", "-n", "-v"]),
|
||||
("containerd", ["systemctl", "status", "containerd", "--no-pager", "-l"]),
|
||||
("kubelet-log", ["journalctl", "-u", "kubelet", "--no-pager", "-n", "60"]),
|
||||
("crictl-images", ["crictl", "images"]),
|
||||
]
|
||||
for (label, args) in cmds {
|
||||
if let r = try? f.run(["exec", node] + args) {
|
||||
let out = r.output.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
let err = r.error.trimmingCharacters(in: .whitespacesAndNewlines)
|
||||
print("[\(label)] exit=\(r.status)")
|
||||
if !out.isEmpty { print(out) }
|
||||
if !err.isEmpty { print("stderr: \(err)") }
|
||||
}
|
||||
}
|
||||
print("=== END NODE DIAGNOSTICS ===")
|
||||
}
|
||||
|
||||
private func kubeconfigPath() -> FilePath {
|
||||
FilePath(FileManager.default.homeDirectoryForCurrentUser.path)
|
||||
.appending(".kube")
|
||||
@@ -106,7 +85,7 @@ struct TestK8sWriteConfigSerial {
|
||||
if result.status != 0 {
|
||||
print("[k8s-cfg] k8s create stderr: \(result.error)")
|
||||
dumpEnv(f, clusterName: name)
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
|
||||
@@ -144,7 +123,7 @@ struct TestK8sWriteConfigSerial {
|
||||
if result.status != 0 {
|
||||
print("[k8s-cfg] k8s create stderr: \(result.error)")
|
||||
dumpEnv(f, clusterName: name)
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
|
||||
@@ -174,7 +153,7 @@ struct TestK8sWriteConfigSerial {
|
||||
if result.status != 0 {
|
||||
print("[k8s-cfg] k8s create stderr: \(result.error)")
|
||||
dumpEnv(f, clusterName: name)
|
||||
dumpNodeDiagnostics(f, node: name)
|
||||
f.dumpNodeDiagnostics(node: name)
|
||||
}
|
||||
try result.check()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user