stream cron output
This commit is contained in:
parent
081c728d12
commit
47030d7d36
|
|
@ -438,9 +438,11 @@ app.get('/job/:app/:name', async c => {
|
|||
<Layout title={`${job.app}/${job.name}`} refresh={job.state === 'running'}>
|
||||
<BackLink href={backUrl}>← Back</BackLink>
|
||||
<DetailHeader>
|
||||
<div style={{ display: 'flex', alignItems: 'center', gap: '12px', flex: 1 }}>
|
||||
<StatusDot style={{ backgroundColor: statusColor(job) }} />
|
||||
<DetailTitle>{job.app}/{job.name}</DetailTitle>
|
||||
<StatusBadge style={statusBadgeStyle(job)}>{statusLabel(job)}</StatusBadge>
|
||||
</div>
|
||||
<form method="post" action={`/run/${job.app}/${job.name}?return=detail&app=${appFilter || ''}`}>
|
||||
<RunButton type="submit" disabled={job.state === 'running'}>
|
||||
{job.state === 'running' ? 'Running...' : 'Run Now'}
|
||||
|
|
@ -449,8 +451,8 @@ app.get('/job/:app/:name', async c => {
|
|||
</DetailHeader>
|
||||
<DetailMeta>
|
||||
<MetaItem><MetaLabel>Schedule</MetaLabel> {job.schedule}</MetaItem>
|
||||
<MetaItem><MetaLabel>Last run</MetaLabel> {formatRelative(job.lastRun)}</MetaItem>
|
||||
<MetaItem><MetaLabel>Duration</MetaLabel> {formatDuration(job.lastDuration)}</MetaItem>
|
||||
<MetaItem><MetaLabel>Last run</MetaLabel> {job.state === 'running' ? 'now' : formatRelative(job.lastRun)}</MetaItem>
|
||||
<MetaItem><MetaLabel>Duration</MetaLabel> {job.state === 'running' ? formatDuration(Date.now() - job.lastRun!) : formatDuration(job.lastDuration)}</MetaItem>
|
||||
<MetaItem><MetaLabel>Next run</MetaLabel> {formatRelative(job.nextRun)}</MetaItem>
|
||||
</DetailMeta>
|
||||
{job.lastError && (
|
||||
|
|
@ -459,17 +461,21 @@ app.get('/job/:app/:name', async c => {
|
|||
<ErrorBlock>{job.lastError}</ErrorBlock>
|
||||
</OutputSection>
|
||||
)}
|
||||
{job.lastOutput && (
|
||||
{job.lastOutput ? (
|
||||
<OutputSection>
|
||||
<OutputLabel>Output</OutputLabel>
|
||||
<OutputBlock>{job.lastOutput}</OutputBlock>
|
||||
</OutputSection>
|
||||
)}
|
||||
{!job.lastError && !job.lastOutput && job.lastRun && (
|
||||
) : job.state === 'running' ? (
|
||||
<OutputSection>
|
||||
<OutputLabel>Output</OutputLabel>
|
||||
<OutputBlock style={{ color: theme('colors-textMuted') }}>Waiting for output...</OutputBlock>
|
||||
</OutputSection>
|
||||
) : job.lastRun && !job.lastError ? (
|
||||
<OutputSection>
|
||||
<EmptyState>No output</EmptyState>
|
||||
</OutputSection>
|
||||
)}
|
||||
) : null}
|
||||
</Layout>
|
||||
)
|
||||
})
|
||||
|
|
|
|||
|
|
@ -3,13 +3,28 @@ import type { CronJob } from './schedules'
|
|||
import { getNextRun } from './scheduler'
|
||||
|
||||
const APPS_DIR = process.env.APPS_DIR!
|
||||
const TOES_DIR = process.env.TOES_DIR!
|
||||
const RUNNER = join(import.meta.dir, 'runner.ts')
|
||||
|
||||
async function readStream(stream: ReadableStream<Uint8Array>, append: (text: string) => void) {
|
||||
const reader = stream.getReader()
|
||||
const decoder = new TextDecoder()
|
||||
while (true) {
|
||||
const { done, value } = await reader.read()
|
||||
if (done) break
|
||||
append(decoder.decode(value, { stream: true }))
|
||||
}
|
||||
}
|
||||
|
||||
export async function executeJob(job: CronJob, onUpdate: () => void): Promise<void> {
|
||||
if (job.state === 'disabled') return
|
||||
|
||||
job.state = 'running'
|
||||
job.lastRun = Date.now()
|
||||
job.lastOutput = undefined
|
||||
job.lastError = undefined
|
||||
job.lastExitCode = undefined
|
||||
job.lastDuration = undefined
|
||||
onUpdate()
|
||||
|
||||
const cwd = join(APPS_DIR, job.app, 'current')
|
||||
|
|
@ -17,29 +32,35 @@ export async function executeJob(job: CronJob, onUpdate: () => void): Promise<vo
|
|||
try {
|
||||
const proc = Bun.spawn(['bun', 'run', RUNNER, job.file], {
|
||||
cwd,
|
||||
env: { ...process.env },
|
||||
env: { ...process.env, DATA_DIR: join(TOES_DIR, job.app) },
|
||||
stdout: 'pipe',
|
||||
stderr: 'pipe',
|
||||
})
|
||||
|
||||
const [stdout, stderr] = await Promise.all([
|
||||
new Response(proc.stdout).text(),
|
||||
new Response(proc.stderr).text(),
|
||||
// Stream output incrementally into job fields
|
||||
await Promise.all([
|
||||
readStream(proc.stdout as ReadableStream<Uint8Array>, text => {
|
||||
job.lastOutput = (job.lastOutput || '') + text
|
||||
}),
|
||||
readStream(proc.stderr as ReadableStream<Uint8Array>, text => {
|
||||
job.lastError = (job.lastError || '') + text
|
||||
}),
|
||||
])
|
||||
|
||||
const code = await proc.exited
|
||||
|
||||
job.lastDuration = Date.now() - job.lastRun
|
||||
job.lastExitCode = code
|
||||
job.lastError = code !== 0 ? stderr || 'Non-zero exit' : undefined
|
||||
job.lastOutput = stdout || undefined
|
||||
if (!job.lastError && code !== 0) job.lastError = 'Non-zero exit'
|
||||
if (code === 0) job.lastError = undefined
|
||||
if (!job.lastOutput) job.lastOutput = undefined
|
||||
job.state = 'idle'
|
||||
job.nextRun = getNextRun(job.id)
|
||||
|
||||
// Log result
|
||||
console.log(`[cron] ${job.id} finished: code=${code} duration=${job.lastDuration}ms`)
|
||||
if (stdout) console.log(stdout)
|
||||
if (stderr) console.error(stderr)
|
||||
if (job.lastOutput) console.log(job.lastOutput)
|
||||
if (job.lastError) console.error(job.lastError)
|
||||
} catch (e) {
|
||||
job.lastDuration = Date.now() - job.lastRun
|
||||
job.lastExitCode = 1
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user