Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
118 changes: 91 additions & 27 deletions packages/k8s/src/hooks/prepare-job.ts
Original file line number Diff line number Diff line change
Expand Up @@ -39,14 +39,25 @@ export async function prepareJob(
args: PrepareJobArgs,
responseFile
): Promise<void> {
core.debug('[prepareJob] Step 1: validating args.container')
if (!args.container) {
throw new Error('Job Container is required.')
}
core.debug(
`[prepareJob] container image: ${args.container.image}, services: ${args.services?.length ?? 0}`
)

core.debug('[prepareJob] Step 2: pruning stale pods')
await prunePods()
core.debug('[prepareJob] Step 2: prunePods done')

core.debug('[prepareJob] Step 3: reading extension from file')
const extension = readExtensionFromFile()
core.debug(
`[prepareJob] Step 3: extension loaded: ${extension ? 'yes' : 'none'}`
)

core.debug('[prepareJob] Step 4: building main container spec')
let container: k8s.V1Container | undefined = undefined
if (args.container?.image) {
container = createContainerSpec(
Expand All @@ -55,24 +66,33 @@ export async function prepareJob(
true,
extension
)
core.debug(`[prepareJob] Step 4: main container spec built: ${container.name}, image: ${container.image}`)
} else {
core.debug('[prepareJob] Step 4: no image specified, skipping main container')
}

core.debug('[prepareJob] Step 5: building service container specs')
let services: k8s.V1Container[] = []
if (args.services?.length) {
services = args.services.map(service => {
return createContainerSpec(
const spec = createContainerSpec(
service,
generateContainerName(service.image),
false,
extension
)
core.debug(`[prepareJob] Step 5: service container spec built: ${spec.name}, image: ${spec.image}`)
return spec
})
} else {
core.debug('[prepareJob] Step 5: no services')
}

if (!container && !services?.length) {
throw new Error('No containers exist, skipping hook invocation')
}

core.debug(`[prepareJob] Step 6: creating job pod (name: ${getJobPodName()})`)
let createdPod: k8s.V1Pod | undefined = undefined
try {
createdPod = await createJobPod(
Expand All @@ -82,78 +102,122 @@ export async function prepareJob(
args.container.registry,
extension
)
core.debug(`[prepareJob] Step 6: job pod created: ${createdPod?.metadata?.name}`)
} catch (err) {
await prunePods()
core.debug(`createPod failed: ${JSON.stringify(err)}`)
core.debug(`[prepareJob] Step 6 FAILED — createPod error: ${JSON.stringify(err)}`)
const message = (err as any)?.response?.body?.message || err
throw new Error(`failed to create job pod: ${message}`)
}

if (!createdPod?.metadata?.name) {
throw new Error('created pod should have metadata.name')
}

const podName = createdPod.metadata.name
const timeoutSecs = getPrepareJobTimeoutSeconds()
core.debug(
`Job pod created, waiting for it to come online ${createdPod?.metadata?.name}`
`[prepareJob] Step 7: waiting for pod "${podName}" to reach RUNNING phase (timeout: ${timeoutSecs}s)`
)

const runnerWorkspace = dirname(process.env.RUNNER_WORKSPACE as string)

let prepareScript: { containerPath: string; runnerPath: string } | undefined
if (args.container?.userMountVolumes?.length) {
prepareScript = prepareJobScript(args.container.userMountVolumes || [])
}

try {
await waitForPodPhases(
createdPod.metadata.name,
podName,
new Set([PodPhase.RUNNING]),
new Set([PodPhase.PENDING]),
getPrepareJobTimeoutSeconds()
timeoutSecs
)
core.debug(`[prepareJob] Step 7: pod "${podName}" is RUNNING`)
} catch (err) {
await prunePods()
throw new Error(`pod failed to come online with error: ${err}`)
}

await execCpToPod(createdPod.metadata.name, runnerWorkspace, '/__w')
const runnerWorkspace = dirname(process.env.RUNNER_WORKSPACE as string)
core.debug(
`[prepareJob] Step 8: copying runnerWorkspace "${runnerWorkspace}" to pod "${podName}" at /__w`
)
try {
await execCpToPod(podName, runnerWorkspace, '/__w')
core.debug('[prepareJob] Step 8: copy to /__w done')
} catch (err) {
core.debug(`[prepareJob] Step 8 FAILED — execCpToPod error: ${JSON.stringify(err)}`)
throw new Error(`failed to copy runner workspace to pod: ${err}`)
}

let prepareScript: { containerPath: string; runnerPath: string } | undefined
if (args.container?.userMountVolumes?.length) {
core.debug(
`[prepareJob] Step 9: generating prepareJobScript for ${args.container.userMountVolumes.length} userMountVolumes`
)
prepareScript = prepareJobScript(args.container.userMountVolumes || [])
core.debug(
`[prepareJob] Step 9: prepareScript containerPath: ${prepareScript.containerPath}`
)
} else {
core.debug('[prepareJob] Step 9: no userMountVolumes, skipping prepareScript')
}

if (prepareScript) {
await execPodStep(
['sh', '-e', prepareScript.containerPath],
createdPod.metadata.name,
JOB_CONTAINER_NAME
core.debug(
`[prepareJob] Step 10: executing prepareScript in pod "${podName}" container "${JOB_CONTAINER_NAME}"`
)
try {
await execPodStep(
['sh', '-e', prepareScript.containerPath],
podName,
JOB_CONTAINER_NAME
)
core.debug('[prepareJob] Step 10: prepareScript execution done')
} catch (err) {
core.debug(`[prepareJob] Step 10 FAILED — execPodStep error: ${JSON.stringify(err)}`)
throw new Error(`failed to execute prepareScript in pod: ${err}`)
}

core.debug('[prepareJob] Step 11: copying userMountVolumes to pod')
const promises: Promise<void>[] = []
for (const vol of args?.container?.userMountVolumes || []) {
core.debug(
`[prepareJob] Step 11: copying "${vol.sourceVolumePath}" -> "${vol.targetVolumePath}"`
)
promises.push(
execCpToPod(
createdPod.metadata.name,
podName,
vol.sourceVolumePath,
vol.targetVolumePath
)
).catch(err => {
core.debug(
`[prepareJob] Step 11 FAILED — copy "${vol.sourceVolumePath}" -> "${vol.targetVolumePath}": ${JSON.stringify(err)}`
)
throw new Error(
`failed to copy volume "${vol.sourceVolumePath}" to pod: ${err}`
)
})
)
}
await Promise.all(promises)
core.debug('[prepareJob] Step 11: all userMountVolumes copied')
}

core.debug('Job pod is ready for traffic')
core.debug('[prepareJob] Step 12: pod is ready for traffic')

core.debug(
`[prepareJob] Step 13: checking if container "${JOB_CONTAINER_NAME}" in pod "${podName}" is Alpine`
)
let isAlpine = false
try {
isAlpine = await isPodContainerAlpine(
createdPod.metadata.name,
JOB_CONTAINER_NAME
)
isAlpine = await isPodContainerAlpine(podName, JOB_CONTAINER_NAME)
core.debug(`[prepareJob] Step 13: isAlpine = ${isAlpine}`)
} catch (err) {
core.debug(
`Failed to determine if the pod is alpine: ${JSON.stringify(err)}`
`[prepareJob] Step 13 FAILED — isPodContainerAlpine error: ${JSON.stringify(err)}`
)
const message = (err as any)?.response?.body?.message || err
throw new Error(`failed to determine if the pod is alpine: ${message}`)
}
core.debug(`Setting isAlpine to ${isAlpine}`)

core.debug(`[prepareJob] Step 14: writing response file "${responseFile}"`)
generateResponseFile(responseFile, args, createdPod, isAlpine)
core.debug('[prepareJob] Step 14: response file written — prepareJob complete')
}

function generateResponseFile(
Expand Down
29 changes: 28 additions & 1 deletion packages/k8s/src/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,36 @@ async function run(): Promise<void> {
throw new Error(`Command not recognized: ${command}`)
}
} catch (error) {
core.error(error as Error)
const errMsg = error instanceof Error ? error.stack || error.message : String(error)
core.error('='.repeat(60))
core.error('HOOK FATAL ERROR — PrepareJob failed')
core.error('='.repeat(60))
core.error(errMsg)
core.error('='.repeat(60))
core.error('Sleeping 10 minutes to allow pod inspection for debugging...')
await new Promise(resolve => setTimeout(resolve, 10 * 60 * 1000))
process.exit(1)
}
}

process.on('SIGTERM', () => {
core.error('[GLOBAL] SIGTERM received — process killed by runner (likely hung too long)')
process.exit(143)
})

process.on('uncaughtException', (err: Error) => {
const e = err as NodeJS.ErrnoException
core.error(`[GLOBAL] uncaughtException: code=${e.code}, message=${e.message}`)
core.error(e.stack ?? String(e))
process.exit(1)
})

process.on('unhandledRejection', (reason: unknown) => {
const err = reason instanceof Error ? reason : new Error(String(reason))
const e = err as NodeJS.ErrnoException
core.error(`[GLOBAL] unhandledRejection: code=${e.code}, message=${e.message}`)
core.error(e.stack ?? String(e))
process.exit(1)
})

void run()
100 changes: 93 additions & 7 deletions packages/k8s/src/k8s/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -391,16 +391,65 @@ export async function execCpToPod(
const exec = new k8s.Exec(kc)
// Use tar to extract with --no-same-owner to avoid ownership issues.
// Then use find to fix permissions. The -m flag helps but we also need to fix permissions after.
// stderr is NOT suppressed on tar so errors are visible in the runner debug log.
const command = [
'sh',
'-c',
`tar xf - --no-same-owner -C ${shlex.quote(containerPath)} 2>/dev/null; ` +
`tar xf - --no-same-owner -C ${shlex.quote(containerPath)}; ` +
`find ${shlex.quote(containerPath)} -type f -exec chmod u+rw {} \\; 2>/dev/null; ` +
`find ${shlex.quote(containerPath)} -type d -exec chmod u+rwx {} \\; 2>/dev/null`
]
const readStream = tar.pack(runnerPath)

// Pre-pack diagnostic: recursively find special files (socket/fifo/device) under runnerPath.
// Top-level check is insufficient — blocking files are often in subdirectories.
await new Promise<void>(resolve => {
const findProc = spawn('find', [
runnerPath, '-not', '-type', 'f',
'-not', '-type', 'd',
'-not', '-type', 'l'
], { stdio: ['ignore', 'pipe', 'ignore'] })
let found = ''
findProc.stdout.on('data', (chunk: Buffer) => { found += chunk.toString() })
findProc.on('close', () => {
core.debug(
`[execCpToPod] recursive special files under "${runnerPath}": ${found.trim() || 'none'}`
)
resolve()
})
findProc.on('error', () => resolve())
})

// Track the last entry being packed so we know which file caused an error.
let lastPackedEntry = '<none yet>'
const readStream = tar.pack(runnerPath, {
map: (header: tar.Headers) => {
lastPackedEntry = header.name
return header
}
})
const errStream = new WritableStreamBuffer()

await new Promise((resolve, reject) => {
// Guard against double-settle: statusCallback, ws.close, and ws.error may all fire.
let settled = false
const safeResolve = (v: unknown): void => { if (!settled) { settled = true; resolve(v) } }
const safeReject = (e: unknown): void => { if (!settled) { settled = true; reject(e) } }

// Without this listener, an error from tar.pack() (e.g. unreadable file)
// would be an unhandled 'error' event and crash the Node.js process.
readStream.on('error', (err: Error) => {
const e = err as NodeJS.ErrnoException
core.debug(
`[execCpToPod] tar.pack error: code=${e.code}, path="${e.path}", lastEntry="${lastPackedEntry}", message="${e.message}"`
)
safeReject(new Error(`tar.pack error [${e.code}] on "${e.path ?? lastPackedEntry}": ${e.message}`))
})

readStream.on('end', () => {
core.debug(`[execCpToPod] tar.pack stream ended, lastEntry="${lastPackedEntry}"`)
})

core.debug('[execCpToPod] calling exec.exec() to open WebSocket...')
exec
.exec(
namespace(),
Expand All @@ -412,17 +461,54 @@ export async function execCpToPod(
readStream,
false,
async status => {
if (errStream.size()) {
reject(
const stderr = errStream.getContentsAsString() || ''
core.debug(
`[execCpToPod] exec callback: status=${JSON.stringify(status)}, lastEntry="${lastPackedEntry}", stderr=${stderr || '(empty)'}`
)
if (status.status === 'Failure' || errStream.size()) {
safeReject(
new Error(
`Error from execCpToPod - status: ${status.status}, details: \n ${errStream.getContentsAsString()}`
`Error from execCpToPod - status: ${JSON.stringify(status)}, stderr:\n${stderr}`
)
)
return
}
resolve(status)
safeResolve(status)
}
)
.catch(e => reject(e))
.then(ws => {
core.debug(
`[execCpToPod] WebSocket established, protocol="${ws.protocol}", streaming tar to pod...`
)
// @kubernetes/client-node does NOT call statusCallback when the WebSocket
// closes without a status frame. This happens with k8s protocol < v5.channel.k8s.io:
// when stdin ends, handleStandardInput() calls ws.close() directly, which closes
// the entire WebSocket before the server can send the status on channel 3.
ws.on('close', (code: number, reason: Buffer) => {
const reasonStr = reason?.toString() || ''
core.debug(
`[execCpToPod] WebSocket closed: code=${code}, reason="${reasonStr}", settled=${settled}`
)
if (code === 1000) {
// Normal close — exec command likely finished. statusCallback may not fire
// if the server protocol < v5.channel.k8s.io. Resolve here and let the
// hash-verification loop below confirm the copy actually landed in the pod.
safeResolve(undefined)
return
}
safeReject(
new Error(`WebSocket closed unexpectedly: code=${code}, reason="${reasonStr}"`)
)
})
ws.on('error', (err: Error) => {
core.debug(`[execCpToPod] WebSocket error: ${err}`)
safeReject(err)
})
})
.catch(e => {
core.debug(`[execCpToPod] exec.exec() rejected: ${e}`)
safeReject(e)
})
})
break
} catch (error) {
Expand Down