mirror of
https://github.com/dredx/prole.git
synced 2026-09-24 19:24:32 +00:00
b2cbe72c4a
- ServiceChecker now runs as a background timer task (30s), skips overlapping runs, and caches per-endpoint state from Preferences (services + kubernetes). Legacy UI flags are kept in sync for StatusView/Overlay. - Auto-reload on Preferences save: ServiceChecker listens to Config.didChange, clears caches, invalidates freshness, and triggers an immediate refresh. - Preferences: ensure in-progress text edits are committed before saving (endEditing) for Kubernetes, Services, and Ports tabs. - PortForwardManager: fix crash in termination handler by reading terminationStatus from the provided Process instance, then updating state on the manager queue. - Minor: keep existing UI wiring; logging intact for diagnostics. TODO: remove (list unused code paths) - Config.primaryKube (unused in prole-app) - Config.string(_:, default:) (unused in prole-app) - Config.int(_:, default:) (unused in prole-app) - PortForwardManager.parsePFCommand tuple parts `modeBackground` and `keepAlive` (never read by callers) - Config.pfModeBackground and Config.pfKeepAlive (currently ineffective because their values aren’t used downstream)
482 lines
21 KiB
Swift
482 lines
21 KiB
Swift
import AppKit
|
|
import Darwin
|
|
|
|
// Concurrency note:
|
|
// - All mutable state is confined to the private serial `queue`.
|
|
// - Background callbacks (Task bodies, termination handlers) hop to `queue`
|
|
// before reading/writing state. UI notifications are posted on main.
|
|
// - We mark this type as `@unchecked Sendable` to silence Sendable-capture
|
|
// warnings; correctness relies on the queue confinement invariant above.
|
|
final class PortForwardManager: @unchecked Sendable {
|
|
static let statusDidChangeNotification = Notification.Name("PortForwardManager.statusDidChange")
|
|
|
|
private let commands: [String]
|
|
private var processes: [Process] = []
|
|
// Adopted external processes (not launched by us) indexed per command
|
|
private var adoptedPIDs: [Int32?] = []
|
|
// If a port is blocked by a non-matching process, record it
|
|
private var blockedBy: [(pid: Int32, cmd: String)?] = []
|
|
// Detached async tasks supervising each launched command. We keep them to maintain lifetime.
|
|
private var tasks: [Task<Void, Never>] = []
|
|
private let queue = DispatchQueue(label: "org.prole.portforward", qos: .utility)
|
|
// Deprecated periodic monitor and shell keepalive are removed in favor of detached tasks
|
|
private var timer: DispatchSourceTimer?
|
|
// State
|
|
private var startTimes: [DispatchTime?] = []
|
|
private var failureCounts: [Int] = []
|
|
private var nextRestartAt: [DispatchTime] = []
|
|
private var lastExitCodes: [Int32?] = []
|
|
private var lastExitTimes: [Date?] = []
|
|
|
|
private let minUptimeSeconds: Double = 3.0
|
|
|
|
init(commands: [String]) {
|
|
self.commands = commands
|
|
}
|
|
|
|
func start() {
|
|
queue.async { [weak self] in
|
|
guard let self = self else { return }
|
|
if self.commands.isEmpty {
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
return
|
|
}
|
|
self.ensureArrays()
|
|
// Optional discovery/adoption of already-running matching processes to avoid duplicates
|
|
if Config.shared.pfDiscovery {
|
|
for idx in self.commands.indices { self.discoverOrAdoptIfPossible(index: idx) }
|
|
}
|
|
// Launch each command in its own detached async Task so UI remains responsive
|
|
for idx in self.commands.indices {
|
|
// Skip launching if already adopted or blocked by another process
|
|
if (self.adoptedPIDs[safe: idx] ?? nil) != nil { continue }
|
|
if (self.blockedBy[safe: idx] ?? nil) != nil { continue }
|
|
self.launchDetached(index: idx)
|
|
}
|
|
// Periodic discovery/liveness check for adopted/blocked and auto-launch when free
|
|
self.startDiscoveryTimer()
|
|
}
|
|
}
|
|
|
|
func stop() {
|
|
queue.async { [weak self] in
|
|
self?.timer?.cancel(); self?.timer = nil
|
|
for p in self?.processes ?? [] { if p.isRunning { p.terminate() } }
|
|
self?.processes.removeAll()
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
}
|
|
}
|
|
|
|
func reset() {
|
|
queue.async { [weak self] in
|
|
guard let self = self else { return }
|
|
self.timer?.cancel(); self.timer = nil
|
|
// Terminate only owned processes; keep adopted processes running
|
|
for (i, p) in self.processes.enumerated() {
|
|
if let adopted = self.adoptedPIDs[safe: i] ?? nil, adopted > 0 {
|
|
continue
|
|
}
|
|
if p.isRunning { p.terminate() }
|
|
}
|
|
// Cancel any supervising tasks (they will end when processes terminate)
|
|
for t in self.tasks { t.cancel() }
|
|
self.processes.removeAll()
|
|
self.tasks.removeAll()
|
|
self.startTimes.removeAll()
|
|
self.failureCounts.removeAll()
|
|
self.nextRestartAt.removeAll()
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
// Restart after a short delay to allow ports to release
|
|
self.queue.asyncAfter(deadline: .now() + 0.5) { [weak self] in
|
|
guard let self = self else { return }
|
|
self.ensureArrays()
|
|
if Config.shared.pfDiscovery {
|
|
for idx in self.commands.indices { self.discoverOrAdoptIfPossible(index: idx) }
|
|
}
|
|
for idx in self.commands.indices {
|
|
if (self.adoptedPIDs[safe: idx] ?? nil) != nil { continue }
|
|
if (self.blockedBy[safe: idx] ?? nil) != nil { continue }
|
|
self.launchDetached(index: idx)
|
|
}
|
|
self.startDiscoveryTimer()
|
|
}
|
|
}
|
|
}
|
|
|
|
var allRunning: Bool {
|
|
return queue.sync {
|
|
if commands.isEmpty { return false }
|
|
var runningCount = 0
|
|
for i in commands.indices {
|
|
if i < processes.count, processes[i].isRunning { runningCount += 1; continue }
|
|
if let pid = adoptedPIDs[safe: i] ?? nil, pid > 0, isPIDAlive(pid) { runningCount += 1; continue }
|
|
}
|
|
return runningCount == commands.count
|
|
}
|
|
}
|
|
|
|
var runningCount: Int {
|
|
return queue.sync {
|
|
var c = 0
|
|
for i in commands.indices {
|
|
if i < processes.count, processes[i].isRunning { c += 1; continue }
|
|
if let pid = adoptedPIDs[safe: i] ?? nil, pid > 0, isPIDAlive(pid) { c += 1 }
|
|
}
|
|
return c
|
|
}
|
|
}
|
|
|
|
struct Detail {
|
|
let command: String
|
|
let pid: Int32?
|
|
let running: Bool
|
|
let lastExitCode: Int32?
|
|
let lastExitAt: Date?
|
|
}
|
|
|
|
var details: [Detail] {
|
|
return queue.sync {
|
|
var out: [Detail] = []
|
|
for (i, rawCmd) in commands.enumerated() {
|
|
let code = i < lastExitCodes.count ? lastExitCodes[i] : nil
|
|
let when = i < lastExitTimes.count ? lastExitTimes[i] : nil
|
|
var cmd = rawCmd
|
|
var pid: Int32? = nil
|
|
var running = false
|
|
if i < processes.count, processes[i].isRunning {
|
|
pid = processes[i].processIdentifier
|
|
running = true
|
|
} else if let apid = adoptedPIDs[safe: i] ?? nil, apid > 0, isPIDAlive(apid) {
|
|
pid = apid
|
|
running = true
|
|
cmd = rawCmd + " [adopted]"
|
|
} else if let blk = blockedBy[safe: i] ?? nil {
|
|
pid = blk.pid
|
|
running = false
|
|
cmd = rawCmd + " [blocked by PID \(blk.pid) \(shortProcessName(blk.cmd))]"
|
|
}
|
|
out.append(Detail(command: cmd, pid: pid, running: running, lastExitCode: code, lastExitAt: when))
|
|
}
|
|
return out
|
|
}
|
|
}
|
|
|
|
// Per-command detached async launcher
|
|
private func launchDetached(index idx: Int) {
|
|
ensureArrays()
|
|
let raw = commands[idx]
|
|
// Per user directive: run inside a detached task and use bash -c "$cmd"
|
|
let parsed = parsePFCommand(raw)
|
|
// Important: ensure foreground execution so our Process stays alive (strip trailing &)
|
|
let body = stripTrailingAmp(parsed.command)
|
|
|
|
let task = Task.detached(priority: .utility) { [weak self] in
|
|
guard let self = self else { return }
|
|
let p = self.makeSimpleBashProcess(command: body, index: idx)
|
|
p.terminationHandler = { [weak self] proc in
|
|
guard let self = self else { return }
|
|
// Capture the terminationStatus from the actual Process instance
|
|
// provided by the handler to avoid races with self.processes[idx]
|
|
// potentially pointing at a different/non-launched Process.
|
|
let status = proc.terminationStatus
|
|
let when = Date()
|
|
self.queue.async {
|
|
if idx < self.lastExitCodes.count { self.lastExitCodes[idx] = status }
|
|
if idx < self.lastExitTimes.count { self.lastExitTimes[idx] = when }
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
}
|
|
}
|
|
do {
|
|
try p.run()
|
|
self.queue.async {
|
|
self.startTimes[idx] = .now()
|
|
if idx < self.processes.count { self.processes[idx] = p } else { self.processes.append(p) }
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
}
|
|
// Wait for the long-running child; this keeps the detached Task alive
|
|
p.waitUntilExit()
|
|
} catch {
|
|
dlog("PortForwardManager: failed to start: \(body) error=\(error)")
|
|
self.queue.async {
|
|
if idx < self.failureCounts.count { self.failureCounts[idx] += 1 }
|
|
if idx < self.processes.count { self.processes[idx] = Process() } else { self.processes.append(Process()) }
|
|
self.lastExitTimes[idx] = Date()
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if idx < tasks.count { tasks[idx] = task } else { tasks.append(task) }
|
|
}
|
|
|
|
// No periodic monitor needed with per-task supervision
|
|
|
|
// Simple bash -lc "$command" process suitable for long-running foreground tasks (kubectl port-forward)
|
|
private func makeSimpleBashProcess(command: String, index: Int) -> Process {
|
|
let cfg = Config.shared
|
|
let p = Process()
|
|
p.launchPath = "/bin/bash"
|
|
p.arguments = ["-lc", command]
|
|
p.standardInput = FileHandle(forReadingAtPath: "/dev/null")
|
|
p.qualityOfService = .utility
|
|
|
|
// Env
|
|
var env = ProcessInfo.processInfo.environment
|
|
let defaultPATH = "/usr/local/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"
|
|
if let cur = env["PATH"], !cur.isEmpty {
|
|
if !cur.contains("/opt/homebrew/bin") || !cur.contains("/usr/local/bin") {
|
|
env["PATH"] = cur + ":" + defaultPATH
|
|
}
|
|
} else { env["PATH"] = defaultPATH }
|
|
if let kube = cfg.kubeconfigPath, !kube.isEmpty { env["KUBECONFIG"] = kube }
|
|
if env["HOME"] == nil { env["HOME"] = NSHomeDirectory() }
|
|
if env["SHELL"] == nil { env["SHELL"] = "/bin/bash" }
|
|
p.environment = env
|
|
|
|
// Logs
|
|
let logsDir: URL
|
|
if let base = FileManager.default.urls(for: .libraryDirectory, in: .userDomainMask).first {
|
|
logsDir = base.appendingPathComponent("Logs/Prole", isDirectory: true)
|
|
} else {
|
|
logsDir = URL(fileURLWithPath: NSTemporaryDirectory()).appendingPathComponent("ProleLogs", isDirectory: true)
|
|
}
|
|
do { try FileManager.default.createDirectory(at: logsDir, withIntermediateDirectories: true) } catch {}
|
|
let name = String(command.hashValue, radix: 16)
|
|
let outURL = logsDir.appendingPathComponent("portforward-\(name).log")
|
|
if let fh = try? FileHandle(forWritingTo: outURL) {
|
|
try? fh.truncate(atOffset: 0)
|
|
p.standardOutput = fh
|
|
p.standardError = fh
|
|
} else if FileManager.default.createFile(atPath: outURL.path, contents: nil), let fh = try? FileHandle(forWritingTo: outURL) {
|
|
p.standardOutput = fh
|
|
p.standardError = fh
|
|
}
|
|
return p
|
|
}
|
|
|
|
// MARK: - Internals
|
|
private func ensureArrays() {
|
|
if processes.count != commands.count { processes = Array(processes.prefix(commands.count)) }
|
|
if adoptedPIDs.count != commands.count { adoptedPIDs = Array(repeating: nil, count: commands.count) }
|
|
if blockedBy.count != commands.count { blockedBy = Array(repeating: nil, count: commands.count) }
|
|
if startTimes.count != commands.count { startTimes = Array(repeating: nil, count: commands.count) }
|
|
if failureCounts.count != commands.count { failureCounts = Array(repeating: 0, count: commands.count) }
|
|
if nextRestartAt.count != commands.count { nextRestartAt = Array(repeating: .now(), count: commands.count) }
|
|
if lastExitCodes.count != commands.count { lastExitCodes = Array(repeating: nil, count: commands.count) }
|
|
if lastExitTimes.count != commands.count { lastExitTimes = Array(repeating: nil, count: commands.count) }
|
|
}
|
|
|
|
private func backoffDelay(forFailures n: Int) -> DispatchTimeInterval {
|
|
// Exponential backoff: 0.5s, 1s, 2s, 4s, capped at 10s
|
|
let base: Double = 0.5
|
|
let seconds = min(10.0, base * pow(2.0, Double(max(0, n-1))))
|
|
return .milliseconds(Int(seconds * 1000.0))
|
|
}
|
|
}
|
|
|
|
// MARK: - Helpers
|
|
private extension PortForwardManager {
|
|
// Attempt to discover an already-running matching PF process for the given command index.
|
|
func discoverOrAdoptIfPossible(index idx: Int) {
|
|
guard Config.shared.pfAdoptExisting else { return }
|
|
let cmd = commands[idx]
|
|
guard let port = localPort(from: cmd) else { return }
|
|
if let hit = findListener(on: port) {
|
|
// Determine if this looks like kubectl port-forward
|
|
let lower = hit.cmd.lowercased()
|
|
if lower.contains("kubectl") && lower.contains("port-forward") {
|
|
adoptedPIDs[idx] = hit.pid
|
|
blockedBy[idx] = nil
|
|
if Config.shared.pfDebug { dlog("PF adopt: idx=\(idx) port=\(port) pid=\(hit.pid) cmd=\(hit.cmd)") }
|
|
} else {
|
|
// Port is in use by something else; mark as blocked so we don't launch and fail.
|
|
blockedBy[idx] = hit
|
|
adoptedPIDs[idx] = nil
|
|
if Config.shared.pfDebug { dlog("PF blocked: idx=\(idx) port=\(port) by pid=\(hit.pid) cmd=\(hit.cmd)") }
|
|
}
|
|
} else {
|
|
adoptedPIDs[idx] = nil
|
|
blockedBy[idx] = nil
|
|
}
|
|
}
|
|
|
|
func startDiscoveryTimer() {
|
|
timer?.cancel()
|
|
let t = DispatchSource.makeTimerSource(queue: queue)
|
|
t.schedule(deadline: .now() + 2.0, repeating: 3.0)
|
|
t.setEventHandler { [weak self] in
|
|
guard let self = self else { return }
|
|
self.ensureArrays()
|
|
var changed = false
|
|
for idx in self.commands.indices {
|
|
// If we own a running process, continue
|
|
if idx < self.processes.count, self.processes[idx].isRunning { continue }
|
|
|
|
// If we have an adopted PID, verify liveness; clear if dead
|
|
if let apid = self.adoptedPIDs[safe: idx] ?? nil {
|
|
if !self.isPIDAlive(apid) {
|
|
self.adoptedPIDs[idx] = nil
|
|
changed = true
|
|
} else {
|
|
// still alive; nothing to do
|
|
continue
|
|
}
|
|
}
|
|
|
|
// Re-discover if enabled
|
|
if Config.shared.pfDiscovery {
|
|
let beforeBlocked = self.blockedBy[idx]?.pid
|
|
self.discoverOrAdoptIfPossible(index: idx)
|
|
if let apid = self.adoptedPIDs[idx], apid > 0 {
|
|
changed = true
|
|
continue
|
|
}
|
|
// If previously blocked but now free, mark change
|
|
if beforeBlocked != self.blockedBy[idx]?.pid { changed = true }
|
|
}
|
|
|
|
// If not adopted and not blocked, and we don't currently have a running owned process, launch
|
|
if (self.blockedBy[safe: idx] ?? nil) == nil {
|
|
// Avoid launching repeatedly if a task is already supervising a non-running Process; ensure we create a fresh one
|
|
self.launchDetached(index: idx)
|
|
changed = true
|
|
}
|
|
}
|
|
if changed {
|
|
DispatchQueue.main.async {
|
|
NotificationCenter.default.post(name: PortForwardManager.statusDidChangeNotification, object: self)
|
|
}
|
|
}
|
|
}
|
|
self.timer = t
|
|
t.resume()
|
|
}
|
|
|
|
func isPIDAlive(_ pid: Int32) -> Bool {
|
|
return kill(pid_t(pid), 0) == 0
|
|
}
|
|
|
|
func shortProcessName(_ full: String) -> String {
|
|
if let last = full.split(separator: "/").last { return String(last) }
|
|
return full
|
|
}
|
|
|
|
func localPort(from command: String) -> Int? {
|
|
// Find first occurrence of pattern like "8080:xxxx"
|
|
let pattern = #"(\s|^)(\d+):\d+"#
|
|
if let regex = try? NSRegularExpression(pattern: pattern, options: []) {
|
|
let ns = command as NSString
|
|
let range = NSRange(location: 0, length: ns.length)
|
|
if let m = regex.firstMatch(in: command, options: [], range: range), m.numberOfRanges >= 3 {
|
|
let portStr = ns.substring(with: m.range(at: 2))
|
|
return Int(portStr)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func findListener(on port: Int) -> (pid: Int32, cmd: String)? {
|
|
// Try lsof to get PID listening on the given TCP port
|
|
let lsofPaths = ["/usr/sbin/lsof", "/usr/bin/lsof", "/bin/lsof", "/usr/local/bin/lsof", "/opt/homebrew/sbin/lsof", "/opt/homebrew/bin/lsof", "lsof"]
|
|
var lsofOut: String = ""
|
|
for path in lsofPaths {
|
|
if let out = runAndCapture(path, ["-nP", "-iTCP:\(port)", "-sTCP:LISTEN", "-Fp"]), !out.isEmpty {
|
|
lsofOut = out; break
|
|
}
|
|
}
|
|
guard !lsofOut.isEmpty else { return nil }
|
|
// Parse first pid line like: p12345
|
|
var pid: Int32 = 0
|
|
for line in lsofOut.split(separator: "\n") {
|
|
if line.hasPrefix("p"), let p = Int32(line.dropFirst()) { pid = p; break }
|
|
}
|
|
if pid <= 0 { return nil }
|
|
// Get full command for that PID
|
|
let psOut = runAndCapture("/bin/ps", ["-p", String(pid), "-o", "command="]) ?? ""
|
|
let cmd = psOut.trimmingCharacters(in: .whitespacesAndNewlines)
|
|
return (pid, cmd)
|
|
}
|
|
|
|
func runAndCapture(_ launchPath: String, _ args: [String]) -> String? {
|
|
let p = Process()
|
|
p.launchPath = launchPath
|
|
p.arguments = args
|
|
let pipe = Pipe()
|
|
p.standardOutput = pipe
|
|
p.standardError = Pipe()
|
|
do {
|
|
try p.run()
|
|
} catch {
|
|
return nil
|
|
}
|
|
p.waitUntilExit()
|
|
let data = try? pipe.fileHandleForReading.readToEnd()
|
|
guard let d = data, let s = String(data: d, encoding: .utf8) else { return nil }
|
|
return s
|
|
}
|
|
func parsePFCommand(_ raw: String) -> (command: String, modeBackground: Bool, keepAlive: Bool) {
|
|
var s = raw.trimmingCharacters(in: .whitespacesAndNewlines)
|
|
var bg = Config.shared.pfModeBackground
|
|
var ka = Config.shared.pfKeepAlive
|
|
// Support multiple prefixes in any order: bg:, fg:, ka:, noka:
|
|
while true {
|
|
let lower = s.lowercased()
|
|
if lower.hasPrefix("bg:") {
|
|
s = String(s.dropFirst(3)).trimmingCharacters(in: .whitespaces)
|
|
bg = true
|
|
continue
|
|
}
|
|
if lower.hasPrefix("fg:") {
|
|
s = String(s.dropFirst(3)).trimmingCharacters(in: .whitespaces)
|
|
bg = false
|
|
continue
|
|
}
|
|
if lower.hasPrefix("ka:") {
|
|
s = String(s.dropFirst(3)).trimmingCharacters(in: .whitespaces)
|
|
ka = true
|
|
continue
|
|
}
|
|
if lower.hasPrefix("noka:") {
|
|
s = String(s.dropFirst(5)).trimmingCharacters(in: .whitespaces)
|
|
ka = false
|
|
continue
|
|
}
|
|
break
|
|
}
|
|
if !bg {
|
|
// In foreground mode, ensure we don't leave a trailing '&'
|
|
s = stripTrailingAmp(s)
|
|
}
|
|
return (s, bg, ka)
|
|
}
|
|
|
|
func stripTrailingAmp(_ s: String) -> String {
|
|
let regex = try? NSRegularExpression(pattern: "\\s*&\\s*$")
|
|
let ns = s as NSString
|
|
let range = NSRange(location: 0, length: ns.length)
|
|
if let r = regex?.firstMatch(in: s, options: [], range: range) {
|
|
let trimmed = ns.replacingCharacters(in: r.range, with: "")
|
|
return trimmed.trimmingCharacters(in: .whitespacesAndNewlines)
|
|
}
|
|
return s
|
|
}
|
|
}
|
|
|
|
// Safe subscript to avoid index crashes inside async handlers
|
|
private extension Array {
|
|
subscript(safe index: Int) -> Element? {
|
|
return indices.contains(index) ? self[index] : nil
|
|
}
|
|
}
|