Files
Checkmate/server/src/service/network/DockerProvider.ts
T

497 lines
20 KiB
TypeScript

import https from "node:https";
import { IStatusProvider } from "@/service/network/IStatusProvider.js";
import { HttpProvider } from "@/service/network/HttpProvider.js";
import { CaptureDockerContainer, CaptureDockerStatusPayload, CheckContext, DockerStatusPayload, MonitorStatusResponse } from "@/types/network.js";
import {
DOCKER_LOG_TAIL_LINES,
DockerContainerInfo,
DockerContainerLogs,
DockerContainerMount,
DockerContainerPort,
DockerContainerState,
DockerContainerStates,
DockerContainerSummary,
DockerHealthStatus,
DockerHealthStatuses,
DockerLogLine,
DockerLogStream,
DockerPortProtocol,
DockerPortProtocols,
} from "@/domain/docker/docker.type.js";
import { Monitor, MonitorType } from "@/domain/monitors/monitor.type.js";
import { ILogger } from "@/utils/logger.js";
import { AppError } from "@/utils/AppError.js";
import Dockerode from "dockerode";
import { timeRequest, peerAnsweredFromError } from "@/service/network/utils.js";
import { NETWORK_ERROR } from "@/types/network.js";
import { IEncryptionService } from "@/service/encryption/encryptionService.js";
import { DOCKER_TLS_URL, isCaptureDockerUrl, isDockerTlsUrl } from "@/utils/dockerHost.js";
import { splitCertificateBundle } from "@/utils/pem.js";
type DockerodeType = typeof Dockerode;
type DockerOptions = Dockerode.DockerOptions & { agent?: https.Agent; connectionTimeout?: number };
type TlsCredentials = { ok: true; key: string } | { ok: false; message: string };
const SERVICE_NAME = "DockerProvider";
const TIMEOUT_MS = 10000;
const STATS_CONCURRENCY = 5;
const DOCKER_LOG_MAX_LINE_BYTES = 4096;
const DOCKER_LOG_TRUNCATION_MARKER = " …[truncated]";
const DOCKER_LOG_TS_REGEX = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z /;
const DOCKER_TLS_DEFAULT_PORT = 2376;
const CAPTURE_FATAL_ERROR_METRICS = new Set(["docker.client", "docker.container.list"]);
const EXITED_STATUS = /^Exited \((\d+)\)/;
const HEALTH_STATUS = /\((healthy|unhealthy|health: starting)\)$/;
export interface DockerError extends Error {
statusCode?: number;
reason?: string;
json?: { message?: string };
}
export class DockerProvider implements IStatusProvider<DockerStatusPayload> {
readonly type = "docker";
constructor(
private logger: ILogger,
private DockerLib: DockerodeType,
private encryptionService: IEncryptionService,
private httpProvider: HttpProvider
) {}
supports(type: MonitorType): boolean {
return type === "docker";
}
private resolveTlsCredentials = (monitor: Monitor, ctx?: CheckContext): TlsCredentials | undefined => {
if (!isDockerTlsUrl(monitor.url)) return undefined;
if (!ctx?.dockerTlsKey) return { ok: false, message: "Docker TLS key is missing" };
try {
return { ok: true, key: this.encryptionService.decrypt(ctx.dockerTlsKey) };
} catch (error) {
if (error instanceof AppError) {
const keyId = typeof error.details?.keyId === "string" ? ` (key id ${error.details.keyId})` : "";
return { ok: false, message: `${error.message}${keyId}` };
}
return { ok: false, message: "Docker TLS key could not be decrypted" };
}
};
private isDockerError(error: unknown): error is DockerError {
return error instanceof Error && ("statusCode" in error || "reason" in error || "json" in error);
}
private splitCaBundle = (pem: string | undefined): string[] | undefined => {
const blocks = pem ? splitCertificateBundle(pem) : [];
return blocks.length > 0 ? blocks : undefined;
};
private invalidUrl = (message: string) => new AppError({ message, status: 422, service: SERVICE_NAME, method: "toDockerOptions" });
private toDockerOptions = (monitor: Monitor, credentials?: TlsCredentials): DockerOptions => {
const url = monitor.url?.trim();
if (!url) throw this.invalidUrl("Docker host URL is required");
// Unix socket
if (url.startsWith("unix://")) {
const socketPath = url.slice("unix://".length);
if (!socketPath.startsWith("/")) throw this.invalidUrl(`Invalid Docker host URL: ${url}`);
return { socketPath, timeout: TIMEOUT_MS, connectionTimeout: TIMEOUT_MS };
}
if (url.startsWith("/")) return { socketPath: url, timeout: TIMEOUT_MS, connectionTimeout: TIMEOUT_MS };
// Docker TLS
const match = DOCKER_TLS_URL.exec(url);
if (!match) throw this.invalidUrl(`Invalid Docker host URL: ${url}`);
if (!credentials?.ok) throw this.invalidUrl(credentials?.message ?? "Docker TLS key is missing");
const [, , host, port] = match;
const ca = monitor.ignoreTlsErrors ? undefined : this.splitCaBundle(monitor.dockerTlsCa);
const cert = monitor.dockerTlsCert;
const key = credentials.key;
return {
host,
port: port ? Number(port) : DOCKER_TLS_DEFAULT_PORT,
protocol: "https",
ca,
cert,
key,
agent: new https.Agent({ ca, cert, key, rejectUnauthorized: !monitor.ignoreTlsErrors }),
socketPath: undefined,
username: undefined,
sshOptions: undefined,
timeout: TIMEOUT_MS,
connectionTimeout: TIMEOUT_MS,
};
};
private async mapWithConcurrency<T, R>(items: T[], limit: number, fn: (item: T) => Promise<R>): Promise<R[]> {
const results: R[] = [];
for (let i = 0; i < items.length; i += limit) {
results.push(...(await Promise.all(items.slice(i, i + limit).map(fn))));
}
return results;
}
private toContainerState(state: string): DockerContainerState {
return (DockerContainerStates as readonly string[]).includes(state) ? (state as DockerContainerState) : "created";
}
private toHealthStatus(status: string | undefined): DockerHealthStatus {
return status && (DockerHealthStatuses as readonly string[]).includes(status) ? (status as DockerHealthStatus) : "none";
}
private exitCodeFromStatus = (status: string): number | undefined => {
const match = EXITED_STATUS.exec(status);
return match ? Number(match[1]) : undefined;
};
private healthFromStatus = (status: string): DockerHealthStatus => {
const match = HEALTH_STATUS.exec(status);
if (!match) return "none";
return match[1] === "health: starting" ? "starting" : (match[1] as DockerHealthStatus);
};
private toContainerSummary(containers: DockerContainerInfo[]): DockerContainerSummary {
return {
total: containers.length,
running: containers.filter((container) => container.state === "running").length,
stopped: containers.filter((container) => container.state === "exited" || container.state === "dead").length,
unhealthy: containers.filter((container) => container.health === "unhealthy").length,
};
}
private toCaptureContainer(container: CaptureDockerContainer): DockerContainerInfo {
const startedAt = container.started_at && container.started_at > 0 ? new Date(container.started_at * 1000).toISOString() : undefined;
// Capture reports zeroed stats for stopped containers; only running ones carry real metrics, matching toContainerInfo.
const stats =
container.status === "running" && container.stats
? {
cpuPct: container.stats.cpu_percent / 100,
memoryUsedBytes: container.stats.memory_usage,
memoryLimitBytes: container.stats.memory_limit,
memoryPct: container.stats.memory_percent / 100,
}
: {};
return {
id: container.container_id,
name: container.container_name || container.container_id.slice(0, 12),
image: container.base_image,
state: this.toContainerState(container.status),
status: container.status,
// Capture derives health from state when no Docker healthcheck exists.
// Keep Checkmate's `none` value in that case so summaries retain their existing semantics.
health: container.health?.source === "container_health_check" ? (container.health.healthy ? "healthy" : "unhealthy") : "none",
startedAt,
ports: (container.exposed_ports ?? []).flatMap((port) => {
const privatePort = Number(port.port);
return Number.isInteger(privatePort) ? [{ privatePort, protocol: this.toPortProtocol(port.protocol) }] : [];
}),
...stats,
};
}
private isCapturePayload = (payload: unknown): payload is CaptureDockerStatusPayload => {
if (!payload || typeof payload !== "object") return false;
const { data, errors } = payload as Partial<CaptureDockerStatusPayload>;
if (!Array.isArray(data) && data !== null) return false;
if (errors === undefined || errors === null) return true;
return Array.isArray(errors) && errors.every((error) => Array.isArray(error?.metric));
};
private handleCapture = async (monitor: Monitor, ctx?: CheckContext): Promise<MonitorStatusResponse<DockerStatusPayload>> => {
const captureUrl = new URL(monitor.url.trim());
captureUrl.searchParams.set("all", "true");
const response = await this.httpProvider.handle<CaptureDockerStatusPayload>({ ...monitor, url: captureUrl.toString() }, ctx);
if (!response.status) return this.failed(monitor, response.code, response.message, response.responseTime ?? 0);
const payload = response.payload;
if (!this.isCapturePayload(payload)) {
this.logger.warn({
message: "Capture returned an invalid Docker metrics payload",
service: SERVICE_NAME,
method: "handleCapture",
});
return this.failed(monitor, NETWORK_ERROR, "Capture returned an invalid Docker metrics payload", response.responseTime ?? 0);
}
const errors = payload.errors ?? [];
const fatalError = errors.find((error) => error.metric.some((metric) => CAPTURE_FATAL_ERROR_METRICS.has(metric)));
if (fatalError) return this.failed(monitor, NETWORK_ERROR, fatalError.err, response.responseTime ?? 0);
if (errors.length > 0) {
this.logger.warn({
message: `Capture returned ${errors.length} partial Docker collection error(s)`,
service: SERVICE_NAME,
method: "handleCapture",
});
}
const containers = (payload.data ?? []).map((container) => this.toCaptureContainer(container));
return {
monitorId: monitor.id,
teamId: monitor.teamId,
type: monitor.type,
status: true,
code: response.code,
message: "Docker host is reachable through Capture",
responseTime: response.responseTime,
payload: { containers, summary: this.toContainerSummary(containers) },
};
};
private computeCpuPct(stats: Dockerode.ContainerStats): number {
const cpuDelta = stats.cpu_stats.cpu_usage.total_usage - stats.precpu_stats.cpu_usage.total_usage;
const systemDelta = stats.cpu_stats.system_cpu_usage - (stats.precpu_stats.system_cpu_usage ?? 0);
const onlineCpus = stats.cpu_stats.online_cpus ?? stats.cpu_stats.cpu_usage.percpu_usage?.length ?? 1;
if (!(systemDelta > 0) || cpuDelta < 0) return 0;
return (cpuDelta / systemDelta) * onlineCpus;
}
private computeMemory(stats: Dockerode.ContainerStats): Pick<DockerContainerInfo, "memoryUsedBytes" | "memoryLimitBytes" | "memoryPct"> {
const raw = stats.memory_stats.usage ?? 0;
const pageCache = stats.memory_stats.stats?.inactive_file ?? stats.memory_stats.stats?.cache ?? 0;
const memoryUsedBytes = Math.max(0, raw - pageCache);
const memoryLimitBytes = stats.memory_stats.limit ?? 0;
return {
memoryUsedBytes,
memoryLimitBytes,
memoryPct: memoryLimitBytes > 0 ? memoryUsedBytes / memoryLimitBytes : 0,
};
}
private toContainerName = (summary: Dockerode.ContainerInfo): string => {
return summary.Names?.[0]?.replace(/^\//, "") ?? summary.Id.slice(0, 12);
};
private async toContainerInfo(docker: Dockerode, summary: Dockerode.ContainerInfo): Promise<DockerContainerInfo> {
const info: DockerContainerInfo = {
id: summary.Id,
name: this.toContainerName(summary),
image: summary.Image,
state: this.toContainerState(summary.State),
status: summary.Status,
health: "none",
};
const container = docker.getContainer(summary.Id);
const [inspectResult, statsResult] = await Promise.allSettled([
container.inspect(),
summary.State === "running" ? container.stats({ stream: false }) : Promise.resolve(null),
]);
if (inspectResult.status === "fulfilled") {
info.restartCount = inspectResult.value.RestartCount;
info.startedAt = inspectResult.value.State?.StartedAt;
info.health = this.toHealthStatus(inspectResult.value.State?.Health?.Status);
info.ports = this.toPorts(inspectResult.value.NetworkSettings?.Ports);
info.mounts = this.toMounts(inspectResult.value.Mounts);
info.exitCode = inspectResult.value.State?.ExitCode;
} else {
info.exitCode = this.exitCodeFromStatus(summary.Status);
info.health = this.healthFromStatus(summary.Status);
this.logger.warn({
message: `Failed to inspect container ${info.name}`,
service: SERVICE_NAME,
method: "toContainerInfo",
details: { containerId: summary.Id },
});
}
if (statsResult.status === "fulfilled" && statsResult.value) {
info.cpuPct = this.computeCpuPct(statsResult.value);
Object.assign(info, this.computeMemory(statsResult.value));
}
return info;
}
private toPortProtocol(protocol: string): DockerPortProtocol {
return (DockerPortProtocols as readonly string[]).includes(protocol) ? (protocol as DockerPortProtocol) : "tcp";
}
private toPorts(ports: Dockerode.ContainerInspectInfo["NetworkSettings"]["Ports"] | undefined): DockerContainerPort[] {
const result: DockerContainerPort[] = [];
for (const [portAndProtocol, bindings] of Object.entries(ports ?? {})) {
const [portString, protocolString = ""] = portAndProtocol.split("/");
const privatePort = Number(portString);
if (!Number.isInteger(privatePort)) continue;
const protocol = this.toPortProtocol(protocolString);
// Exposed but unpublished ports have no bindings
if (!bindings || bindings.length === 0) {
result.push({ privatePort, protocol });
continue;
}
for (const binding of bindings) {
const publicPort = Number(binding.HostPort);
result.push({
privatePort,
protocol,
publicPort: Number.isInteger(publicPort) ? publicPort : undefined,
hostIp: binding.HostIp || undefined,
});
}
}
return result;
}
private toMounts(mounts: Dockerode.ContainerInspectInfo["Mounts"] | undefined): DockerContainerMount[] {
return (mounts ?? []).map((mount) => ({
type: mount.Type ?? "",
name: mount.Name,
source: mount.Source ?? "",
destination: mount.Destination ?? "",
mode: mount.Mode ?? "",
rw: mount.RW ?? true,
}));
}
// Log parsing
private isMultiplexed(buffer: Buffer): boolean {
return buffer.length >= 8 && buffer.readUInt8(0) <= 2 && buffer.readUInt8(1) === 0 && buffer.readUInt8(2) === 0 && buffer.readUInt8(3) === 0;
}
// Docker will will merge stdout and stderr if stared with `-t`.
// Most containers will _not_ be run with `-t` though, in which case logs come in frames of [header][payload]
private demuxLogBuffer = (buffer: Buffer): { stream: DockerLogStream; chunk: string }[] => {
if (!this.isMultiplexed(buffer)) return [{ stream: "stdout", chunk: buffer.toString("utf8") }]; // `-t` case, log is just a string, return it
const frames: { stream: DockerLogStream; chunk: string }[] = [];
let offset = 0; // Start reading at the first byte of the first header
while (offset + 8 <= buffer.length) {
// Every 8 bytes is a header, if < 8 bytes at the end, invalid frame, stop
const streamType = buffer.readUint8(offset); // First byte of the header is the stream type: 0 = stdin, 1 = stdout, 2 = stderr
const payloadSize = buffer.readUint32BE(offset + 4); // Byte 4 to 7 describe how many payload bytes follow the header
const payloadStart = offset + 8; // Advance to the payload (header 8 bytes long)
const payloadEnd = Math.min(payloadStart + payloadSize, buffer.length); // End is the start of the payload + payload size.
frames.push({ stream: streamType === 2 ? "stderr" : "stdout", chunk: buffer.toString("utf-8", payloadStart, payloadEnd) });
offset = payloadEnd;
}
return frames;
};
private truncateLog = (text: string): string => {
if (Buffer.byteLength(text, "utf8") <= DOCKER_LOG_MAX_LINE_BYTES) return text;
const cut = Buffer.from(text, "utf8").subarray(0, DOCKER_LOG_MAX_LINE_BYTES).toString("utf8").replace(/�$/, "");
return cut + DOCKER_LOG_TRUNCATION_MARKER;
};
// Docker drops trailing zeroes form timestamps. Pad out timestamps for correct string comparison
private toLogLine(stream: DockerLogStream, raw: string): DockerLogLine | null {
const match = raw.match(DOCKER_LOG_TS_REGEX);
if (!match) return null; // If there's no time stamp, something is wrong with this line, drop it
const ts = `${match[1]}.${(match[2] ?? "").padEnd(9, "0")}Z`;
const text = raw.slice(match[0].length).replace(/\r$/, "");
return { ts, stream, text: this.truncateLog(text) };
}
private parseLogBuffer = (buffer: Buffer): DockerLogLine[] => {
const lines: DockerLogLine[] = [];
const demuxedBuffer = this.demuxLogBuffer(buffer);
for (const byteStream of demuxedBuffer) {
const { stream, chunk } = byteStream;
for (const raw of chunk.split("\n")) {
if (raw.length === 0) continue;
const line = this.toLogLine(stream, raw);
if (line) lines.push(line);
}
}
return lines;
};
private toContainerLogs = async (docker: Dockerode, summary: Dockerode.ContainerInfo): Promise<DockerContainerLogs | null> => {
const containerName = this.toContainerName(summary);
try {
const buffer = await docker
.getContainer(summary.Id)
.logs({ follow: false, stdout: true, stderr: true, timestamps: true, tail: DOCKER_LOG_TAIL_LINES });
return { containerId: summary.Id, containerName, lines: this.parseLogBuffer(buffer) };
} catch (error: unknown) {
this.logger.warn({
message: `Failed to read logs for container ${containerName}`,
service: SERVICE_NAME,
method: "toContainerLogs",
details: { containerId: summary.Id, error: error instanceof Error ? error.message : String(error) },
});
return null;
}
};
private collectLogs = async (docker: Dockerode, summaries: Dockerode.ContainerInfo[]): Promise<DockerContainerLogs[]> => {
const rawLogs = await this.mapWithConcurrency(summaries, STATS_CONCURRENCY, (s) => this.toContainerLogs(docker, s));
return rawLogs.filter((log) => log !== null);
};
private failed = (
monitor: Monitor,
code: number,
message: string,
responseTime: number,
peerResponded = false
): MonitorStatusResponse<DockerStatusPayload> => ({
monitorId: monitor.id,
teamId: monitor.teamId,
type: monitor.type,
status: false,
code,
peerResponded,
message,
responseTime,
payload: null,
});
handle = async (monitor: Monitor, ctx?: CheckContext): Promise<MonitorStatusResponse<DockerStatusPayload>> => {
try {
if (isCaptureDockerUrl(monitor.url)) return await this.handleCapture(monitor, ctx);
const credentials = this.resolveTlsCredentials(monitor, ctx);
if (credentials && !credentials.ok) return this.failed(monitor, NETWORK_ERROR, credentials.message, 0);
const docker = new this.DockerLib(this.toDockerOptions(monitor, credentials));
// Host reachability is the monitor's status; ping latency is the responseTime
const { responseTime, error } = await timeRequest(() => docker.ping());
if (error) {
let message = "Docker host is unreachable";
let code = NETWORK_ERROR;
// A status code from the daemon is the daemon answering; otherwise only a refusal or a reset
// proves we reached the host, and anything else could be the instance's own loss of egress.
let peerResponded = peerAnsweredFromError(error);
if (this.isDockerError(error)) {
code = error.statusCode ?? NETWORK_ERROR;
message = error.json?.message ?? error.reason ?? error.message;
peerResponded = peerResponded || error.statusCode !== undefined;
} else if (error instanceof Error) {
message = error.message;
}
return this.failed(monitor, code, message, responseTime, peerResponded);
}
const summaries = await docker.listContainers({ all: true });
const containers = await this.mapWithConcurrency(summaries, STATS_CONCURRENCY, (s) => this.toContainerInfo(docker, s));
const logs = monitor.dockerLogsEnabled ? await this.collectLogs(docker, summaries) : undefined;
const summary = this.toContainerSummary(containers);
return {
monitorId: monitor.id,
teamId: monitor.teamId,
type: monitor.type,
status: true,
code: 200,
peerResponded: true,
message: "Docker host is reachable",
responseTime,
payload: { containers, summary, logs },
};
} catch (error: unknown) {
if (error instanceof AppError) throw error;
throw new AppError({
message: error instanceof Error ? error.message : "Error performing Docker request",
service: SERVICE_NAME,
method: "handle",
details: {
url: monitor.url,
},
});
}
};
}