Files
peardock/server/handlers/containers.js
T
snxraven b62bcd697e Track E: recreate, system prune, Hub search, swarm secrets/configs
Investigate Docker/Portainer gaps and ship high-value ops: recreate
container, system prune + builder prune UI, Docker Hub search in
pull modal, Swarm secrets/configs tables. Update roadmap Track E/F.
2026-07-10 23:48:30 -04:00

787 lines
25 KiB
JavaScript

/**
* Container RPC handlers.
*/
import { PassThrough } from 'stream'
import { docker, extractIpAddress } 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 { validateCreateOptions } from '../utils/engine-capabilities.js'
import logger from '../utils/logger.js'
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 {
await container.start()
} 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)
const detailed = await Promise.all(
page.map(async (container) => {
try {
const details = await docker.getContainer(container.Id).inspect()
return { ...container, ipAddress: extractIpAddress(details) }
} catch (error) {
logger.error('Failed to inspect container', { id: container.Id, error: error.message })
return { ...container, ipAddress: 'Error Retrieving IP' }
}
})
)
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) => {
await docker.getContainer(args.id).start()
return { success: true, message: `Container ${args.id} started` }
})
session.respond('stopContainer', async (args) => {
const opts = {}
if (args.t != null || args.timeout != null) {
opts.t = Number(args.t ?? args.timeout)
}
await docker.getContainer(args.id).stop(opts)
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)
}
await docker.getContainer(args.id).restart(opts)
return { success: true, message: `Container ${args.id} restarted` }
})
session.respond('killContainer', async (args) => {
const opts = {}
if (args.signal) opts.signal = String(args.signal)
await docker.getContainer(args.id).kill(opts)
return {
success: true,
message: `Container ${args.id} killed${opts.signal ? ` (${opts.signal})` : ''}`,
}
})
session.respond('pauseContainer', async (args) => {
await docker.getContainer(args.id).pause()
return { success: true, message: `Container ${args.id} paused` }
})
session.respond('unpauseContainer', async (args) => {
await docker.getContainer(args.id).unpause()
return { success: true, message: `Container ${args.id} unpaused` }
})
session.respond('removeContainer', async (args) => {
const id = args.id
session._cleanupLogsForContainer?.(id)
await docker.getContainer(id).remove({
force: args.force !== false,
v: Boolean(args.v || args.removeVolumes),
link: Boolean(args.link),
})
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 })
return { success: true, type: 'statsHistory', id: args.id, data }
})
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 container.start()
break
case 'stop':
await container.stop()
break
case 'restart':
await container.restart()
break
case 'pause':
await container.pause()
break
case 'unpause':
await container.unpause()
break
case 'kill':
await container.kill(args.signal ? { signal: args.signal } : {})
break
case 'remove':
await container.remove({ 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 newContainer.start()
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)
session._cleanupLogsForContainer?.(id)
try {
if (wasRunning) {
await container.stop({ t: Number(args.timeout) >= 0 ? Number(args.timeout) : 10 })
}
} catch {
// already stopped
}
await container.remove({
force: true,
v: Boolean(args.removeVolumes || args.v),
})
const createOpts = createOptsFromInspect(inspect, name)
const created = await docker.createContainer(createOpts)
if (shouldStart) {
await created.start()
}
await broadcastContainers()
return {
success: true,
message: shouldStart
? `Container '${name}' recreated and started`
: `Container '${name}' recreated`,
id: created.id,
name,
started: shouldStart,
}
}
export async function broadcastContainers() {
try {
const containers = await docker.listContainers({ all: true })
peers.broadcast(Pushes.containers, { type: 'containers', data: containers })
} 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')
}