Compare commits

...

10 Commits

16 changed files with 7926 additions and 6290 deletions

3
.gitignore vendored
View File

@ -33,4 +33,5 @@ lerna-debug.log*
!.vscode/launch.json
!.vscode/extensions.json
/config.yml
/config.yml
tsconfig.build.tsbuildinfo

13819
package-lock.json generated

File diff suppressed because it is too large Load Diff

View File

@ -38,7 +38,7 @@
"class-validator": "^0.13.1",
"debug": "^4.3.1",
"graphql": "^15.5.0",
"graphql-tools": "^7.0.2",
"graphql-tools": "^8.1.0",
"ioredis": "^4.25.0",
"jose": "^3.14.0",
"js-yaml": "^4.0.0",
@ -48,6 +48,7 @@
"observable-to-async-generator": "^1.0.1-rc",
"pg": "^8.5.1",
"pino-pretty": "^4.7.1",
"pm2": "^5.1.0",
"ramda": "^0.27.1",
"reflect-metadata": "^0.1.13",
"rimraf": "^3.0.2",
@ -56,7 +57,7 @@
"typeorm": "^0.2.30"
},
"devDependencies": {
"@nestjs/cli": "^7.5.7",
"@nestjs/cli": "^7.6.0",
"@nestjs/schematics": "^7.1.3",
"@nestjs/testing": "^7.5.1",
"@types/body-parser": "^1.19.0",

View File

@ -42,9 +42,9 @@ describe('PipelineTaskRunner', () => {
setTimeout(() => {
logger.handleEvent(event);
});
expect(message$.pipe(take(1), timeout(100)).toPromise()).rejects.toMatch(
'timeout',
);
await expect(
message$.pipe(take(1), timeout(100)).toPromise(),
).rejects.toThrow(/timeout/i);
});
it('multiple subscribers', async () => {
const event = new PipelineTaskEvent();
@ -53,13 +53,16 @@ describe('PipelineTaskRunner', () => {
const message2$ = logger.getMessage$('test');
setTimeout(() => {
logger.handleEvent(event);
logger.handleEvent(event);
});
expect(message$.pipe(take(1), timeout(100)).toPromise()).resolves.toEqual(
event,
);
expect(
message2$.pipe(take(1), timeout(100)).toPromise(),
).resolves.toEqual(event);
await Promise.all([
expect(
message$.pipe(take(1), timeout(100)).toPromise(),
).resolves.toEqual(event),
expect(
message2$.pipe(take(1), timeout(100)).toPromise(),
).resolves.toEqual(event),
]);
});
});

View File

@ -13,7 +13,8 @@ import {
@Injectable()
export class PipelineTaskLogger implements OnModuleDestroy {
private readonly messageSubject = new Subject<PipelineTaskEvent>();
private readonly message$: Observable<PipelineTaskEvent> = this.messageSubject.pipe();
private readonly message$: Observable<PipelineTaskEvent> =
this.messageSubject.pipe();
@RabbitSubscribe({
exchange: EXCHANGE_PIPELINE_TASK_FANOUT,

View File

@ -1,3 +1,4 @@
import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
import { Test, TestingModule } from '@nestjs/testing';
import { ReposService } from '../repos/repos.service';
import { PipelineUnits } from './enums/pipeline-units.enum';
@ -11,7 +12,7 @@ import { WorkUnitMetadata } from './models/work-unit-metadata.model';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('PipelineTaskRunner', () => {
let runner: PipelineTaskRunner;
let reposService: ReposService;
let deployByPM2Service: DeployByPm2Service;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
@ -36,11 +37,18 @@ describe('PipelineTaskRunner', () => {
provide: AmqpConnection,
useValue: {},
},
{
provide: DeployByPm2Service,
useValue: {
deploy: () => Promise.resolve(),
},
},
],
}).compile();
reposService = module.get(ReposService);
module.get(ReposService);
runner = module.get(PipelineTaskRunner);
deployByPM2Service = module.get(DeployByPm2Service);
});
it('should be defined', () => {
@ -65,7 +73,7 @@ describe('PipelineTaskRunner', () => {
beforeEach(() => {
emitEvent = jest
.spyOn(runner, 'emitEvent')
.mockImplementation((..._) => Promise.resolve());
.mockImplementation(() => Promise.resolve());
});
describe('doTask', () => {
@ -75,10 +83,10 @@ describe('PipelineTaskRunner', () => {
beforeEach(() => {
checkout = jest
.spyOn(runner, 'checkout')
.mockImplementation((..._) => Promise.resolve('/null'));
.mockImplementation(() => Promise.resolve('/null'));
doTaskUnit = jest
.spyOn(runner, 'doTaskUnit')
.mockImplementation((..._) => Promise.resolve());
.mockImplementation(() => Promise.resolve());
});
it('only checkout', async () => {
@ -171,9 +179,7 @@ describe('PipelineTaskRunner', () => {
doTaskUnit = jest
.spyOn(runner, 'doTaskUnit')
.mockImplementation((..._) =>
Promise.reject(new Error('test error')),
);
.mockImplementation(() => Promise.reject(new Error('test error')));
await runner.doTask(task);
expect(checkout).toBeCalledTimes(1);
@ -189,7 +195,7 @@ describe('PipelineTaskRunner', () => {
it('success', async () => {
const runScript = jest
.spyOn(runner, 'runScript')
.mockImplementation((..._) => Promise.resolve());
.mockImplementation(() => Promise.resolve());
const task = new PipelineTask();
const unit = PipelineUnits.test;
@ -210,9 +216,7 @@ describe('PipelineTaskRunner', () => {
it('failed', async () => {
const runScript = jest
.spyOn(runner, 'runScript')
.mockImplementation((..._) =>
Promise.reject(new Error('test error')),
);
.mockImplementation(() => Promise.reject(new Error('test error')));
const task = new PipelineTask();
const unit = PipelineUnits.test;
@ -230,7 +234,7 @@ describe('PipelineTaskRunner', () => {
describe('runScript', () => {
it('normal', async () => {
const spawn = jest.fn((..._: any[]) => ({
const spawn = jest.fn<any, any>(() => ({
stdout: {
on: () => undefined,
},
@ -256,7 +260,7 @@ describe('PipelineTaskRunner', () => {
});
});
it('failed', async () => {
const spawn = jest.fn((..._: any[]) => ({
const spawn = jest.fn(() => ({
stdout: {
on: () => undefined,
},
@ -291,7 +295,7 @@ describe('PipelineTaskRunner', () => {
}, 10);
}, 10);
});
const spawn = jest.fn((..._: any[]) => ({
const spawn = jest.fn(() => ({
stdout: {
on,
},
@ -304,7 +308,7 @@ describe('PipelineTaskRunner', () => {
}));
let emitSuccessCount = 0;
jest.spyOn(runner, 'emitEvent').mockImplementation((..._: any[]) => {
jest.spyOn(runner, 'emitEvent').mockImplementation(() => {
return new Promise((resolve) => {
setTimeout(() => {
emitSuccessCount++;
@ -323,4 +327,25 @@ describe('PipelineTaskRunner', () => {
});
});
});
describe('tryRunDeployScript', () => {
it('should be call deploy with right args', async () => {
const deploy = jest.spyOn(deployByPM2Service, 'deploy');
await expect(
runner['tryRunDeployScript'](
'/test/dir',
'@@DEPLOY ecosystem.config.js',
),
).resolves.toBe(true);
expect(deploy.mock.calls[0][0]).toEqual('/test/dir/ecosystem.config.js');
});
it('should return false', async () => {
await expect(
runner['tryRunDeployScript'](
'/test/dir',
'pm2 start ecosystem.config.js',
),
).resolves.toBe(false);
});
});
});

View File

@ -1,3 +1,4 @@
import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
import { ReposService } from '../repos/repos.service';
import { spawn, ChildProcessWithoutNullStreams } from 'child_process';
import { PipelineTask } from './pipeline-task.entity';
@ -27,8 +28,9 @@ import {
getSelfInstanceQueueKey,
getSelfInstanceRouteKey,
} from '../commons/utils/rabbit-mq';
import { rm, unlink } from 'fs/promises';
import { rm } from 'fs/promises';
import { rename } from 'fs/promises';
import { join } from 'path';
type Spawn = typeof spawn;
@ -43,6 +45,7 @@ export class PipelineTaskRunner {
@Inject('spawn')
private readonly spawn: Spawn,
private readonly amqpConnection: AmqpConnection,
private readonly deployByPM2Service: DeployByPm2Service,
) {}
@RabbitSubscribe({
exchange: 'new-pipeline-task',
@ -274,6 +277,9 @@ export class PipelineTaskRunner {
unit: PipelineUnits,
): Promise<void> {
await this.emitEvent(task, unit, TaskStatuses.working, script, 'stdin');
if (await this.tryRunDeployScript(workspaceRoot, script)) {
return;
}
return new Promise((resolve, reject) => {
const sub = this.spawn(script, {
shell: true,
@ -317,4 +323,17 @@ export class PipelineTaskRunner {
});
});
}
private async tryRunDeployScript(workspaceRoot: string, script: string) {
const match = /^@@DEPLOY\s+(\S*)/.exec(script);
if (match) {
await this.deployByPM2Service.deploy(
join(workspaceRoot, match[1]),
workspaceRoot,
);
return true;
} else {
return false;
}
}
}

View File

@ -7,3 +7,5 @@ export const ROUTE_PIPELINE_TASK_DONE = 'pipeline-task-done';
export const QUEUE_PIPELINE_TASK_DONE = 'pipeline-task-done';
export const ROUTE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
export const QUEUE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
export const PM2_TOKEN = Symbol('pm2-token');

View File

@ -17,6 +17,7 @@ import {
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
import { CommonsModule } from '../commons/commons.module';
import { DeployByPm2Service } from './runners/deploy-by-pm2/deploy-by-pm2.service';
@Module({
imports: [
@ -68,6 +69,7 @@ import { CommonsModule } from '../commons/commons.module';
useValue: spawn,
},
PipelineTaskFlushService,
DeployByPm2Service,
],
exports: [PipelineTasksService],
})

View File

@ -0,0 +1,117 @@
import { Test, TestingModule } from '@nestjs/testing';
import { getLoggerToken, PinoLogger } from 'nestjs-pino';
import { join } from 'path';
import { DeployByPm2Service } from './deploy-by-pm2.service';
describe('DeployByPm2Service', () => {
let service: DeployByPm2Service;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [
DeployByPm2Service,
{
provide: getLoggerToken(DeployByPm2Service.name),
useValue: new PinoLogger({
pinoHttp: {
level: 'silent',
},
}),
},
],
}).compile();
service = module.get<DeployByPm2Service>(DeployByPm2Service);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
describe('getAppsSn', () => {
it('should return right value', () => {
expect(
service['getAppsSn']([
{ name: 'app' },
{ name: 'app#4' },
{ name: 'app#1' },
]),
).toEqual(4 + 1);
});
it('should return 1 when no match', () => {
expect(
service['getAppsSn']([
{ name: 'bar' },
{ name: 'foo#4' },
{ name: 'foo#1' },
]),
).toEqual(4 + 1);
});
});
describe('filterOldApps', () => {
it('should return right value', () => {
expect(
service['filterOldApps'](
[{ name: 'app' }],
[
{ name: 'app' },
{ name: 'app#4' },
{ name: 'foo#2' },
{ name: 'bar' },
],
),
).toEqual([{ name: 'app' }, { name: 'app#4' }]);
});
it('should return [] when no match', () => {
expect(
service['filterOldApps'](
[{ name: 'app' }],
[{ name: 'foo#2' }, { name: 'bar' }],
),
).toEqual([]);
});
});
describe('replaceAppName', () => {
it('should be replaced with right value', () => {
const getAppsSn = jest
.spyOn(service, 'getAppsSn' as any)
.mockImplementation(() => 1);
const options = [{ name: 'app' }, { name: 'foo' }];
service['replaceAppName'](options, []);
expect(options).toEqual([{ name: 'app#1' }, { name: 'foo#1' }]);
expect(getAppsSn).toBeCalledTimes(1);
expect(getAppsSn.mock.calls[0][0]).toEqual([]);
});
});
describe('deploy', () => {
it('should be success', async () => {
const filterOldApps = jest
.spyOn(service, 'filterOldApps' as any)
.mockImplementation(() => [{ name: 'app#1' }, { name: 'app#2' }]);
const replaceAppName = jest
.spyOn(service, 'replaceAppName' as any)
.mockImplementation((options) => (options[0].name = 'app#2'));
const stopApps = jest
.spyOn(service, 'stopApps' as any)
.mockImplementation(() => Promise.resolve());
await expect(
service['deploy'](
join(
__dirname,
'../../../../test/__mocks__/deploy-service/ecosystem.config.js',
),
join(__dirname, '../../../../test/__mocks__/deploy-service'),
),
).resolves.toBeFalsy();
expect(filterOldApps).toBeCalledTimes(1);
expect(replaceAppName).toBeCalledTimes(1);
expect(stopApps).toBeCalledTimes(1);
stopApps.mockReset();
await service['stopApps']([{ name: 'app#2' }]);
}, 10_000);
});
});

View File

@ -0,0 +1,98 @@
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import { Injectable } from '@nestjs/common';
import { promisify } from 'util';
import * as pm2 from 'pm2';
import { Proc, ProcessDescription, StartOptions } from 'pm2';
import { clone, last } from 'ramda';
@Injectable()
export class DeployByPm2Service {
constructor(
@InjectPinoLogger(DeployByPm2Service.name)
private readonly logger: PinoLogger,
) {}
async deploy(filePath: string, workspace: string) {
const baseConfig: { apps: StartOptions[] } = await import(filePath);
const appOptionsList: StartOptions[] = clone(baseConfig.apps);
await promisify<void>(pm2.connect.bind(pm2))();
const allApps = await promisify(pm2.list.bind(pm2))();
try {
if (!Array.isArray(baseConfig.apps)) {
this.logger.error(
'the "apps" in the PM2 ecosystem configuration is not array',
);
throw new Error('apps is not array');
}
const oldApps = this.filterOldApps(appOptionsList, allApps);
this.replaceAppName(appOptionsList, oldApps);
for (const appOptions of appOptionsList) {
const proc = await promisify<StartOptions, Proc>(pm2.start.bind(pm2))({
...appOptions,
cwd: workspace,
});
this.logger.info({ proc }, `start ${appOptions.name}`);
}
await this.stopApps(oldApps);
} catch (err) {
await this.stopApps(appOptionsList);
throw err;
} finally {
pm2.disconnect();
}
}
private async stopApps(apps: ProcessDescription[] | StartOptions[]) {
await Promise.all(
apps.map(async (app: ProcessDescription | StartOptions) => {
let procAtStop: ProcessDescription;
let procAtDelete: ProcessDescription;
try {
const idOrName = 'pm_id' in app ? app.pm_id : app.name;
procAtStop = await promisify(pm2.stop.bind(pm2))(idOrName);
procAtDelete = await promisify(pm2.delete.bind(pm2))(idOrName);
this.logger.info('stop & delete %s success', app.name);
} catch (error) {
this.logger.error(
{ error, procAtStop, procAtDelete },
'stop & delete %s error',
app.name,
);
}
}),
);
}
private replaceAppName(
optionsList: StartOptions[],
oldApps: ProcessDescription[],
) {
const appSn = this.getAppsSn(oldApps);
optionsList.forEach((options) => {
if (!options.name) {
this.logger.error('please give a name for application');
throw new Error('app name is not given');
}
options.name = `${options.name}#${appSn}`;
});
}
private filterOldApps(
optionsList: StartOptions[],
apps: ProcessDescription[],
) {
return apps.filter((app) =>
optionsList.some((options) => app.name.split('#')[0] === options.name),
);
}
private getAppsSn(oldApps: ProcessDescription[]) {
const appsSn: number[] = oldApps.map(
(app) => +(app.name.split('#')?.[1] ?? 0),
);
return (last(appsSn.sort()) ?? 0) + 1;
}
}

View File

@ -0,0 +1,15 @@
module.exports = {
apps: [
{
name: 'app',
script: __dirname + '/index.js',
watch: false,
ignore_watch: ['node_modules'],
log_date_format: 'MM-DD HH:mm:ss.SSS Z',
env: {},
max_restarts: 5,
kill_timeout: 10_000,
wait_ready: true,
},
],
};

View File

@ -0,0 +1,23 @@
import { createServer } from 'http';
var app = createServer(function (req, res) {
res.writeHead(200);
setTimeout(() => {
res.end('hey');
}, 2000);
});
var listener = app.listen(0, function () {
console.log('Listening on port ' + listener.address().port);
setTimeout(() => {
// Here we send the ready signal to PM2
process.send('ready');
}, 5000);
});
process.on('SIGINT', function () {
listener.close();
setTimeout(() => {
process.exit(0);
}, 2000);
});

View File

@ -0,0 +1,10 @@
{
"name": "deploy-service",
"version": "1.0.0",
"description": "For Test",
"main": "index.js",
"type": "module",
"scripts": {},
"author": "Ivan Li",
"license": "ISC"
}

View File

@ -0,0 +1,16 @@
import { createServer } from 'http';
var app = createServer(function (req, res) {
res.writeHead(200);
setTimeout(() => {
res.end('pm2');
}, 2000);
});
var listener = app.listen(33333, function () {
console.log('Listening on port ' + listener.address().port);
setTimeout(() => {
// Here we send the ready signal to PM2
process.send('ready');
}, 5000);
});

View File

@ -10,7 +10,7 @@
"lib": ["ES2021"],
"sourceMap": true,
"outDir": "./dist",
"rootDir": "./src",
"baseUrl": "./",
"incremental": true
}
},
}