373 lines
14 KiB
Swift
373 lines
14 KiB
Swift
import Foundation
|
|
|
|
public struct LogCenterService: Sendable {
|
|
private struct ProbeOutcome: Sendable {
|
|
var status: ConnectionStatus
|
|
var outcome: ConnectionOutcome
|
|
var state: String
|
|
var errorCode: String?
|
|
var httpStatus: Int?
|
|
var durationMs: Int
|
|
var message: String
|
|
var version: String?
|
|
var platform: String?
|
|
}
|
|
|
|
private struct InstanceCollection: Sendable {
|
|
var runtime: [RuntimeLogEntry] = []
|
|
var connections: [ConnectionLogEntry] = []
|
|
var diagnostic: InstanceDiagnostic?
|
|
}
|
|
|
|
private let client: any SimAdminClientProtocol
|
|
private let sessions: InstanceSessionStore
|
|
private let credentials: (any CredentialStore)?
|
|
private let store: any LogStore
|
|
|
|
public init(
|
|
client: any SimAdminClientProtocol,
|
|
sessions: InstanceSessionStore,
|
|
credentials: (any CredentialStore)? = nil,
|
|
store: any LogStore
|
|
) {
|
|
self.client = client
|
|
self.sessions = sessions
|
|
self.credentials = credentials
|
|
self.store = store
|
|
}
|
|
|
|
public func refresh(_ instances: [Instance]) async throws -> LogSnapshot {
|
|
var collected: [InstanceCollection] = []
|
|
await withTaskGroup(of: InstanceCollection.self) { group in
|
|
for instance in instances {
|
|
group.addTask { await self.collect(instance) }
|
|
}
|
|
for await value in group {
|
|
collected.append(value)
|
|
}
|
|
}
|
|
|
|
collected.sort { ($0.diagnostic?.name ?? "") < ($1.diagnostic?.name ?? "") }
|
|
let runtime = collected.flatMap(\.runtime).sorted { $0.occurredAt > $1.occurredAt }
|
|
let connections = collected.flatMap(\.connections).sorted { $0.observedAt > $1.observedAt }
|
|
let diagnostics = collected.compactMap(\.diagnostic)
|
|
try await store.append(runtime: runtime, connections: connections)
|
|
|
|
var counts: [RuntimeLogSource: Int] = [:]
|
|
for source in RuntimeLogSource.allCases {
|
|
counts[source] = runtime.filter { $0.source == source }.count
|
|
}
|
|
return LogSnapshot(
|
|
observedAt: Date(),
|
|
runtime: runtime,
|
|
counts: counts,
|
|
connections: connections,
|
|
diagnostics: diagnostics
|
|
)
|
|
}
|
|
|
|
private func collect(_ instance: Instance) async -> InstanceCollection {
|
|
var result = InstanceCollection()
|
|
let probe = await probe(instance)
|
|
var modules: [String: Result<Bool, Error>] = [
|
|
"connection": .success(true),
|
|
"device": await moduleResult(instance, .device)
|
|
]
|
|
if probe.outcome == .failed {
|
|
modules["connection"] = .failure(UpstreamError.notSupported)
|
|
modules["device"] = .failure(UpstreamError.notSupported)
|
|
}
|
|
result.connections.append(ConnectionLogEntry(
|
|
id: "probe-\(instance.id.uuidString)-\(Int(Date().timeIntervalSince1970 * 1000))",
|
|
instanceID: instance.id,
|
|
instanceName: instance.name,
|
|
outcome: probe.outcome,
|
|
state: probe.state,
|
|
errorCode: probe.errorCode,
|
|
httpStatus: probe.httpStatus,
|
|
durationMs: probe.durationMs,
|
|
observedAt: Date()
|
|
))
|
|
if probe.outcome == .failed || probe.outcome == .stale {
|
|
result.runtime.append(RuntimeLogEntry(
|
|
id: "connection-\(instance.id.uuidString)-\(Int(Date().timeIntervalSince1970 * 1000))",
|
|
source: .delivery,
|
|
level: probe.outcome == .failed ? .error : .warning,
|
|
message: probe.message,
|
|
instanceID: instance.id,
|
|
instanceName: instance.name,
|
|
occurredAt: Date(),
|
|
code: probe.errorCode,
|
|
durationMs: probe.durationMs
|
|
))
|
|
}
|
|
|
|
let deliveryEntries = probe.outcome == .failed ? [] : await deliveryLogs(instance)
|
|
result.runtime.append(contentsOf: deliveryEntries)
|
|
if probe.outcome == .failed {
|
|
modules["delivery"] = .failure(UpstreamError.notSupported)
|
|
} else if deliveryEntries.contains(where: { $0.code == "UNAUTHORIZED" }) {
|
|
modules["delivery"] = .failure(UpstreamError.unauthorized)
|
|
} else if deliveryEntries.contains(where: { $0.code == "UPSTREAM" }) {
|
|
modules["delivery"] = .failure(UpstreamError.invalidResponse)
|
|
} else {
|
|
modules["delivery"] = .success(true)
|
|
}
|
|
result.diagnostic = diagnostic(
|
|
instance,
|
|
probe: probe,
|
|
modules: modules,
|
|
connections: result.connections
|
|
)
|
|
return result
|
|
}
|
|
|
|
private func probe(_ instance: Instance) async -> ProbeOutcome {
|
|
let startedAt = Date()
|
|
do {
|
|
let response = try await client.send(.health, baseURL: instance.origin, cookie: nil)
|
|
let duration = elapsed(startedAt)
|
|
let root = (try? JSONSerialization.jsonObject(with: response.data)) as? [String: Any] ?? [:]
|
|
return ProbeOutcome(
|
|
status: .online,
|
|
outcome: .success,
|
|
state: "online",
|
|
errorCode: nil,
|
|
httpStatus: response.status,
|
|
durationMs: duration,
|
|
message: "节点连接成功",
|
|
version: Self.text(root["version"]),
|
|
platform: Self.text(root["platform"])
|
|
)
|
|
} catch UpstreamError.unauthorized, UpstreamError.forbidden {
|
|
return ProbeOutcome(
|
|
status: .authRequired,
|
|
outcome: .stale,
|
|
state: "auth_required",
|
|
errorCode: "UNAUTHORIZED",
|
|
httpStatus: 401,
|
|
durationMs: elapsed(startedAt),
|
|
message: "节点需要登录",
|
|
version: nil,
|
|
platform: nil
|
|
)
|
|
} catch UpstreamError.timeout {
|
|
return ProbeOutcome(
|
|
status: .offline,
|
|
outcome: .failed,
|
|
state: "unreachable",
|
|
errorCode: "TIMEOUT",
|
|
httpStatus: nil,
|
|
durationMs: elapsed(startedAt),
|
|
message: "节点连接超时",
|
|
version: nil,
|
|
platform: nil
|
|
)
|
|
} catch {
|
|
return ProbeOutcome(
|
|
status: .degraded,
|
|
outcome: .failed,
|
|
state: "unreachable",
|
|
errorCode: "TRANSPORT",
|
|
httpStatus: nil,
|
|
durationMs: elapsed(startedAt),
|
|
message: "节点连接失败",
|
|
version: nil,
|
|
platform: nil
|
|
)
|
|
}
|
|
}
|
|
|
|
private func moduleResult(
|
|
_ instance: Instance,
|
|
_ route: UpstreamRoute
|
|
) async -> Result<Bool, Error> {
|
|
do {
|
|
_ = try await request(route, instance)
|
|
return .success(true)
|
|
} catch {
|
|
return .failure(error)
|
|
}
|
|
}
|
|
|
|
private func deliveryLogs(_ instance: Instance) async -> [RuntimeLogEntry] {
|
|
do {
|
|
let response = try await request(.notificationLogs(limit: 50, offset: 0), instance)
|
|
let entries = Self.deliveryEntries(response, instance)
|
|
if entries.isEmpty {
|
|
throw UpstreamError.invalidResponse
|
|
}
|
|
return entries
|
|
} catch UpstreamError.unauthorized, UpstreamError.forbidden {
|
|
return [RuntimeLogEntry(
|
|
id: "delivery-\(instance.id.uuidString)-auth",
|
|
source: .delivery,
|
|
level: .warning,
|
|
message: "通知日志未同步：需要登录",
|
|
instanceID: instance.id,
|
|
instanceName: instance.name,
|
|
occurredAt: Date(),
|
|
code: "UNAUTHORIZED",
|
|
durationMs: nil
|
|
)]
|
|
} catch {
|
|
return [RuntimeLogEntry(
|
|
id: "delivery-\(instance.id.uuidString)-failed",
|
|
source: .delivery,
|
|
level: .warning,
|
|
message: "通知日志未同步：接口异常",
|
|
instanceID: instance.id,
|
|
instanceName: instance.name,
|
|
occurredAt: Date(),
|
|
code: "UPSTREAM",
|
|
durationMs: nil
|
|
)]
|
|
}
|
|
}
|
|
|
|
private func diagnostic(
|
|
_ instance: Instance,
|
|
probe: ProbeOutcome,
|
|
modules: [String: Result<Bool, Error>],
|
|
connections: [ConnectionLogEntry]
|
|
) -> InstanceDiagnostic {
|
|
let capabilities = LogCapabilitySummary(
|
|
total: modules.count,
|
|
supported: modules.values.filter {
|
|
if case .success = $0 { return true }
|
|
return false
|
|
}.count,
|
|
unsupported: modules.values.filter {
|
|
if case .failure(UpstreamError.notSupported) = $0 { return true }
|
|
return false
|
|
}.count,
|
|
authRequired: modules.values.filter {
|
|
if case .failure(UpstreamError.unauthorized) = $0 { return true }
|
|
if case .failure(UpstreamError.forbidden) = $0 { return true }
|
|
return false
|
|
}.count,
|
|
degraded: modules.values.filter {
|
|
if case .failure = $0 {
|
|
if case .failure(UpstreamError.notSupported) = $0 { return false }
|
|
if case .failure(UpstreamError.unauthorized) = $0 { return false }
|
|
if case .failure(UpstreamError.forbidden) = $0 { return false }
|
|
return true
|
|
}
|
|
return false
|
|
}.count,
|
|
unknown: 0
|
|
)
|
|
let success = connections.filter { $0.outcome == .success }.count
|
|
let failed = connections.filter { $0.outcome == .failed }.count
|
|
let average = connections.isEmpty
|
|
? 0
|
|
: connections.map(\.durationMs).reduce(0, +) / connections.count
|
|
return InstanceDiagnostic(
|
|
id: instance.id,
|
|
name: instance.name,
|
|
origin: instance.origin.absoluteString,
|
|
tags: instance.tags,
|
|
state: probe.status,
|
|
healthStatus: Self.healthLabel(probe),
|
|
version: probe.version,
|
|
platform: probe.platform,
|
|
observedAt: Date(),
|
|
capabilities: capabilities,
|
|
connection: ConnectionSummary(
|
|
total: connections.count,
|
|
success: success,
|
|
failed: failed,
|
|
averageDurationMs: average,
|
|
availabilityPercent: connections.isEmpty ? 0 : success * 100 / connections.count
|
|
)
|
|
)
|
|
}
|
|
|
|
static func deliveryEntries(
|
|
_ response: UpstreamResponse,
|
|
_ instance: Instance
|
|
) -> [RuntimeLogEntry] {
|
|
guard let root = try? JSONSerialization.jsonObject(with: response.data) as? [String: Any],
|
|
let data = root["data"] as? [String: Any],
|
|
let raw = data["logs"] as? [[String: Any]]
|
|
else {
|
|
return []
|
|
}
|
|
return raw.compactMap { item in
|
|
guard let id = Self.text(item["id"]),
|
|
let status = Self.text(item["status"]),
|
|
let eventType = Self.text(item["event_type"])
|
|
else {
|
|
return nil
|
|
}
|
|
let level: RuntimeLogLevel = status == "failed" ? .error : .info
|
|
return RuntimeLogEntry(
|
|
id: "delivery-\(instance.id.uuidString)-\(id)",
|
|
source: .delivery,
|
|
level: level,
|
|
message: Self.text(item["message"]) ?? "\(eventType) · \(status)",
|
|
eventType: eventType,
|
|
instanceID: instance.id,
|
|
instanceName: instance.name,
|
|
occurredAt: Self.date(item["created_at"]),
|
|
code: status,
|
|
durationMs: nil
|
|
)
|
|
}
|
|
}
|
|
|
|
private func request(_ route: UpstreamRoute, _ instance: Instance) async throws -> UpstreamResponse {
|
|
var cookie = await sessions.cookie(for: instance.id, origin: instance.origin)
|
|
do {
|
|
return try await client.send(route, baseURL: instance.origin, cookie: cookie)
|
|
} catch UpstreamError.unauthorized, UpstreamError.forbidden {
|
|
cookie = try await login(instance)
|
|
return try await client.send(route, baseURL: instance.origin, cookie: cookie)
|
|
}
|
|
}
|
|
|
|
private func login(_ instance: Instance) async throws -> String {
|
|
guard let credentials,
|
|
let password = try credentials.get(instanceID: instance.id),
|
|
!password.isEmpty
|
|
else {
|
|
throw UpstreamError.unauthorized
|
|
}
|
|
let service = InstanceSessionService(store: sessions, client: client)
|
|
let result = try await service.login(instance, password: password)
|
|
guard result.authenticated,
|
|
let cookie = await sessions.cookie(for: instance.id, origin: instance.origin)
|
|
else {
|
|
throw UpstreamError.unauthorized
|
|
}
|
|
return cookie
|
|
}
|
|
|
|
private func elapsed(_ startedAt: Date) -> Int {
|
|
max(0, Int(Date().timeIntervalSince(startedAt) * 1000))
|
|
}
|
|
|
|
private static func healthLabel(_ probe: ProbeOutcome) -> String {
|
|
switch probe.state {
|
|
case "online", "UP", "ok": return "在线"
|
|
case "auth_required": return "需登录"
|
|
case "unreachable": return probe.errorCode == "TIMEOUT" ? "超时" : "不可达"
|
|
default: return probe.state
|
|
}
|
|
}
|
|
|
|
private static func text(_ value: Any?) -> String? {
|
|
guard let value = value as? String, !value.isEmpty, value.count <= 512 else { return nil }
|
|
return value
|
|
}
|
|
|
|
private static func date(_ value: Any?) -> Date {
|
|
guard let text = value as? String else { return Date() }
|
|
let formatter = ISO8601DateFormatter()
|
|
formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds]
|
|
if let parsed = formatter.date(from: text) { return parsed }
|
|
formatter.formatOptions = [.withInternetDateTime]
|
|
return formatter.date(from: text) ?? Date()
|
|
}
|
|
}
|