import { clean, positiveInt, requestFlyApi } from './fly-core.ts'
import type { FlyLogsInput } from './types.ts'
const nanosecondsPerMillisecond = 1000000n
const httpLinePattern = /^(GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS)\s+(\S+)\s+(\d{3})\s+(?:\S+\s+)?-\s+([\d.]+)\s+ms/
const maxSafePages = 1000
const maxSafeEntries = 100000
function compactMessage(message, length = 300) {
return String(message ?? '').replace(/\s+/g, ' ').trim().slice(0, length)
}
function counterEntries(counter, limit = 20) {
return [...counter.entries()]
.sort((a, b) => b[1] - a[1] || String(a[0]).localeCompare(String(b[0])))
.slice(0, limit)
.map(([key, count]) => ({ key, count }))
}
function increment(counter, key, amount = 1) {
const normalizedKey = key || '(none)'
counter.set(normalizedKey, (counter.get(normalizedKey) ?? 0) + amount)
}
function parseHttpLine(message) {
const match = String(message ?? '').match(httpLinePattern)
if (!match) return null
return {
method: match[1],
url: match[2],
status: Number(match[3]),
durationMs: Number(match[4]),
route: `${match[1]} ${match[2]}`,
}
}
function parseTimestamp(value) {
const milliseconds = Date.parse(String(value ?? ''))
return Number.isFinite(milliseconds) ? milliseconds : null
}
function toNanosecondToken(value) {
const cleaned = clean(value)
if (!cleaned) return null
if (/^\d+$/.test(cleaned)) return cleaned
const milliseconds = parseTimestamp(cleaned)
if (milliseconds === null) throw new Error('Invalid timestamp: ' + cleaned)
return String(BigInt(milliseconds) * nanosecondsPerMillisecond)
}
function minuteOf(timestamp) {
return String(timestamp).slice(0, 16) + ':00Z'
}
function normalizeEventMessage(message) {
return compactMessage(message, 240)
.replace(/Instance [0-9a-f]+/g, 'Instance <id>')
.replace(/\b[0-9a-f]{12,}\b/g, '<hex>')
.replace(/\d+\.\d+ ms/g, '<ms> ms')
}
function classifyEvent(entry) {
const message = String(entry.message ?? '')
const lower = message.toLowerCase()
if (entry.method) {
if (entry.status !== null && entry.status >= 500) return 'http-5xx'
if (entry.level === 'error') return 'error'
if (entry.level === 'warn') return 'warning'
return null
}
if (entry.level === 'warn' && lower.includes('hard limit of 50 concurrent requests')) return 'concurrency-limit'
if (lower.includes('could not find a good candidate')) return 'no-routing-candidate'
if (lower.includes('machines api returned an error') && lower.includes('rate limit exceeded')) return 'machines-api-rate-limit'
if (lower.includes('lease currently held')) return 'machine-lease-held'
if (lower.includes('starting machine') || lower.includes('machine created and started')) return 'machine-start'
if (lower.includes('health check') && lower.includes('failed')) return 'health-check-failed'
if (lower.includes('health check') && lower.includes('passing')) return 'health-check-passing'
if (/sigterm|sigkill|exited|shutdown|reboot|restart|killed|oom/.test(lower)) return 'machine-stop-or-restart'
if (/^(deploy|deployed|deployment|release)\b/.test(lower) || /\bnew release\b|\brelease .*complete\b/.test(lower)) return 'deploy'
if (entry.level === 'error') return 'error'
if (entry.level === 'warn') return 'warning'
return null
}
function samplePush(samples, sample, maxSamples) {
if (samples.length < maxSamples) samples.push(sample)
}
export function normalizeLogEntry(entry) {
const attributes = entry?.attributes ?? entry ?? {}
const message = String(attributes.message ?? '')
const http = parseHttpLine(message)
const timestamp = attributes.timestamp ?? null
const meta = attributes.meta ?? {}
return {
id: entry?.id ?? attributes.id ?? null,
timestamp,
level: String(attributes.level ?? '').toLowerCase(),
instance: attributes.instance ?? meta.instance ?? null,
region: attributes.region ?? meta.region ?? null,
status: http?.status ?? null,
method: http?.method ?? null,
url: http?.url ?? null,
route: http?.route ?? null,
durationMs: http?.durationMs ?? null,
eventType: null,
message,
}
}
export function logsToJsonl(entries = []) {
return entries.map((entry) => JSON.stringify(entry)).join('\n') + (entries.length ? '\n' : '')
}
export function analyzeLogEntries(entries = [], options = {}) {
const sampleLimit = positiveInt(options.sampleLimit, 25, 200)
const gapSeconds = positiveInt(options.gapSeconds, 90, 3600)
const levels = new Map()
const statuses = new Map()
const routes = new Map()
const fiveXxRoutes = new Map()
const warnings = new Map()
const errors = new Map()
const eventCounts = new Map()
const minuteStats = new Map()
const instances = new Map()
const samples = []
const slowestRequests = []
const gaps = []
let firstTimestamp = null
let lastTimestamp = null
let previousTimestamp = null
const normalizedEntries = entries.map((entry) => {
const normalized = normalizeLogEntry(entry)
normalized.eventType = classifyEvent(normalized)
return normalized
})
for (const entry of normalizedEntries) {
if (!entry.timestamp) continue
const timestampMs = parseTimestamp(entry.timestamp)
firstTimestamp ??= entry.timestamp
lastTimestamp = entry.timestamp
increment(levels, entry.level)
if (previousTimestamp !== null && timestampMs !== null) {
const seconds = Math.round((timestampMs - previousTimestamp) / 1000)
if (seconds >= gapSeconds) {
gaps.push({ from: previousTimestamp === null ? null : new Date(previousTimestamp).toISOString(), to: entry.timestamp, seconds })
}
}
if (timestampMs !== null) previousTimestamp = timestampMs
const minute = minuteOf(entry.timestamp)
const minuteRow = minuteStats.get(minute) ?? {
minute,
total: 0,
http: 0,
s2xx: 0,
s3xx: 0,
s4xx: 0,
s5xx: 0,
warn: 0,
error: 0,
events: 0,
}
minuteRow.total += 1
if (entry.level === 'warn') minuteRow.warn += 1
if (entry.level === 'error') minuteRow.error += 1
if (entry.status !== null) {
minuteRow.http += 1
increment(statuses, String(entry.status))
increment(routes, `${entry.route} ${entry.status}`)
if (entry.status >= 500) {
minuteRow.s5xx += 1
increment(fiveXxRoutes, entry.route)
} else if (entry.status >= 400) {
minuteRow.s4xx += 1
} else if (entry.status >= 300) {
minuteRow.s3xx += 1
} else if (entry.status >= 200) {
minuteRow.s2xx += 1
}
}
if (entry.eventType) {
minuteRow.events += 1
increment(eventCounts, entry.eventType)
}
if (entry.level === 'warn') increment(warnings, normalizeEventMessage(entry.message))
if (entry.level === 'error') increment(errors, normalizeEventMessage(entry.message))
if (entry.instance) {
const instance = instances.get(entry.instance) ?? {
instance: entry.instance,
region: entry.region,
count: 0,
first: entry.timestamp,
last: entry.timestamp,
warnings: 0,
errors: 0,
s5xx: 0,
}
instance.count += 1
instance.last = entry.timestamp
if (entry.level === 'warn') instance.warnings += 1
if (entry.level === 'error') instance.errors += 1
if (entry.status !== null && entry.status >= 500) instance.s5xx += 1
instances.set(entry.instance, instance)
}
if (entry.durationMs !== null) {
slowestRequests.push({
timestamp: entry.timestamp,
route: entry.route,
status: entry.status,
durationMs: entry.durationMs,
})
}
if (entry.eventType || entry.level === 'warn' || entry.level === 'error' || (entry.status !== null && entry.status >= 500)) {
samplePush(samples, {
timestamp: entry.timestamp,
level: entry.level,
eventType: entry.eventType,
instance: entry.instance,
region: entry.region,
status: entry.status,
route: entry.route,
message: compactMessage(entry.message, 500),
}, sampleLimit)
}
minuteStats.set(minute, minuteRow)
}
const minuteRows = [...minuteStats.values()].sort((a, b) => a.minute.localeCompare(b.minute))
return {
total: normalizedEntries.length,
firstTimestamp,
lastTimestamp,
levelCounts: Object.fromEntries(levels),
statusCounts: Object.fromEntries(statuses),
eventCounts: Object.fromEntries(eventCounts),
topRoutes: counterEntries(routes, positiveInt(options.routeLimit, 20, 100)),
top5xxRoutes: counterEntries(fiveXxRoutes, positiveInt(options.routeLimit, 20, 100)),
topWarnings: counterEntries(warnings, positiveInt(options.groupLimit, 20, 100)),
topErrors: counterEntries(errors, positiveInt(options.groupLimit, 20, 100)),
anomalyMinutes: minuteRows.filter((row) => row.s5xx || row.error || row.warn || row.events),
gaps: gaps.sort((a, b) => b.seconds - a.seconds).slice(0, positiveInt(options.gapLimit, 20, 100)),
instances: [...instances.values()].sort((a, b) => a.first.localeCompare(b.first)),
slowestRequests: slowestRequests
.sort((a, b) => b.durationMs - a.durationMs)
.slice(0, positiveInt(options.slowRequestLimit, 20, 100)),
samples,
}
}
export async function fetchAppLogs(params = {}) {
const appName = clean(params.appName || params.app)
if (!appName) throw new Error('appName or app is required.')
const startToken = clean(params.nextToken || params.startToken) || toNanosecondToken(params.start || params.since)
const endMs = parseTimestamp(params.end || params.until)
const maxPages = positiveInt(params.maxPages, 20, maxSafePages)
const maxEntries = positiveInt(params.maxEntries, 2000, maxSafeEntries)
const entries = []
const seenIds = new Set()
let cursor = startToken
let pages = 0
let nextToken = null
let stoppedByEnd = false
for (let page = 0; page < maxPages; page += 1) {
const body = await requestFlyApi(`/api/v1/apps/${encodeURIComponent(appName)}/logs`, {
account: params.account,
secretName: params.secretName,
search: {
next_token: cursor,
region: params.region,
instance: params.instance || params.instanceId,
},
})
const pageEntries = Array.isArray(body?.data) ? body.data : []
nextToken = body?.meta?.next_token ?? null
pages += 1
if (pageEntries.length === 0) break
for (const entry of pageEntries) {
if (entry?.id && seenIds.has(entry.id)) continue
if (entry?.id) seenIds.add(entry.id)
const normalized = normalizeLogEntry(entry)
const timestampMs = parseTimestamp(normalized.timestamp)
if (endMs !== null && timestampMs !== null && timestampMs > endMs) {
stoppedByEnd = true
continue
}
entries.push(normalized)
if (entries.length >= maxEntries) {
stoppedByEnd = true
break
}
}
if (stoppedByEnd || !nextToken || nextToken === cursor || pageEntries.length < 100) break
cursor = nextToken
}
return {
appName,
window: {
start: params.start || params.since || null,
end: params.end || params.until || null,
},
paging: {
pages,
maxPages,
maxEntries,
nextToken,
stoppedByEnd,
truncated: pages >= maxPages || entries.length >= maxEntries,
},
entries,
}
}
/**
* Fetch Fly app logs for a time window and summarize outage-related signals.
* @param params.appName - Fly app name to read logs from.
* @param params.start - Window start timestamp (ISO string).
* @returns Log analysis summary; pass `format: 'jsonl'` to include raw JSONL.
* @example
* import logs from 'kody:@kody/fly/logs'
* const result = await logs({ appName: 'my-app', start: '2026-06-24T14:00:00Z', end: '2026-06-24T15:00:00Z' })
* // => { appName: 'my-app', analysis: { http5xx: 3, ... }, ... }
*/
export default async function appLogs(params: FlyLogsInput = {}) {
const result = await fetchAppLogs(params)
const analysis = analyzeLogEntries(result.entries, params)
const response = {
appName: result.appName,
window: result.window,
paging: result.paging,
analysis,
}
if (params.format === 'jsonl') {
return { ...response, jsonl: logsToJsonl(result.entries) }
}
if (params.includeEntries === true || params.format === 'entries') {
return { ...response, entries: result.entries }
}
return response
}