Skip to content
Merged
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
46 changes: 45 additions & 1 deletion app/src/server/backups.ts
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,45 @@ export interface BackupRequest {
at?: string;
}

export interface Schedule {
enabled: boolean;
everyHours: number;
hour: number;
keepDays: number;
}

export const defaultSchedule: Schedule = { enabled: true, everyHours: 24, hour: 3, keepDays: 14 };
export const periods = [1, 2, 3, 4, 6, 8, 12, 24, 48, 72, 168];

export function nextRun(schedule: Schedule, now: number) {
const period = schedule.everyHours * 3600_000;
const anchor = schedule.hour * 3600_000;
return anchor + Math.ceil((now - anchor) / period) * period;
}

export async function readSchedule(root = backupRoot()): Promise<Schedule> {
return { ...defaultSchedule, ...(await json<Partial<Schedule>>(path.join(root, '.schedule.json')).catch(missing)) };
}

export async function writeSchedule(input: unknown, login: string, root = backupRoot()) {
const { enabled, everyHours, hour, keepDays } = (input ?? {}) as Record<string, unknown>;
const whole = (value: unknown, least: number, most: number) => Number.isInteger(value) && (value as number) >= least && (value as number) <= most;
if (typeof enabled !== 'boolean' || !periods.includes(everyHours as number) || !whole(hour, 0, 23) || !whole(keepDays, 1, 90)) {
throw new Refusal(400, `A schedule is on or off, every ${periods.join(', ')} hours, at an hour from 0 to 23, kept 1 to 90 days.`);
}
await mkdir(root, { recursive: true });
await writeFile(path.join(root, '.schedule.json.partial'), JSON.stringify({ enabled, everyHours, hour, keepDays }));
await rename(path.join(root, '.schedule.json.partial'), path.join(root, '.schedule.json'));
await audit(login, 'schedule', '*', `${enabled ? `every ${everyHours} h from ${hour}:00 UTC` : 'off'}, ${keepDays} days`, root);
}

function missing(error: NodeJS.ErrnoException) {
if (error.code === 'ENOENT') {
return undefined;
}
throw error;
}

export const projectPattern = /^[a-z0-9][a-z0-9_-]*$/;
export const stampPattern = /^\d{8}T\d{6}Z$/;
const hour = 3600_000;
Expand Down Expand Up @@ -79,7 +118,12 @@ export function stampTime(stamp: string) {
return Date.parse(`${stamp.slice(0, 4)}-${stamp.slice(4, 6)}-${stamp.slice(6, 8)}T${stamp.slice(9, 11)}:${stamp.slice(11, 13)}:${stamp.slice(13, 15)}Z`);
}

export async function backupConditions(now: number, root = backupRoot(), hours = 26) {
export async function backupConditions(now: number, root = backupRoot(), least = 26) {
const schedule = await readSchedule(root);
if (!schedule.enabled) {
return [];
}
const hours = Math.max(least, schedule.everyHours + 2);
return (await listBackups(root)).flatMap(({ project, archives }) => {
const latest = archives.find((archive) => !archive.imported);
if (!latest) {
Expand Down
2 changes: 2 additions & 0 deletions app/src/server/config.ts
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@ export const config = {
dockerSocket: process.env.DOCKER_SOCKET ?? '/var/run/docker.sock',
hostRoot: process.env.HOST_ROOT ?? '/proc/1/root',
dataDir: process.env.DATA_DIR ?? '/data',
logsDir: process.env.LOGS_DIR ?? '/logs',
logsKeepDays: Number(process.env.LOGS_KEEP_DAYS ?? 7),
sablierUrl: process.env.SABLIER_URL ?? 'http://sablier:10000',
self: process.env.HOSTNAME ?? '',
sshTarget: process.env.SSH_TARGET ?? 'ubuntu@localhost',
Expand Down
224 changes: 224 additions & 0 deletions app/src/server/logstore.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,224 @@
import { mkdirSync } from 'node:fs';
import type http from 'node:http';
import path from 'node:path';
import { DatabaseSync, type SQLInputValue } from 'node:sqlite';
import { Demuxer, docker, LogLines, stream } from './docker.ts';
import { resumeAfter } from './terminal.ts';

export const levels = ['error', 'warning', 'info', 'debug'] as const;
export type Level = (typeof levels)[number];
const day = 86_400_000;
const longest = 8192;

export function levelOf(text: string): Level {
const lower = text.toLowerCase();
if (/\b(error|err|failed|failure|exception|fatal|panic|critical)\b/.test(lower)) return 'error';
if (/\b(warn|warning|wrn|caution)\b/.test(lower)) return 'warning';
if (/\b(debug|dbg|trace|verbose)\b/.test(lower)) return 'debug';
return 'info';
}

export interface StoredLine {
id: number;
at: number;
time: string;
project: string;
service: string;
container: string;
stream: 'stdout' | 'stderr';
level: Level;
text: string;
}

export interface LogQuery {
projects?: string[];
containers?: string[];
streams?: string[];
levels?: string[];
text?: string;
from?: number;
to?: number;
before?: number;
after?: number;
limit?: number;
}

interface Source {
id: string;
name: string;
project: string;
service: string;
}

export class LogStore {
private readonly db: DatabaseSync;
private readonly following = new Map<string, http.IncomingMessage | null>();
private pending: (Omit<StoredLine, 'id' | 'level'> & { level: Level })[] = [];

constructor(
directory: string,
private readonly keepDays = 7,
) {
mkdirSync(directory, { recursive: true });
this.db = new DatabaseSync(path.join(directory, 'logs.db'));
this.db.exec(`
PRAGMA journal_mode = WAL;
CREATE TABLE IF NOT EXISTS lines (
id INTEGER PRIMARY KEY, at INTEGER NOT NULL, time TEXT NOT NULL, project TEXT NOT NULL, service TEXT NOT NULL,
container TEXT NOT NULL, stream TEXT NOT NULL, level TEXT NOT NULL, text TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS lines_at ON lines (at);
CREATE INDEX IF NOT EXISTS lines_container_at ON lines (container, at);
CREATE TABLE IF NOT EXISTS cursors (container TEXT PRIMARY KEY, time TEXT NOT NULL);
`);
}

add(line: Omit<StoredLine, 'id' | 'level' | 'at'>) {
this.pending.push({ ...line, at: Date.parse(line.time) || Date.now(), level: levelOf(line.text), text: line.text.slice(0, longest) });
}

flush() {
if (!this.pending.length) {
return;
}
const lines = this.pending;
this.pending = [];
const insert = this.db.prepare('INSERT INTO lines (at, time, project, service, container, stream, level, text) VALUES (?, ?, ?, ?, ?, ?, ?, ?)');
const cursor = this.db.prepare('INSERT INTO cursors (container, time) VALUES (?, ?) ON CONFLICT (container) DO UPDATE SET time = excluded.time');
this.db.exec('BEGIN');
try {
for (const line of lines) {
insert.run(line.at, line.time, line.project, line.service, line.container, line.stream, line.level, line.text);
cursor.run(line.container, line.time);
}
this.db.exec('COMMIT');
} catch (error) {
this.db.exec('ROLLBACK');
throw error;
}
}

prune(now = Date.now()) {
this.db.prepare('DELETE FROM lines WHERE at < ?').run(now - this.keepDays * day);
}

cursor(container: string) {
return (this.db.prepare('SELECT time FROM cursors WHERE container = ?').get(container) as { time: string } | undefined)?.time;
}

query(filter: LogQuery) {
const clauses: string[] = [];
const values: SQLInputValue[] = [];
const within = (column: string, list?: string[]) => {
if (list?.length) {
clauses.push(`${column} IN (${list.map(() => '?').join(', ')})`);
values.push(...list);
}
};
within('project', filter.projects);
within('container', filter.containers);
within('stream', filter.streams);
within('level', filter.levels);
for (const [clause, value] of [
['at >= ?', filter.from],
['at <= ?', filter.to],
['id < ?', filter.before],
['id > ?', filter.after],
] as const) {
if (value !== undefined) {
clauses.push(clause);
values.push(value);
}
}
if (filter.text) {
clauses.push("text LIKE ? ESCAPE '\\'");
values.push(`%${filter.text.replace(/[\\%_]/g, '\\$&')}%`);
}
const limit = Math.min(Math.max(filter.limit ?? 200, 1), 1000);
const order = filter.after !== undefined ? 'ASC' : 'DESC';
const rows = this.db
.prepare(`SELECT * FROM lines ${clauses.length ? `WHERE ${clauses.join(' AND ')}` : ''} ORDER BY id ${order} LIMIT ?`)
.all(...values, limit + 1) as unknown as StoredLine[];
const lines = rows.slice(0, limit);
return { lines: order === 'ASC' ? lines.reverse() : lines, more: rows.length > limit };
}

stats() {
const totals = this.db.prepare("SELECT count(*) AS lines, min(at) AS oldest, sum(level = 'error') AS errors FROM lines").get() as {
lines: number;
oldest: number | null;
errors: number | null;
};
const { size } = this.db.prepare('SELECT page_count * page_size AS size FROM pragma_page_count(), pragma_page_size()').get() as { size: number };
return { lines: totals.lines, oldest: totals.oldest, errors: totals.errors ?? 0, bytes: size };
}

sources(now = Date.now()) {
return this.db
.prepare('SELECT project, service, container, max(at) AS last FROM lines WHERE at >= ? GROUP BY project, service, container ORDER BY project, service')
.all(now - this.keepDays * day) as { project: string; service: string; container: string; last: number }[];
}

private async follow(source: Source) {
const since = this.cursor(source.name);
const query = `since=${since ? resumeAfter(since) : Math.floor((Date.now() - this.keepDays * day) / 1000)}`;
const { Config } = await docker<{ Config: { Tty: boolean } }>('GET', `/containers/${source.id}/json`);
const output = await stream('GET', `/containers/${source.id}/logs?stdout=1&stderr=1&timestamps=1&follow=1&${query}`);
if (!this.following.has(source.id)) {
output.destroy();
return;
}
this.following.set(source.id, output);
const demuxer = Config.Tty ? undefined : new Demuxer();
const lines = new LogLines();
output.on('data', (chunk: Buffer) => {
const parsed = demuxer
? demuxer.push(chunk).flatMap((frame) => lines.push(frame.stream === 2 ? 'stderr' : 'stdout', frame.data))
: lines.push('stdout', chunk);
for (const line of parsed) {
this.add({ time: line.time, project: source.project, service: source.service, container: source.name, stream: line.stream, text: line.text });
}
});
output.on('close', () => this.following.delete(source.id));
}

track(views: { id: string; name: string; project: string; service: string; state: string }[]) {
const running = new Set(views.filter((view) => view.state === 'running').map((view) => view.id));
for (const [id, output] of this.following) {
if (!running.has(id)) {
output?.destroy();
this.following.delete(id);
}
}
for (const view of views.filter((candidate) => running.has(candidate.id) && !this.following.has(candidate.id))) {
this.following.set(view.id, null);
this.follow(view).catch((error: unknown) => {
this.following.delete(view.id);
console.error(`following the logs of ${view.name} failed:`, error);
});
}
}

start(list: () => Promise<{ id: string; name: string; project: string; service: string; state: string }[]>) {
const report = (task: string) => (error: unknown) => console.error(`${task} failed:`, error);
const timers = [
setInterval(() => void list().then((views) => this.track(views), report('log sources')), 5000),
setInterval(() => {
try {
this.flush();
} catch (error) {
report('storing logs')(error);
}
}, 1000),
setInterval(() => this.prune(), 3600_000),
];
void list().then((views) => this.track(views), report('log sources'));
this.prune();
return () => timers.forEach(clearInterval);
}

close() {
this.flush();
this.db.close();
}
}
Loading
Loading