feat: add http gateway (#3)

This commit is contained in:
Morten Olsen
2024-01-12 21:10:48 +01:00
committed by GitHub
parent 9c5249956e
commit 1115ce2fb3
22 changed files with 397 additions and 74 deletions

View File

@@ -17,12 +17,12 @@ const bundle = async ({ entry, autoInstall }: BundleOptions) => {
const entryFile = resolve(entry);
const codeBundler = await rollup({
plugins: [
fix(json)(),
fix(sucrase)({
transforms: ['typescript', 'jsx'],
}),
...[autoInstall ? fix(auto) : []],
nodeResolve({ extensions: ['.js', '.jsx', '.ts', '.tsx'] }),
fix(json)(),
nodeResolve({ preferBuiltins: true, extensions: ['.js', '.jsx', '.ts', '.tsx'] }),
fix(commonjs)({ include: /node_modules/ }),
],
input: entryFile,

View File

@@ -25,7 +25,7 @@ push
const code = await step('Bundling', async () => {
return await bundle({ entry: location, autoInstall: opts.autoInstall });
});
const id = await step('Creating load', async () => {
const id = await step(`Creating load ${(code.length / 1024).toFixed(0)}`, async () => {
return await client.loads.set.mutate({
id: opts.id,
name: opts.name,
@@ -34,9 +34,10 @@ push
});
console.log('created load with id', id);
if (opts.run) {
await step('Creating run', async () => {
await client.runs.create.mutate({ loadId: id });
const runId = await step('Creating run', async () => {
return await client.runs.create.mutate({ loadId: id });
});
console.log('created run with id', runId);
}
});

View File

@@ -0,0 +1,59 @@
import { Command } from 'commander';
import { createClient } from '../../client/client.js';
import { step } from '../../utils/step.js';
import { Context } from '../../context/context.js';
import inquirer from 'inquirer';
import { Config } from '../../config/config.js';
const remove = new Command('remove');
const toInt = (value?: string) => {
if (!value) {
return undefined;
}
return parseInt(value, 10);
};
remove
.alias('ls')
.description('List logs')
.option('-l, --load-id <loadId>', 'Load ID')
.option('-o, --offset <offset>', 'Offset')
.option('-a, --limit <limit>', 'Limit', '1000')
.action(async () => {
const { loadId, offset, limit } = remove.opts();
const config = new Config();
const context = new Context(config.context);
const client = await step('Connecting to server', async () => {
return createClient(context);
});
const response = await step('Preparing to delete', async () => {
return await client.runs.prepareRemove.query({
loadId,
offset: toInt(offset),
limit: toInt(limit),
});
});
if (!response.ids.length) {
console.log('No logs to delete');
return;
}
const { confirm } = await inquirer.prompt([
{
type: 'confirm',
name: 'confirm',
message: `Are you sure you want to delete ${response.ids.length} logs?`,
},
]);
if (!confirm) {
return;
}
await step('Deleting artifacts', async () => {
await client.runs.remove.mutate(response);
});
});
export { remove };

View File

@@ -0,0 +1,23 @@
import { Command } from 'commander';
import { createClient } from '../../client/client.js';
import { step } from '../../utils/step.js';
import { Context } from '../../context/context.js';
import { Config } from '../../config/config.js';
const terminate = new Command('terminate');
terminate
.description('Terminate an in progress run')
.argument('run-id', 'Run ID')
.action(async (runId) => {
const config = new Config();
const context = new Context(config.context);
const client = await step('Connecting to server', async () => {
return createClient(context);
});
await step('Terminating run', async () => {
await client.runs.terminate.mutate(runId);
});
});
export { terminate };

View File

@@ -1,8 +1,14 @@
import { Command } from 'commander';
import { create } from './runs.create.js';
import { list } from './runs.list.js';
import { remove } from './runs.remove.js';
import { terminate } from './runs.terminate.js';
const runs = new Command('runs');
runs.description('Manage runs').addCommand(create).addCommand(list);
runs.description('Manage runs');
runs.addCommand(create);
runs.addCommand(list);
runs.addCommand(remove);
runs.addCommand(terminate);
export { runs };

View File

@@ -18,9 +18,9 @@
}
},
"devDependencies": {
"@morten-olsen/mini-loader-configs": "workspace:^",
"@morten-olsen/mini-loader-cli": "workspace:^",
"@morten-olsen/mini-loader": "workspace:^",
"@morten-olsen/mini-loader-cli": "workspace:^",
"@morten-olsen/mini-loader-configs": "workspace:^",
"@types/node": "^20.10.8",
"typescript": "^5.3.3"
},
@@ -28,5 +28,8 @@
"repository": {
"type": "git",
"url": "https://github.com/morten-olsen/mini-loader"
},
"dependencies": {
"fastify": "^4.25.2"
}
}

View File

@@ -0,0 +1,12 @@
import { http } from '@morten-olsen/mini-loader';
import fastify from 'fastify';
const server = fastify();
server.all('*', async (req) => {
return req.url;
});
server.listen({
path: http.getPath(),
});

View File

@@ -3,6 +3,7 @@ import { artifacts, logger } from '@morten-olsen/mini-loader';
const run = async () => {
await logger.info('Hello world');
await artifacts.create('foo', 'bar');
process.exit(0);
};
run();

View File

@@ -0,0 +1,7 @@
const getPath = () => process.env.HTTP_GATEWAY_PATH!;
const http = {
getPath,
};
export { http };

View File

@@ -8,3 +8,4 @@ export { logger } from './logger/logger.js';
export { artifacts } from './artifacts/artifacts.js';
export { input } from './input/input.js';
export { secrets } from './secrets/secrets.js';
export { http } from './http/http.js';

View File

@@ -17,12 +17,12 @@
}
},
"devDependencies": {
"@morten-olsen/mini-loader": "workspace:^",
"@morten-olsen/mini-loader-configs": "workspace:^",
"@types/node": "^20.10.8",
"typescript": "^5.3.3"
},
"dependencies": {
"@morten-olsen/mini-loader": "workspace:^",
"eventemitter3": "^5.0.1",
"nanoid": "^5.0.4"
},

View File

@@ -1,17 +1,5 @@
import { Worker } from 'worker_threads';
import os from 'os';
import { EventEmitter } from 'eventemitter3';
import { Event } from '@morten-olsen/mini-loader';
import { join } from 'path';
import { createServer } from 'http';
import { nanoid } from 'nanoid';
import { chmod, mkdir, rm, writeFile } from 'fs/promises';
type RunEvents = {
message: (event: Event) => void;
error: (error: Error) => void;
exit: () => void;
};
import { setup } from './setup/setup.js';
type RunOptions = {
script: string;
@@ -20,44 +8,17 @@ type RunOptions = {
};
const run = async ({ script, input, secrets }: RunOptions) => {
const dataDir = join(os.tmpdir(), 'mini-loader', nanoid());
await mkdir(dataDir, { recursive: true });
await chmod(dataDir, 0o700);
const hostSocket = join(dataDir, 'host');
const server = createServer();
const inputLocation = join(dataDir, 'input');
const info = await setup({ script, input, secrets });
if (input) {
await writeFile(inputLocation, input);
}
const emitter = new EventEmitter<RunEvents>();
server.on('connection', (socket) => {
socket.on('data', (data) => {
const message = JSON.parse(data.toString());
emitter.emit('message', message);
});
});
server.listen(hostSocket);
const worker = new Worker(script, {
eval: true,
const worker = new Worker(info.scriptLocation, {
stdin: false,
stdout: false,
stderr: false,
env: {
HOST_SOCKET: hostSocket,
SECRETS: JSON.stringify(secrets),
INPUT_PATH: inputLocation,
},
workerData: {
input,
},
env: info.env,
});
worker.stdout?.on('data', (data) => {
emitter.emit('message', {
info.emitter.emit('message', {
type: 'log',
payload: {
severity: 'info',
@@ -67,7 +28,7 @@ const run = async ({ script, input, secrets }: RunOptions) => {
});
worker.stderr?.on('data', (data) => {
emitter.emit('message', {
info.emitter.emit('message', {
type: 'log',
payload: {
severity: 'error',
@@ -78,20 +39,24 @@ const run = async ({ script, input, secrets }: RunOptions) => {
const promise = new Promise<void>((resolve, reject) => {
worker.on('exit', async () => {
server.close();
await rm(dataDir, { recursive: true, force: true });
await info.teardown();
resolve();
});
worker.on('error', async (error) => {
server.close();
reject(error);
});
});
return {
emitter,
...info,
teardown: async () => {
worker.terminate();
},
promise,
};
};
type RunInfo = Awaited<ReturnType<typeof run>>;
export type { RunInfo };
export { run };

View File

@@ -0,0 +1,71 @@
import { join } from 'path';
import os from 'os';
import { nanoid } from 'nanoid';
import { chmod, mkdir, rm, writeFile } from 'fs/promises';
import { createServer } from 'net';
import { EventEmitter } from 'eventemitter3';
type SetupOptions = {
input?: Buffer | string;
script: string;
secrets?: Record<string, string>;
};
type RunEvents = {
message: (event: any) => void;
error: (error: Error) => void;
exit: () => void;
};
const setup = async (options: SetupOptions) => {
const { input, script, secrets } = options;
const emitter = new EventEmitter<RunEvents>();
const dataDir = join(os.tmpdir(), 'mini-loader', nanoid());
await mkdir(dataDir, { recursive: true });
await chmod(dataDir, 0o700);
const hostSocket = join(dataDir, 'host');
const httpGatewaySocket = join(dataDir, 'socket');
const server = createServer();
const inputLocation = join(dataDir, 'input');
const scriptLocation = join(dataDir, 'script.js');
if (input) {
await writeFile(inputLocation, input);
}
await writeFile(scriptLocation, script);
const env = {
HOST_SOCKET: hostSocket,
SECRETS: JSON.stringify(secrets || {}),
INPUT_PATH: inputLocation,
HTTP_GATEWAY_PATH: httpGatewaySocket,
};
const teardown = async () => {
server.close();
await rm(dataDir, { recursive: true, force: true });
};
server.on('connection', (socket) => {
socket.on('data', (data) => {
const message = JSON.parse(data.toString());
emitter.emit('message', message);
});
});
server.listen(hostSocket);
return {
env,
emitter,
teardown,
httpGatewaySocket,
scriptLocation,
hostSocket,
};
};
type Setup = Awaited<ReturnType<typeof setup>>;
export type { Setup };
export { setup };

View File

@@ -27,6 +27,7 @@
"typescript": "^5.3.3"
},
"dependencies": {
"@fastify/reply-from": "^9.7.0",
"@trpc/client": "^10.45.0",
"@trpc/server": "^10.45.0",
"commander": "^11.1.0",

View File

@@ -0,0 +1,34 @@
import { FastifyPluginAsync } from 'fastify';
import FastifyReplyFrom from '@fastify/reply-from';
import { escape } from 'querystring';
import { Runtime } from '../runtime/runtime.js';
type Options = {
runtime: Runtime;
};
const gateway: FastifyPluginAsync<Options> = async (fastify, { runtime }) => {
await fastify.register(FastifyReplyFrom, {
http: {},
});
fastify.all('/gateway/*', (req, res) => {
const [runId, ...pathSegments] = (req.params as any)['*'].split('/').filter(Boolean);
const run = runtime.runner.getInstance(runId);
if (!run) {
res.statusCode = 404;
res.send({ error: 'Run not found' });
return;
}
const socketPath = run.run?.httpGatewaySocket;
if (!socketPath) {
res.statusCode = 404;
res.send({ error: 'No socket path to run' });
return;
}
const path = pathSegments.join('/');
res.from(`unix+http://${escape(socketPath)}/${path}`);
});
};
export { gateway };

View File

@@ -3,6 +3,7 @@ import { EventEmitter } from 'eventemitter3';
import { Database } from '../../database/database.js';
import { CreateRunOptions, FindRunsOptions, UpdateRunOptions } from './runs.schemas.js';
import { LoadRepo } from '../loads/loads.js';
import { createHash } from 'crypto';
type RunRepoEvents = {
created: (args: { id: string; loadId: string }) => void;
@@ -18,13 +19,22 @@ type RunRepoOptions = {
class RunRepo extends EventEmitter<RunRepoEvents> {
#options: RunRepoOptions;
#isReady: Promise<void>;
constructor(options: RunRepoOptions) {
super();
this.#options = options;
this.#isReady = this.#setup();
}
#setup = async () => {
const { database } = this.#options;
const db = await database.instance;
await db('runs').update({ status: 'failed', error: 'server was shut down' }).where({ status: 'running' });
};
public getById = async (id: string) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
@@ -36,6 +46,7 @@ class RunRepo extends EventEmitter<RunRepoEvents> {
};
public getByLoadId = async (loadId: string) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
@@ -44,6 +55,7 @@ class RunRepo extends EventEmitter<RunRepoEvents> {
};
public find = async (options: FindRunsOptions) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
const query = db('runs').select(['id', 'status', 'startedAt', 'status', 'error', 'endedAt']);
@@ -62,19 +74,41 @@ class RunRepo extends EventEmitter<RunRepoEvents> {
return runs;
};
public remove = async (options: FindRunsOptions) => {
public prepareRemove = async (options: FindRunsOptions) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
const query = db('runs');
const query = db('runs').select('id');
if (options.loadId) {
query.where({ loadId: options.loadId });
}
await query.del();
const result = await query;
const ids = result.map((row) => row.id);
const token = ids.map((id) => Buffer.from(id).toString('base64')).join('|');
const hash = createHash('sha256').update(token).digest('hex');
return {
ids,
hash,
};
};
public remove = async (hash: string, ids: string[]) => {
const { database } = this.#options;
const db = await database.instance;
const token = ids.map((id) => Buffer.from(id).toString('base64')).join('|');
const actualHash = createHash('sha256').update(token).digest('hex');
if (hash !== actualHash) {
throw new Error('Invalid hash');
}
await db('runs').whereIn('id', ids).delete();
};
public started = async (id: string) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
const current = await this.getById(id);
@@ -92,6 +126,7 @@ class RunRepo extends EventEmitter<RunRepoEvents> {
};
public finished = async (id: string, options: UpdateRunOptions) => {
await this.#isReady;
const { database } = this.#options;
const db = await database.instance;
const { loadId } = await this.getById(id);
@@ -114,6 +149,7 @@ class RunRepo extends EventEmitter<RunRepoEvents> {
};
public create = async (options: CreateRunOptions) => {
await this.#isReady;
const { database, loads } = this.#options;
const id = nanoid();
const db = await database.instance;

View File

@@ -1,3 +1,4 @@
import { z } from 'zod';
import { createRunSchema, findRunsSchema } from '../repos/repos.js';
import { publicProcedure, router } from './router.utils.js';
@@ -17,17 +18,50 @@ const find = publicProcedure.input(findRunsSchema).query(async ({ input, ctx })
return results;
});
const remove = publicProcedure.input(findRunsSchema).mutation(async ({ input, ctx }) => {
const prepareRemove = publicProcedure.input(findRunsSchema).query(async ({ input, ctx }) => {
const { runtime } = ctx;
const { repos } = runtime;
const { runs } = repos;
await runs.remove(input);
return await runs.prepareRemove(input);
});
const remove = publicProcedure
.input(
z.object({
hash: z.string(),
ids: z.array(z.string()),
}),
)
.mutation(async ({ input, ctx }) => {
const { runtime } = ctx;
const { repos } = runtime;
const { runs } = repos;
for (const id of input.ids) {
const instance = runtime.runner.getInstance(id);
if (instance) {
await instance.run?.teardown();
}
}
await runs.remove(input.hash, input.ids);
});
const terminate = publicProcedure.input(z.string()).mutation(async ({ input, ctx }) => {
const { runtime } = ctx;
const { runner } = runtime;
const instance = runner.getInstance(input);
if (!instance || !instance.run) {
return;
}
await instance.run.teardown();
});
const runsRouter = router({
create,
find,
remove,
prepareRemove,
terminate,
});
export { runsRouter };

View File

@@ -1,5 +1,5 @@
import { EventEmitter } from 'eventemitter3';
import { run } from '@morten-olsen/mini-loader-runner';
import { RunInfo, run } from '@morten-olsen/mini-loader-runner';
import { Repos } from '../repos/repos.js';
import { LoggerEvent } from '../../../mini-loader/dist/esm/logger/logger.js';
import { ArtifactCreateEvent } from '../../../mini-loader/dist/esm/artifacts/artifacts.js';
@@ -20,12 +20,17 @@ type RunnerInstanceOptions = {
class RunnerInstance extends EventEmitter<RunnerInstanceEvents> {
#options: RunnerInstanceOptions;
#run?: RunInfo;
constructor(options: RunnerInstanceOptions) {
super();
this.#options = options;
}
public get run() {
return this.#run;
}
#addLog = async (event: LoggerEvent['payload']) => {
const { repos, id, loadId } = this.#options;
const { logs } = repos;
@@ -58,11 +63,13 @@ class RunnerInstance extends EventEmitter<RunnerInstanceEvents> {
const script = await readFile(scriptLocation, 'utf-8');
const allSecrets = await secrets.getAll();
await runs.started(id);
const { promise, emitter } = await run({
const current = await run({
script,
secrets: allSecrets,
input,
});
this.#run = current;
const { promise, emitter } = current;
emitter.on('message', (message) => {
switch (message.type) {
case 'log': {
@@ -84,9 +91,11 @@ class RunnerInstance extends EventEmitter<RunnerInstanceEvents> {
}
await runs.finished(id, { status: 'failed', error: errorMessage });
} finally {
this.#run = undefined;
this.emit('completed', { id });
}
};
}
export type { RunInfo };
export { RunnerInstance };

View File

@@ -36,6 +36,10 @@ class Runner {
this.#instances.set(args.id, instance);
await instance.start();
};
public getInstance = (id: string) => {
return this.#instances.get(id);
};
}
export { Runner };

View File

@@ -3,9 +3,16 @@ import fastify from 'fastify';
import { RootRouter, rootRouter } from '../router/router.js';
import { createContext } from '../router/router.utils.js';
import { Runtime } from '../runtime/runtime.js';
import { gateway } from '../gateway/gateway.js';
const createServer = async (runtime: Runtime) => {
const server = fastify({});
const server = fastify({
maxParamLength: 10000,
bodyLimit: 30 * 1024 * 1024,
logger: {
level: 'warn',
},
});
server.get('/', async () => {
return { hello: 'world' };
});
@@ -33,6 +40,14 @@ const createServer = async (runtime: Runtime) => {
},
} satisfies FastifyTRPCPluginOptions<RootRouter>['trpcOptions'],
});
server.register(gateway, {
runtime,
});
server.addHook('onError', async (request, reply, error) => {
console.error(error);
});
await server.ready();
return server;

47
pnpm-lock.yaml generated
View File

@@ -101,6 +101,10 @@ importers:
packages/configs: {}
packages/examples:
dependencies:
fastify:
specifier: ^4.25.2
version: 4.25.2
devDependencies:
'@morten-olsen/mini-loader':
specifier: workspace:^
@@ -132,6 +136,9 @@ importers:
packages/runner:
dependencies:
'@morten-olsen/mini-loader':
specifier: workspace:^
version: link:../mini-loader
eventemitter3:
specifier: ^5.0.1
version: 5.0.1
@@ -139,9 +146,6 @@ importers:
specifier: ^5.0.4
version: 5.0.4
devDependencies:
'@morten-olsen/mini-loader':
specifier: workspace:^
version: link:../mini-loader
'@morten-olsen/mini-loader-configs':
specifier: workspace:^
version: link:../configs
@@ -154,6 +158,9 @@ importers:
packages/server:
dependencies:
'@fastify/reply-from':
specifier: ^9.7.0
version: 9.7.0
'@trpc/client':
specifier: ^10.45.0
version: 10.45.0(@trpc/server@10.45.0)
@@ -476,6 +483,11 @@ packages:
fast-uri: 2.3.0
dev: false
/@fastify/busboy@2.1.0:
resolution: {integrity: sha512-+KpH+QxZU7O4675t3mnkQKcZZg56u+K/Ct2K+N2AZYNVK8kyeo/bI18tI8aPm3tvNNRyTWfj6s5tnGNlcbQRsA==}
engines: {node: '>=14'}
dev: false
/@fastify/deepmerge@1.3.0:
resolution: {integrity: sha512-J8TOSBq3SoZbDhM9+R/u77hP93gz/rajSA+K2kGyijPpORPWUXHUpTaleoj+92As0S9uPRP7Oi8IqMf0u+ro6A==}
dev: false
@@ -490,6 +502,19 @@ packages:
fast-json-stringify: 5.10.0
dev: false
/@fastify/reply-from@9.7.0:
resolution: {integrity: sha512-/F1QBl3FGlTqStjmiuoLRDchVxP967TZh6FZPwQteWhdLsDec8mqSACE+cRzw6qHUj3v9hfdd7JNgmb++fyFhQ==}
dependencies:
'@fastify/error': 3.4.1
end-of-stream: 1.4.4
fast-content-type-parse: 1.1.0
fast-querystring: 1.1.2
fastify-plugin: 4.5.1
pump: 3.0.0
tiny-lru: 11.2.5
undici: 5.28.2
dev: false
/@gar/promisify@1.1.3:
resolution: {integrity: sha512-k2Ty1JcVojjJFwrg/ThKi2ujJ7XNLYaFGNB/bWT9wGR+oSMJHMa5w+CUq6p/pVrKeNNgA7pCqEcjSnHVoqJQFw==}
requiresBuild: true
@@ -2683,6 +2708,10 @@ packages:
resolution: {integrity: sha512-eel5UKGn369gGEWOqBShmFJWfq/xSJvsgDzgLYC845GneayWvXBf0lJCBn5qTABfewy1ZDPoaR5OZCP+kssfuw==}
dev: false
/fastify-plugin@4.5.1:
resolution: {integrity: sha512-stRHYGeuqpEZTL1Ef0Ovr2ltazUT9g844X5z/zEBFLG8RYlpDiOCIG+ATvYEp+/zmc7sN29mcIMp8gvYplYPIQ==}
dev: false
/fastify@4.25.2:
resolution: {integrity: sha512-SywRouGleDHvRh054onj+lEZnbC1sBCLkR0UY3oyJwjD4BdZJUrxBqfkfCaqn74pVCwBaRHGuL3nEWeHbHzAfw==}
dependencies:
@@ -5151,6 +5180,11 @@ packages:
engines: {node: '>=8'}
dev: false
/tiny-lru@11.2.5:
resolution: {integrity: sha512-JpqM0K33lG6iQGKiigcwuURAKZlq6rHXfrgeL4/I8/REoyJTGU+tEMszvT/oTRVHG2OiylhGDjqPp1jWMlr3bw==}
engines: {node: '>=12'}
dev: false
/tmp@0.0.33:
resolution: {integrity: sha512-jRCJlojKnZ3addtTOjdIqoRuPEKBvNXcGYqzO6zWZX8KfKEpnGY5jfggJQ3EjKuu8D4bJRr0y+cYJFmYbImXGw==}
engines: {node: '>=0.6.0'}
@@ -5368,6 +5402,13 @@ packages:
/undici-types@5.26.5:
resolution: {integrity: sha512-JlCMO+ehdEIKqlFxk6IfVoAUVmgz7cU7zD/h9XZ0qzeosSHmUJVOzSQvvYSYWXkFXC+IfLKSIffhv0sVZup6pA==}
/undici@5.28.2:
resolution: {integrity: sha512-wh1pHJHnUeQV5Xa8/kyQhO7WFa8M34l026L5P/+2TYiakvGy5Rdc8jWZVyG7ieht/0WgJLEd3kcU5gKx+6GC8w==}
engines: {node: '>=14.0'}
dependencies:
'@fastify/busboy': 2.1.0
dev: false
/unique-filename@1.1.1:
resolution: {integrity: sha512-Vmp0jIp2ln35UTXuryvjzkjGdRyf9b2lTXuSYUiPmzRcl3FDtYqAwOnTJkAngD9SWhnoJzDbTKwaOrZ+STtxNQ==}
requiresBuild: true

View File

@@ -1,15 +1,15 @@
{
"include": [],
"references": [
{
"path": "./packages/mini-loader/tsconfig.json"
},
{
"path": "./packages/runner/tsconfig.json"
},
{
"path": "./packages/server/tsconfig.json"
},
{
"path": "./packages/mini-loader/tsconfig.json"
},
{
"path": "./packages/cli/tsconfig.json"
},