Raven Scott
bbf0607aaf
Release rolling / release (push) Successful in 9m13s
Lifecycle RPCs now use a 120s operation timeout, remove cleans up stats/terminal/log streams then SIGKILLs before force-remove, and the UI awaits the request instead of a fragile 30s wait race.
885 lines
28 KiB
JavaScript
885 lines
28 KiB
JavaScript
/**
|
||
* Container RPC handlers.
|
||
*/
|
||
import { PassThrough } from 'stream'
|
||
import { docker, startContainerNoBody, containerLifecycleNoBody } from '../services/docker.js'
|
||
import * as validation from '../utils/validation.js'
|
||
import { Pushes } from '../../shared/protocol.js'
|
||
import { peers } from '../core/peer-registry.js'
|
||
import { getHistory } from '../services/stats-history.js'
|
||
import { destroyStatsForContainer } from '../services/stats.js'
|
||
import { validateCreateOptions } from '../utils/engine-capabilities.js'
|
||
import logger from '../utils/logger.js'
|
||
|
||
/**
|
||
* True when Docker reports the container is already gone.
|
||
* @param {unknown} err
|
||
*/
|
||
function isNoSuchContainer(err) {
|
||
const status = err?.statusCode || err?.status
|
||
if (status === 404) return true
|
||
return /no such container/i.test(String(err?.message || err || ''))
|
||
}
|
||
|
||
/**
|
||
* Release attachments that can delay Docker force-remove, then delete.
|
||
* @param {string} id
|
||
* @param {import('../rpc/session.js').PeerSession} session
|
||
* @param {{ force?: boolean, v?: boolean, removeVolumes?: boolean, link?: boolean }} args
|
||
*/
|
||
async function forceRemoveContainer(id, session, args = {}) {
|
||
session._cleanupLogsForContainer?.(id)
|
||
session._cleanupTerminalsForContainer?.(id)
|
||
try {
|
||
destroyStatsForContainer(id)
|
||
} catch {
|
||
// ignore
|
||
}
|
||
|
||
const container = docker.getContainer(id)
|
||
const force = args.force !== false
|
||
const removeOpts = {
|
||
force,
|
||
v: Boolean(args.v || args.removeVolumes),
|
||
}
|
||
if (args.link) removeOpts.link = true
|
||
|
||
// SIGKILL first so remove does not wait on a stuck PID / long stop period
|
||
if (force) {
|
||
try {
|
||
await container.kill({ signal: 'SIGKILL' })
|
||
} catch (err) {
|
||
// not running / already dead — fine
|
||
if (!isNoSuchContainer(err) && !/is not running|already stopped/i.test(String(err?.message || ''))) {
|
||
logger.debug('pre-remove kill skipped', { id: id.slice(0, 12), error: err?.message })
|
||
}
|
||
}
|
||
}
|
||
|
||
try {
|
||
await container.remove(removeOpts)
|
||
} catch (err) {
|
||
if (isNoSuchContainer(err)) return
|
||
// Concurrent remove in progress — wait briefly, then treat as gone if inspect 404s
|
||
if (/already in progress|removal of container/i.test(String(err?.message || ''))) {
|
||
await new Promise((r) => setTimeout(r, 750))
|
||
try {
|
||
await container.inspect()
|
||
} catch (inspectErr) {
|
||
if (isNoSuchContainer(inspectErr)) return
|
||
}
|
||
}
|
||
throw err
|
||
}
|
||
}
|
||
|
||
export function registerContainerHandlers(session) {
|
||
session.respond('validateCreateOptions', async (args) => {
|
||
const result = await validateCreateOptions(args?.options || args || {}, {
|
||
strict: args?.strict !== false,
|
||
})
|
||
return { success: result.ok, type: 'validateCreateOptions', ...result }
|
||
})
|
||
|
||
session.respond('createContainer', async (args) => {
|
||
// Full Docker Engine create API passthrough with Engine-version validation
|
||
const name = args.name || args.Name
|
||
if (name) {
|
||
const sanitized = validation.sanitizeString(name, 63)
|
||
if (!validation.isValidContainerName(sanitized)) {
|
||
throw new Error('Invalid container name')
|
||
}
|
||
args.name = sanitized
|
||
}
|
||
const createOpts = { ...args }
|
||
delete createOpts.start
|
||
delete createOpts.skipValidation
|
||
|
||
if (args.skipValidation !== true) {
|
||
const check = await validateCreateOptions(createOpts, { strict: true })
|
||
if (!check.ok) {
|
||
const err = new Error(
|
||
`Create options incompatible with Engine API ${check.apiVersion}: ${check.errors.join('; ')}`
|
||
)
|
||
err.code = 'ENGINE_CAPABILITY'
|
||
throw err
|
||
}
|
||
}
|
||
|
||
// dockerode uses `name` as separate option
|
||
const container = await docker.createContainer(createOpts)
|
||
if (args.start) {
|
||
try {
|
||
// Empty-body start (Engine API ≥1.24 rejects non-empty start body)
|
||
await startContainerNoBody(container.id)
|
||
} catch (startErr) {
|
||
// Rollback: remove container if start failed after create
|
||
try {
|
||
await container.remove({ force: true })
|
||
} catch {
|
||
// ignore
|
||
}
|
||
throw new Error(`Container created but start failed (rolled back): ${startErr.message}`)
|
||
}
|
||
}
|
||
await broadcastContainers()
|
||
return {
|
||
success: true,
|
||
message: `Container created${args.start ? ' and started' : ''}`,
|
||
id: container.id,
|
||
data: { Id: container.id },
|
||
}
|
||
})
|
||
|
||
session.respond('listContainers', async (args = {}) => {
|
||
const listOpts = { all: args.all !== false }
|
||
if (args.filters) listOpts.filters = args.filters
|
||
// Convenience state filter → Docker filters
|
||
if (args.state && !listOpts.filters) {
|
||
listOpts.filters = { status: [String(args.state)] }
|
||
}
|
||
|
||
let containers = await docker.listContainers(listOpts)
|
||
|
||
if (args.name) {
|
||
const q = String(args.name).toLowerCase()
|
||
containers = containers.filter((c) =>
|
||
(c.Names || []).some((n) => n.toLowerCase().includes(q))
|
||
)
|
||
}
|
||
|
||
const total = containers.length
|
||
const offset = Math.max(0, Number(args.offset) || 0)
|
||
const limit = args.limit != null ? Math.min(Number(args.limit) || 50, 1000) : null
|
||
const page = limit != null ? containers.slice(offset, offset + limit) : containers.slice(offset)
|
||
|
||
// IP comes from list payload NetworkSettings — no N× inspect on every refresh
|
||
// (inspect was slow, raced with event pushes, and caused client list flicker).
|
||
const detailed = page.map((container) => ({
|
||
...container,
|
||
ipAddress: ipFromListContainer(container),
|
||
}))
|
||
return {
|
||
type: 'containers',
|
||
data: detailed,
|
||
total,
|
||
offset,
|
||
limit: limit ?? detailed.length,
|
||
hasMore: limit != null ? offset + limit < total : false,
|
||
}
|
||
})
|
||
|
||
session.respond('inspectContainer', async (args) => {
|
||
const config = await docker.getContainer(args.id).inspect()
|
||
return { type: 'containerConfig', data: config }
|
||
})
|
||
|
||
session.respond('startContainer', async (args) => {
|
||
// Raw unix-socket POST with Content-Length:0 (dockerode/bare-http can send a body)
|
||
const id = args.id || args.containerId || args.name
|
||
await startContainerNoBody(id)
|
||
return { success: true, message: `Container ${id} started` }
|
||
})
|
||
|
||
session.respond('stopContainer', async (args) => {
|
||
const opts = {}
|
||
if (args.t != null || args.timeout != null) {
|
||
opts.t = Number(args.t ?? args.timeout)
|
||
}
|
||
// Only pass opts when we have real stop parameters (avoid empty-body quirks)
|
||
if (Object.keys(opts).length) {
|
||
await docker.getContainer(args.id).stop(opts)
|
||
} else {
|
||
await docker.getContainer(args.id).stop()
|
||
}
|
||
return { success: true, message: `Container ${args.id} stopped` }
|
||
})
|
||
|
||
session.respond('restartContainer', async (args) => {
|
||
const opts = {}
|
||
if (args.t != null || args.timeout != null) {
|
||
opts.t = Number(args.t ?? args.timeout)
|
||
}
|
||
if (Object.keys(opts).length) {
|
||
await docker.getContainer(args.id).restart(opts)
|
||
} else {
|
||
await docker.getContainer(args.id).restart()
|
||
}
|
||
return { success: true, message: `Container ${args.id} restarted` }
|
||
})
|
||
|
||
session.respond('killContainer', async (args) => {
|
||
if (args.signal) {
|
||
await docker.getContainer(args.id).kill({ signal: String(args.signal) })
|
||
} else {
|
||
await docker.getContainer(args.id).kill()
|
||
}
|
||
return {
|
||
success: true,
|
||
message: `Container ${args.id} killed${args.signal ? ` (${args.signal})` : ''}`,
|
||
}
|
||
})
|
||
|
||
session.respond('pauseContainer', async (args) => {
|
||
await containerLifecycleNoBody(args.id, 'pause')
|
||
return { success: true, message: `Container ${args.id} paused` }
|
||
})
|
||
|
||
session.respond('unpauseContainer', async (args) => {
|
||
await containerLifecycleNoBody(args.id, 'unpause')
|
||
return { success: true, message: `Container ${args.id} unpaused` }
|
||
})
|
||
|
||
session.respond('removeContainer', async (args) => {
|
||
const id = args.id
|
||
await forceRemoveContainer(id, session, args)
|
||
return { success: true, message: `Container ${id} removed` }
|
||
})
|
||
|
||
session.respond('renameContainer', async (args) => {
|
||
const newName = validation.sanitizeString(args.name, 63)
|
||
if (!newName || !validation.isValidContainerName(newName)) {
|
||
throw new Error('Invalid container name. Must be alphanumeric with dashes/underscores, 1-63 characters.')
|
||
}
|
||
await docker.getContainer(args.id).rename({ name: newName })
|
||
return { success: true, message: `Container renamed to "${newName}"` }
|
||
})
|
||
|
||
session.respond('commitContainer', async (args) => {
|
||
const commitOptions = {
|
||
repo: validation.sanitizeString(args.repo, 255),
|
||
tag: validation.sanitizeString(args.tag || 'latest', 128),
|
||
}
|
||
if (args.message) commitOptions.comment = validation.sanitizeString(args.message, 500)
|
||
if (args.author) commitOptions.author = validation.sanitizeString(args.author, 255)
|
||
const image = await docker.getContainer(args.id).commit(commitOptions)
|
||
return {
|
||
success: true,
|
||
message: `Container committed as ${commitOptions.repo}:${commitOptions.tag}`,
|
||
data: image.id,
|
||
}
|
||
})
|
||
|
||
session.respond('exportContainer', async (args) => {
|
||
// Full container filesystem export as tar (size-capped, chunked push)
|
||
const maxBytes = Math.min(Number(args.maxBytes) || 50 * 1024 * 1024, 100 * 1024 * 1024)
|
||
const chunkSize = Math.min(Number(args.chunkSize) || 256 * 1024, 512 * 1024)
|
||
const stream = await docker.getContainer(args.id).export()
|
||
const transferId = `export-${String(args.id).slice(0, 12)}-${Date.now()}`
|
||
const chunks = []
|
||
let total = 0
|
||
await new Promise((resolve, reject) => {
|
||
stream.on('data', (chunk) => {
|
||
total += chunk.length
|
||
if (total > maxBytes) {
|
||
stream.destroy()
|
||
reject(new Error(`Export exceeds maxBytes (${maxBytes})`))
|
||
return
|
||
}
|
||
chunks.push(chunk)
|
||
})
|
||
stream.on('end', resolve)
|
||
stream.on('error', reject)
|
||
})
|
||
const buf = Buffer.concat(chunks)
|
||
let index = 0
|
||
for (let offset = 0; offset < buf.length; offset += chunkSize) {
|
||
const slice = buf.subarray(offset, Math.min(offset + chunkSize, buf.length))
|
||
session.push(Pushes.binaryChunk, {
|
||
type: 'binaryChunk',
|
||
transferId,
|
||
kind: 'containerExport',
|
||
containerId: args.id,
|
||
index,
|
||
totalBytes: buf.length,
|
||
encoding: 'base64',
|
||
data: slice.toString('base64'),
|
||
done: offset + chunkSize >= buf.length,
|
||
})
|
||
index += 1
|
||
}
|
||
return {
|
||
success: true,
|
||
type: 'containerExport',
|
||
transferId,
|
||
id: args.id,
|
||
size: buf.length,
|
||
chunks: index,
|
||
encoding: 'base64',
|
||
data: buf.length <= 2 * 1024 * 1024 ? buf.toString('base64') : null,
|
||
}
|
||
})
|
||
|
||
session.respond('attachContainer', async (args) => {
|
||
const container = docker.getContainer(args.id)
|
||
const stream = await container.attach({
|
||
stream: true,
|
||
stdin: args.stdin !== false,
|
||
stdout: true,
|
||
stderr: true,
|
||
logs: Boolean(args.logs),
|
||
})
|
||
const attachKey = `attach:${args.id}`
|
||
// Close previous attach for this container
|
||
const prev = session.state.get(attachKey)
|
||
if (prev?.stream) {
|
||
try {
|
||
prev.stream.end()
|
||
} catch {
|
||
// ignore
|
||
}
|
||
}
|
||
session.state.set(attachKey, { stream, containerId: args.id })
|
||
stream.on('data', (chunk) => {
|
||
session.push(Pushes.attachOutput, {
|
||
type: 'attachOutput',
|
||
containerId: args.id,
|
||
data: chunk.toString('base64'),
|
||
encoding: 'base64',
|
||
})
|
||
})
|
||
stream.on('end', () => {
|
||
session.state.delete(attachKey)
|
||
})
|
||
stream.on('error', (err) => {
|
||
logger.error('Attach stream error', { id: args.id, error: err.message })
|
||
session.state.delete(attachKey)
|
||
})
|
||
return { success: true, message: `Attached to container ${args.id}`, containerId: args.id }
|
||
})
|
||
|
||
session.respond('attachInput', async (args) => {
|
||
const key = `attach:${args.containerId || args.id}`
|
||
const entry = session.state.get(key)
|
||
if (!entry) throw new Error('No active attach session for container')
|
||
const inputData =
|
||
args.encoding === 'base64'
|
||
? Buffer.from(args.data, 'base64')
|
||
: Buffer.from(args.data || '', 'utf8')
|
||
entry.stream.write(inputData)
|
||
return { success: true }
|
||
})
|
||
|
||
session.respond('getStatsHistory', async (args) => {
|
||
if (!args.id) throw new Error('container id required')
|
||
const data = getHistory(args.id, {
|
||
limit: args.limit,
|
||
since: args.since,
|
||
})
|
||
return {
|
||
success: true,
|
||
type: 'statsHistory',
|
||
id: args.id,
|
||
data,
|
||
since: args.since || null,
|
||
count: data.length,
|
||
}
|
||
})
|
||
|
||
session.respond('getContainerLogs', async (args) => {
|
||
const id = args.id || args.containerId
|
||
if (!id) throw new Error('container id required')
|
||
const opts = {
|
||
stdout: args.stdout !== false,
|
||
stderr: args.stderr !== false,
|
||
timestamps: args.timestamps !== false,
|
||
tail: Math.min(Number(args.tail) || 500, 10000),
|
||
follow: false,
|
||
}
|
||
if (args.since != null) opts.since = args.since
|
||
if (args.until != null) opts.until = args.until
|
||
const buf = await docker.getContainer(id).logs(opts)
|
||
let text = Buffer.isBuffer(buf) ? demuxLogs(buf) : String(buf)
|
||
// Client-side style filters applied server-side for download
|
||
if (args.search) {
|
||
const q = String(args.search).toLowerCase()
|
||
text = text
|
||
.split('\n')
|
||
.filter((line) => line.toLowerCase().includes(q))
|
||
.join('\n')
|
||
}
|
||
if (args.level) {
|
||
const level = String(args.level).toLowerCase()
|
||
const patterns = {
|
||
error: /\b(error|err|fatal|panic)\b/i,
|
||
warn: /\b(warn|warning)\b/i,
|
||
info: /\b(info)\b/i,
|
||
debug: /\b(debug|trace)\b/i,
|
||
}
|
||
const re = patterns[level]
|
||
if (re) {
|
||
text = text
|
||
.split('\n')
|
||
.filter((line) => re.test(line))
|
||
.join('\n')
|
||
}
|
||
}
|
||
return {
|
||
success: true,
|
||
type: 'containerLogs',
|
||
id,
|
||
encoding: 'utf8',
|
||
size: text.length,
|
||
data: text,
|
||
}
|
||
})
|
||
|
||
session.respond('updateContainer', async (args) => {
|
||
const id = args.id
|
||
if (!id) throw new Error('Container id required')
|
||
|
||
// Docker Engine update API: HostConfig subset (CPU/memory/restart/…)
|
||
const update = {}
|
||
if (args.Memory != null || args.memory != null) {
|
||
const mem = args.Memory ?? args.memory
|
||
update.Memory = typeof mem === 'number' && mem < 1e6 ? mem * 1024 * 1024 : Number(mem)
|
||
}
|
||
if (args.MemoryReservation != null) update.MemoryReservation = Number(args.MemoryReservation)
|
||
if (args.MemorySwap != null) update.MemorySwap = Number(args.MemorySwap)
|
||
if (args.CpuShares != null) update.CpuShares = Number(args.CpuShares)
|
||
if (args.NanoCpus != null) update.NanoCpus = Number(args.NanoCpus)
|
||
if (args.CpuQuota != null) update.CpuQuota = Number(args.CpuQuota)
|
||
if (args.CpuPeriod != null) update.CpuPeriod = Number(args.CpuPeriod)
|
||
if (args.CpusetCpus != null) update.CpusetCpus = String(args.CpusetCpus)
|
||
if (args.CpusetMems != null) update.CpusetMems = String(args.CpusetMems)
|
||
if (args.BlkioWeight != null) update.BlkioWeight = Number(args.BlkioWeight)
|
||
if (args.RestartPolicy) {
|
||
update.RestartPolicy =
|
||
typeof args.RestartPolicy === 'string'
|
||
? { Name: args.RestartPolicy }
|
||
: args.RestartPolicy
|
||
}
|
||
if (args.PidsLimit != null) update.PidsLimit = Number(args.PidsLimit)
|
||
|
||
// Allow passthrough of raw HostConfig-style keys under args.update
|
||
if (args.update && typeof args.update === 'object') {
|
||
Object.assign(update, args.update)
|
||
}
|
||
|
||
if (Object.keys(update).length === 0) {
|
||
throw new Error(
|
||
'No update fields provided. Supported: Memory, CpuShares, NanoCpus, RestartPolicy, CpusetCpus, …'
|
||
)
|
||
}
|
||
|
||
const data = await docker.getContainer(id).update(update)
|
||
return {
|
||
success: true,
|
||
message: `Container ${id} updated`,
|
||
data,
|
||
applied: update,
|
||
}
|
||
})
|
||
|
||
session.respond('containerTop', async (args) => {
|
||
const opts = {}
|
||
if (args.ps_args || args.psArgs) opts.ps_args = args.ps_args || args.psArgs
|
||
const data = await docker.getContainer(args.id).top(opts)
|
||
return { success: true, type: 'containerTop', id: args.id, data }
|
||
})
|
||
|
||
session.respond('containerStats', async (args) => {
|
||
const stream = args.stream === true
|
||
const stats = await docker.getContainer(args.id).stats({ stream: false })
|
||
// dockerode returns a stream when stream:true; one-shot when false
|
||
if (stream) {
|
||
return {
|
||
success: true,
|
||
type: 'containerStats',
|
||
id: args.id,
|
||
note: 'Live stats are pushed via push:allStats; this is a one-shot snapshot',
|
||
data: stats,
|
||
}
|
||
}
|
||
return { success: true, type: 'containerStats', id: args.id, data: stats }
|
||
})
|
||
|
||
session.respond('waitContainer', async (args) => {
|
||
const opts = {}
|
||
if (args.condition) opts.condition = String(args.condition)
|
||
const result = await docker.getContainer(args.id).wait(opts)
|
||
return { success: true, type: 'containerWait', id: args.id, data: result }
|
||
})
|
||
|
||
session.respond('archiveContainerGet', async (args) => {
|
||
const path = validation.sanitizeString(args.path || '/', 4096)
|
||
if (!path || path.includes('..')) throw new Error('Invalid path')
|
||
const maxBytes = Math.min(Number(args.maxBytes) || 5 * 1024 * 1024, 20 * 1024 * 1024)
|
||
|
||
const stream = await docker.getContainer(args.id).getArchive({ path })
|
||
const chunks = []
|
||
let total = 0
|
||
await new Promise((resolve, reject) => {
|
||
stream.on('data', (chunk) => {
|
||
total += chunk.length
|
||
if (total > maxBytes) {
|
||
stream.destroy()
|
||
reject(new Error(`Archive exceeds maxBytes (${maxBytes})`))
|
||
return
|
||
}
|
||
chunks.push(chunk)
|
||
})
|
||
stream.on('end', resolve)
|
||
stream.on('error', reject)
|
||
})
|
||
|
||
const buf = Buffer.concat(chunks)
|
||
return {
|
||
success: true,
|
||
type: 'containerArchive',
|
||
id: args.id,
|
||
path,
|
||
encoding: 'base64',
|
||
size: buf.length,
|
||
data: buf.toString('base64'),
|
||
}
|
||
})
|
||
|
||
session.respond('archiveContainerPut', async (args) => {
|
||
const path = validation.sanitizeString(args.path || '/', 4096)
|
||
if (!path || path.includes('..')) throw new Error('Invalid path')
|
||
if (!args.data) throw new Error('Archive data (base64) required')
|
||
|
||
const buf = Buffer.from(args.data, args.encoding === 'utf8' ? 'utf8' : 'base64')
|
||
const maxBytes = 20 * 1024 * 1024
|
||
if (buf.length > maxBytes) throw new Error(`Archive exceeds ${maxBytes} bytes`)
|
||
|
||
await docker.getContainer(args.id).putArchive(buf, { path })
|
||
return {
|
||
success: true,
|
||
message: `Archive extracted to ${path} in container ${args.id}`,
|
||
size: buf.length,
|
||
}
|
||
})
|
||
|
||
session.respond('pruneContainers', async (args) => {
|
||
const opts = {}
|
||
if (args.filters) opts.filters = args.filters
|
||
const result = await docker.pruneContainers(opts)
|
||
await broadcastContainers()
|
||
return {
|
||
success: true,
|
||
type: 'pruneContainers',
|
||
message: 'Unused containers pruned',
|
||
data: result,
|
||
}
|
||
})
|
||
|
||
session.respond('bulkContainerOperation', async (args) => {
|
||
const { containerIds, operation } = args
|
||
if (!Array.isArray(containerIds) || containerIds.length === 0) {
|
||
throw new Error('No containers specified')
|
||
}
|
||
if (!['start', 'stop', 'restart', 'pause', 'unpause', 'remove', 'kill'].includes(operation)) {
|
||
throw new Error('Invalid operation')
|
||
}
|
||
|
||
const results = []
|
||
for (const containerId of containerIds) {
|
||
try {
|
||
const container = docker.getContainer(containerId)
|
||
switch (operation) {
|
||
case 'start':
|
||
await startContainerNoBody(containerId)
|
||
break
|
||
case 'stop':
|
||
await container.stop()
|
||
break
|
||
case 'restart':
|
||
await container.restart()
|
||
break
|
||
case 'pause':
|
||
await containerLifecycleNoBody(containerId, 'pause')
|
||
break
|
||
case 'unpause':
|
||
await containerLifecycleNoBody(containerId, 'unpause')
|
||
break
|
||
case 'kill':
|
||
if (args.signal) await container.kill({ signal: args.signal })
|
||
else await container.kill()
|
||
break
|
||
case 'remove':
|
||
await forceRemoveContainer(containerId, session, { force: true })
|
||
break
|
||
}
|
||
results.push({ id: containerId, success: true })
|
||
} catch (err) {
|
||
results.push({ id: containerId, success: false, error: err.message })
|
||
}
|
||
}
|
||
return { success: true, message: 'Bulk operation completed', results }
|
||
})
|
||
|
||
session.respond('duplicateContainer', async (args) => {
|
||
return duplicateContainer(args, session)
|
||
})
|
||
|
||
/**
|
||
* Recreate container with the same config (Portainer-style).
|
||
* Stops/removes the old container, creates a new one with the same name, starts if it was running.
|
||
*/
|
||
session.respond('recreateContainer', async (args) => {
|
||
return recreateContainer(args, session)
|
||
})
|
||
|
||
session.respond('execContainer', async (args) => {
|
||
const container = docker.getContainer(args.id)
|
||
const exec = await container.exec({
|
||
Cmd: args.cmd || ['/bin/sh'],
|
||
AttachStdin: true,
|
||
AttachStdout: true,
|
||
AttachStderr: true,
|
||
Tty: args.tty !== false,
|
||
})
|
||
const stream = await exec.start({ hijack: true, stdin: true })
|
||
const stdout = new PassThrough()
|
||
const stderr = new PassThrough()
|
||
container.modem.demuxStream(stream, stdout, stderr)
|
||
|
||
const execKey = `exec:${exec.id}`
|
||
session.state.set(execKey, { stream, exec, containerId: args.id })
|
||
|
||
stdout.on('data', (chunk) => {
|
||
session.push(Pushes.execOutput, {
|
||
type: 'execOutput',
|
||
containerId: args.id,
|
||
execId: exec.id,
|
||
data: chunk.toString('base64'),
|
||
encoding: 'base64',
|
||
})
|
||
})
|
||
stderr.on('data', (chunk) => {
|
||
session.push(Pushes.execErrorOutput, {
|
||
type: 'execErrorOutput',
|
||
containerId: args.id,
|
||
execId: exec.id,
|
||
data: chunk.toString('base64'),
|
||
encoding: 'base64',
|
||
})
|
||
})
|
||
|
||
return { success: true, message: 'Exec session started', execId: exec.id }
|
||
})
|
||
|
||
session.respond(
|
||
'execInput',
|
||
(args) => {
|
||
const key = `exec:${args.execId}`
|
||
const entry = session.state.get(key)
|
||
if (!entry) return null
|
||
const inputData =
|
||
args.encoding === 'base64'
|
||
? Buffer.from(args.data || '', 'base64')
|
||
: Buffer.from(args.data || '', 'utf8')
|
||
if (inputData.length && entry.stream && !entry.stream.writableEnded) {
|
||
entry.stream.write(inputData)
|
||
}
|
||
return null
|
||
},
|
||
{ hot: true }
|
||
)
|
||
}
|
||
|
||
async function duplicateContainer(args, session) {
|
||
const { name, image, hostname, netmode, cpu, memory, config } = args
|
||
const memoryInMB = memory * 1024 * 1024
|
||
|
||
const sanitizedConfig = { ...(config || {}) }
|
||
for (const key of [
|
||
'Id', 'State', 'Created', 'NetworkSettings', 'Mounts', 'Path', 'Args',
|
||
'Image', 'Hostname', 'CpuCount', 'Memory', 'CpuShares', 'CpusetCpus',
|
||
]) {
|
||
delete sanitizedConfig[key]
|
||
}
|
||
|
||
const existing = await docker.listContainers({ all: true })
|
||
if (existing.some((c) => c.Names.includes(`/${name}`))) {
|
||
throw new Error(`Container name '${name}' already exists.`)
|
||
}
|
||
|
||
const cpusetCpus = Array.from({ length: cpu }, (_, i) => i).join(',')
|
||
const nanoCpus = cpu * 1e9
|
||
|
||
const newContainer = await docker.createContainer({
|
||
...sanitizedConfig.Config,
|
||
name,
|
||
Hostname: hostname,
|
||
Image: image,
|
||
HostConfig: {
|
||
CpusetCpus: cpusetCpus.toString(),
|
||
NanoCpus: nanoCpus,
|
||
Memory: Number(memoryInMB),
|
||
MemoryReservation: Number(memoryInMB),
|
||
NetworkMode: String(netmode),
|
||
},
|
||
})
|
||
await startContainerNoBody(newContainer.id)
|
||
|
||
await broadcastContainers()
|
||
return { success: true, message: `Container '${name}' duplicated and started successfully.` }
|
||
}
|
||
|
||
/**
|
||
* Build dockerode create options from an inspect result.
|
||
* @param {object} inspect
|
||
* @param {string} name
|
||
*/
|
||
function createOptsFromInspect(inspect, name) {
|
||
const config = { ...(inspect.Config || {}) }
|
||
// Runtime-only / identity fields must not be passed to create
|
||
for (const key of [
|
||
'Hostname',
|
||
'Domainname',
|
||
'Image',
|
||
'AttachStdin',
|
||
'AttachStdout',
|
||
'AttachStderr',
|
||
'Tty',
|
||
'OpenStdin',
|
||
'StdinOnce',
|
||
]) {
|
||
// keep these intentionally from Config below
|
||
void key
|
||
}
|
||
|
||
const hostConfig = { ...(inspect.HostConfig || {}) }
|
||
// Drop empty/null host config noise that can break recreate
|
||
for (const [k, v] of Object.entries(hostConfig)) {
|
||
if (v === null || v === undefined || v === '' || (Array.isArray(v) && v.length === 0)) {
|
||
delete hostConfig[k]
|
||
}
|
||
}
|
||
|
||
const networks = inspect.NetworkSettings?.Networks || {}
|
||
const endpoints = {}
|
||
for (const [netName, conf] of Object.entries(networks)) {
|
||
if (!conf || netName === 'host' || netName === 'none') continue
|
||
endpoints[netName] = {
|
||
Aliases: conf.Aliases || undefined,
|
||
IPAMConfig: conf.IPAMConfig || undefined,
|
||
Links: conf.Links || undefined,
|
||
NetworkID: conf.NetworkID || undefined,
|
||
}
|
||
}
|
||
|
||
return {
|
||
name,
|
||
Image: config.Image,
|
||
Env: config.Env,
|
||
Cmd: config.Cmd,
|
||
Entrypoint: config.Entrypoint,
|
||
Labels: config.Labels,
|
||
WorkingDir: config.WorkingDir,
|
||
User: config.User,
|
||
Hostname: config.Hostname,
|
||
Domainname: config.Domainname,
|
||
Tty: config.Tty,
|
||
OpenStdin: config.OpenStdin,
|
||
StdinOnce: config.StdinOnce,
|
||
ExposedPorts: config.ExposedPorts,
|
||
Volumes: config.Volumes,
|
||
StopSignal: config.StopSignal,
|
||
StopTimeout: config.StopTimeout,
|
||
Healthcheck: config.Healthcheck,
|
||
HostConfig: hostConfig,
|
||
NetworkingConfig:
|
||
Object.keys(endpoints).length > 0 ? { EndpointsConfig: endpoints } : undefined,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* @param {{ id: string, start?: boolean, removeVolumes?: boolean, timeout?: number }} args
|
||
*/
|
||
async function recreateContainer(args, session) {
|
||
const id = args.id
|
||
if (!id) throw Object.assign(new Error('id required'), { code: 'INVALID_ARGS' })
|
||
|
||
const container = docker.getContainer(id)
|
||
const inspect = await container.inspect()
|
||
const name = String(inspect.Name || '').replace(/^\//, '') || id.slice(0, 12)
|
||
const wasRunning = Boolean(inspect.State?.Running)
|
||
const shouldStart = args.start !== false && (args.start === true || wasRunning)
|
||
|
||
try {
|
||
if (wasRunning) {
|
||
await container.stop({ t: Number(args.timeout) >= 0 ? Number(args.timeout) : 10 })
|
||
}
|
||
} catch {
|
||
// already stopped
|
||
}
|
||
|
||
await forceRemoveContainer(id, session, {
|
||
force: true,
|
||
v: Boolean(args.removeVolumes || args.v),
|
||
removeVolumes: Boolean(args.removeVolumes || args.v),
|
||
})
|
||
|
||
const createOpts = createOptsFromInspect(inspect, name)
|
||
const created = await docker.createContainer(createOpts)
|
||
|
||
if (shouldStart) {
|
||
await startContainerNoBody(created.id)
|
||
}
|
||
|
||
await broadcastContainers()
|
||
return {
|
||
success: true,
|
||
message: shouldStart
|
||
? `Container '${name}' recreated and started`
|
||
: `Container '${name}' recreated`,
|
||
id: created.id,
|
||
name,
|
||
started: shouldStart,
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Prefer NetworkSettings from docker list; fall back to inspect extract shape.
|
||
* @param {object} container
|
||
* @returns {string}
|
||
*/
|
||
function ipFromListContainer(container) {
|
||
const nets = container?.NetworkSettings?.Networks
|
||
if (nets && typeof nets === 'object') {
|
||
for (const net of Object.values(nets)) {
|
||
if (net?.IPAddress) return net.IPAddress
|
||
}
|
||
}
|
||
// Some list responses put IP on NetworkSettings.IPAddress
|
||
if (container?.NetworkSettings?.IPAddress) return container.NetworkSettings.IPAddress
|
||
return 'No IP Assigned'
|
||
}
|
||
|
||
export async function broadcastContainers() {
|
||
try {
|
||
const containers = await docker.listContainers({ all: true })
|
||
const data = containers.map((c) => ({
|
||
...c,
|
||
ipAddress: ipFromListContainer(c),
|
||
}))
|
||
peers.broadcast(Pushes.containers, { type: 'containers', data })
|
||
} catch (err) {
|
||
logger.error('Failed to broadcast containers', { error: err.message })
|
||
}
|
||
}
|
||
|
||
/** Strip docker multiplex headers from log buffers when possible. */
|
||
function demuxLogs(buffer) {
|
||
try {
|
||
const chunks = []
|
||
let offset = 0
|
||
while (offset + 8 <= buffer.length) {
|
||
const size = buffer.readUInt32BE(offset + 4)
|
||
if (size < 0 || offset + 8 + size > buffer.length) {
|
||
return buffer.toString('utf8')
|
||
}
|
||
chunks.push(buffer.subarray(offset + 8, offset + 8 + size).toString('utf8'))
|
||
offset += 8 + size
|
||
}
|
||
if (chunks.length) return chunks.join('')
|
||
} catch {
|
||
// fall through
|
||
}
|
||
return buffer.toString('utf8')
|
||
}
|