Fennec/fennec-be
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Compare commits

base: Fennec:ead32a1204640c7edaa4e27c65f7812c534bd5a7
Branches Tags
Fennec:master
Fennec:feat-configuration-center
Fennec:single-deploy-folder
Fennec:feat-jwt-auth
Fennec:feat_the_progress_of_tasks
Fennec:feat-api-for-webhook
Fennec:feat-pipelines
Fennec:feat-repo+mq
..
compare: Fennec:3ee41ece67ce14fa68a4573fe1414390ca3669f1
Branches Tags
Fennec:feat-configuration-center
Fennec:master
Fennec:single-deploy-folder
Fennec:feat-jwt-auth
Fennec:feat_the_progress_of_tasks
Fennec:feat-api-for-webhook
Fennec:feat-pipelines
Fennec:feat-repo+mq

No commits in common. "ead32a1204640c7edaa4e27c65f7812c534bd5a7" and "3ee41ece67ce14fa68a4573fe1414390ca3669f1" have entirely different histories.
ead32a1204 ... 3ee41ece67

13 changed files with 10 additions and 338 deletions
Show Stats Expand all files Collapse all files

2
src/pipeline-tasks/enums/task-statuses.enum.ts
View File

@ -11,5 +11,3 @@ registerEnumType(TaskStatuses, {
name: 'TaskStatuses',
description: '任务状态',
});
export const terminalTaskStatuses = [TaskStatuses.success, TaskStatuses.failed];

12
src/pipeline-tasks/models/pipeline-task-event.ts
View File

@ -1,25 +1,13 @@
import { Field, ObjectType } from '@nestjs/graphql';
import { PipelineUnits } from '../enums/pipeline-units.enum';
import { TaskStatuses } from '../enums/task-statuses.enum';
import { Type } from 'class-transformer';
@ObjectType()
export class PipelineTaskEvent {
@Field()
taskId: string;
@Field()
pipelineId: string;
@Field()
projectId: string;
@Field(() => PipelineUnits, { nullable: true })
unit: PipelineUnits | null;
@Field()
@Type(() => Date)
emittedAt: Date;
@Field()
message: string;
@Field()
messageType: 'stdout' | 'stderr' | 'stdin';
@Field(() => TaskStatuses)
status: TaskStatuses;
}

80
src/pipeline-tasks/pipeline-task-flush.service.spec.ts
View File

@ -1,80 +0,0 @@
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import { Test, TestingModule } from '@nestjs/testing';
import { RedisService } from 'nestjs-redis';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { TaskStatuses } from './enums/task-statuses.enum';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
ROUTE_PIPELINE_TASK_DONE,
} from './pipeline-tasks.constants';
describe('PipelineTaskFlushService', () => {
let service: PipelineTaskFlushService;
let redisService: RedisService;
let amqpConnection: AmqpConnection;
beforeEach(async () => {
const redisClient = {
rpush: jest.fn(() => Promise.resolve()),
lrange: jest.fn(() => Promise.resolve()),
};
const module: TestingModule = await Test.createTestingModule({
providers: [
PipelineTaskFlushService,
{
provide: RedisService,
useValue: {
getClient() {
return redisClient;
},
},
},
{
provide: AmqpConnection,
useValue: {
request: jest.fn(() => Promise.resolve()),
},
},
],
}).compile();
service = module.get<PipelineTaskFlushService>(PipelineTaskFlushService);
redisService = module.get<RedisService>(RedisService);
amqpConnection = module.get<AmqpConnection>(AmqpConnection);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
describe('write', () => {
it('normal', async () => {
const testEvent = new PipelineTaskEvent();
testEvent.taskId = 'test';
testEvent.status = TaskStatuses.working;
const rpush = jest.spyOn(redisService.getClient(), 'rpush');
const request = jest.spyOn(amqpConnection, 'request');
await service.write(testEvent);
expect(rpush).toBeCalledTimes(1);
expect(rpush.mock.calls[0][0]).toEqual('p-task:log:test');
expect(rpush.mock.calls[0][1]).toEqual(JSON.stringify(testEvent));
expect(request).toBeCalledTimes(0);
});
it('event for which task done', async () => {
const testEvent = new PipelineTaskEvent();
testEvent.taskId = 'test';
testEvent.status = TaskStatuses.success;
const rpush = jest.spyOn(redisService.getClient(), 'rpush');
const request = jest.spyOn(amqpConnection, 'request');
await service.write(testEvent);
expect(rpush).toBeCalledTimes(1);
expect(request).toBeCalledTimes(1);
expect(request.mock.calls[0][0]).toMatchObject({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: ROUTE_PIPELINE_TASK_DONE,
payload: { taskId: 'test', status: TaskStatuses.success },
});
});
});
});

57
src/pipeline-tasks/pipeline-task-flush.service.ts
View File

@ -1,57 +0,0 @@
import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable } from '@nestjs/common';
import { deserialize } from 'class-transformer';
import { RedisService } from 'nestjs-redis';
import { isNil } from 'ramda';
import { getSelfInstanceQueueKey } from '../commons/utils/rabbit-mq';
import { terminalTaskStatuses } from './enums/task-statuses.enum';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import {
EXCHANGE_PIPELINE_TASK_TOPIC,
ROUTE_PIPELINE_TASK_DONE,
} from './pipeline-tasks.constants';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
ROUTE_PIPELINE_TASK_LOG,
QUEUE_WRITE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
@Injectable()
export class PipelineTaskFlushService {
constructor(
private readonly redisService: RedisService,
private readonly amqpConnection: AmqpConnection,
) {}
@RabbitSubscribe({
exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
routingKey: ROUTE_PIPELINE_TASK_LOG,
queue: getSelfInstanceQueueKey(QUEUE_WRITE_PIPELINE_TASK_LOG),
queueOptions: {
autoDelete: true,
},
})
async write(message: PipelineTaskEvent) {
await this.redisService
.getClient()
.rpush(this.getKey(message.taskId), JSON.stringify(message));
if (isNil(message.unit) && terminalTaskStatuses.includes(message.status)) {
this.amqpConnection.request({
exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
routingKey: ROUTE_PIPELINE_TASK_DONE,
payload: { taskId: message.taskId, status: message.status },
});
}
}
async read(taskId: string) {
const raw = await this.redisService
.getClient()
.lrange(this.getKey(taskId), 0, -1);
return raw.map((it) => deserialize(PipelineTaskEvent, it));
}
private getKey(taskId: string) {
return `p-task:log:${taskId}`;
}
}

74
src/pipeline-tasks/pipeline-task.logger.spec.ts
View File

@ -1,74 +0,0 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { take, timeout } from 'rxjs/operators';
describe('PipelineTaskRunner', () => {
let logger: PipelineTaskLogger;
let module: TestingModule;
beforeEach(async () => {
module = await Test.createTestingModule({
providers: [PipelineTaskLogger],
}).compile();
logger = module.get(PipelineTaskLogger);
});
it('should be defined', () => {
expect(logger).toBeDefined();
});
describe('getMessage$', () => {
it('normal', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const emittedAt = new Date();
event.emittedAt = emittedAt.toISOString() as any;
const message$ = logger.getMessage$('test');
let receiveEvent;
message$.pipe(take(1)).subscribe((value) => (receiveEvent = value));
await logger.handleEvent(event);
expect(receiveEvent).toMatchObject({
...event,
emittedAt,
});
});
it('no match', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const message$ = logger.getMessage$('other');
setTimeout(() => {
logger.handleEvent(event);
});
expect(message$.pipe(take(1), timeout(100)).toPromise()).rejects.toMatch(
'timeout',
);
});
it('multiple subscribers', async () => {
const event = new PipelineTaskEvent();
event.taskId = 'test';
const message$ = logger.getMessage$('test');
const message2$ = logger.getMessage$('test');
setTimeout(() => {
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);
});
});
describe('onModuleDestroy', () => {
it('complete observable when destroying module', async () => {
logger.onModuleDestroy();
await expect(
(logger as any).message$.toPromise(),
).resolves.toBeUndefined();
});
});
});

37
src/pipeline-tasks/pipeline-task.logger.ts
View File

@ -1,37 +0,0 @@
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { plainToClass } from 'class-transformer';
import { Observable, Subject } from 'rxjs';
import { filter } from 'rxjs/operators';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
ROUTE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
@Injectable()
export class PipelineTaskLogger implements OnModuleDestroy {
private readonly messageSubject = new Subject<PipelineTaskEvent>();
private readonly message$: Observable<PipelineTaskEvent> = this.messageSubject.pipe();
@RabbitSubscribe({
exchange: EXCHANGE_PIPELINE_TASK_FANOUT,
routingKey: ROUTE_PIPELINE_TASK_LOG,
queue: QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT,
queueOptions: {
autoDelete: true,
},
})
async handleEvent(message: PipelineTaskEvent) {
this.messageSubject.next(plainToClass(PipelineTaskEvent, message));
}
getMessage$(taskId: string) {
return this.message$.pipe(filter((event) => event.taskId === taskId));
}
onModuleDestroy() {
this.messageSubject.complete();
}
}

5
src/pipeline-tasks/pipeline-task.runner.spec.ts
View File

@ -8,7 +8,6 @@ import { TaskStatuses } from './enums/task-statuses.enum';
import { getLoggerToken, PinoLogger } from 'nestjs-pino';
import { PipelineTaskRunner } from './pipeline-task.runner';
import { WorkUnitMetadata } from './models/work-unit-metadata.model';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('PipelineTaskRunner', () => {
let runner: PipelineTaskRunner;
let reposService: ReposService;
@ -32,10 +31,6 @@ describe('PipelineTaskRunner', () => {
useValue: () => undefined,
},
PipelineTaskRunner,
{
provide: AmqpConnection,
useValue: {},
},
],
}).compile();

16
src/pipeline-tasks/pipeline-task.runner.ts
View File

@ -5,14 +5,10 @@ import { ApplicationException } from '../commons/exceptions/application.exceptio
import { PipelineUnits } from './enums/pipeline-units.enum';
import { TaskStatuses } from './enums/task-statuses.enum';
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { PipelineTaskEvent } from './models/pipeline-task-event';
import { last } from 'ramda';
import { Inject } from '@nestjs/common';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
ROUTE_PIPELINE_TASK_LOG,
} from './pipeline-tasks.constants';
type Spawn = typeof spawn;
@ -25,7 +21,6 @@ export class PipelineTaskRunner {
private readonly logger: PinoLogger,
@Inject('spawn')
private readonly spawn: Spawn,
private readonly amqpConnection: AmqpConnection,
) {}
@RabbitSubscribe({
exchange: 'new-pipeline-task',
@ -206,15 +201,6 @@ export class PipelineTaskRunner {
messageType,
status,
};
this.amqpConnection
.publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event)
.catch((error) => {
this.logger.error(
{ error, event },
'send event message to queue failed. %s',
error.message,
);
});
}
async runScript(

6
src/pipeline-tasks/pipeline-tasks.constants.ts
View File

@ -1,6 +0,0 @@
export const EXCHANGE_PIPELINE_TASK_TOPIC = 'pipeline-task.topic';
export const EXCHANGE_PIPELINE_TASK_FANOUT = 'pipeline-task.fanout';
export const ROUTE_PIPELINE_TASK_LOG = 'pipeline-task-log';
export const QUEUE_HANDLE_PIPELINE_TASK_LOG_EVENT = 'pipeline-task-log';
export const QUEUE_WRITE_PIPELINE_TASK_LOG = 'write-pipeline-task-log';
export const ROUTE_PIPELINE_TASK_DONE = 'pipeline-task-done';

27
src/pipeline-tasks/pipeline-tasks.module.ts
View File

@ -10,12 +10,6 @@ import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { PipelineTaskRunner } from './pipeline-task.runner';
import { spawn } from 'child_process';
import {
EXCHANGE_PIPELINE_TASK_FANOUT,
EXCHANGE_PIPELINE_TASK_TOPIC,
} from './pipeline-tasks.constants';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTaskFlushService } from './pipeline-task-flush.service';
@Module({
imports: [
@ -33,7 +27,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
type: 'fanout',
options: {
durable: true,
autoDelete: true,
},
},
{
@ -41,7 +34,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
type: 'fanout',
options: {
durable: true,
autoDelete: true,
},
},
{
@ -49,23 +41,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
{
name: EXCHANGE_PIPELINE_TASK_FANOUT,
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
{
name: EXCHANGE_PIPELINE_TASK_TOPIC,
type: 'topic',
options: {
durable: false,
autoDelete: true,
},
},
],
@ -77,12 +52,10 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
PipelineTasksService,
PipelineTasksResolver,
PipelineTaskRunner,
PipelineTaskLogger,
{
provide: 'spawn',
useValue: spawn,
},
PipelineTaskFlushService,
],
exports: [PipelineTasksService],
})

5
src/pipeline-tasks/pipeline-tasks.resolver.spec.ts
View File

@ -1,5 +1,4 @@
import { Test, TestingModule } from '@nestjs/testing';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { PipelineTasksResolver } from './pipeline-tasks.resolver';
import { PipelineTasksService } from './pipeline-tasks.service';
@ -14,10 +13,6 @@ describe('PipelineTasksResolver', () => {
provide: PipelineTasksService,
useValue: {},
},
{
provide: PipelineTaskLogger,
useValue: {},
},
],
}).compile();

22
src/pipeline-tasks/pipeline-tasks.resolver.ts
View File

@ -2,35 +2,29 @@ import { Resolver, Args, Mutation, Subscription, Query } from '@nestjs/graphql';
import { PipelineTask } from './pipeline-task.entity';
import { PipelineTasksService } from './pipeline-tasks.service';
import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input';
import { PipelineTaskLogMessage } from './models/pipeline-task-log-message.module';
import { PipelineTaskLogArgs } from './dtos/pipeline-task-log.args';
import { plainToClass } from 'class-transformer';
import { PipelineTaskLogger } from './pipeline-task.logger';
import { observableToAsyncIterable } from '@graphql-tools/utils';
import { PipelineTaskEvent } from './models/pipeline-task-event';
@Resolver()
export class PipelineTasksResolver {
constructor(
private readonly service: PipelineTasksService,
private readonly taskLogger: PipelineTaskLogger,
) {}
constructor(private readonly service: PipelineTasksService) {}
@Mutation(() => PipelineTask)
async createPipelineTask(@Args('task') taskDto: CreatePipelineTaskInput) {
return await this.service.addTask(taskDto);
}
@Subscription(() => PipelineTaskEvent, {
@Subscription(() => PipelineTaskLogMessage, {
resolve: (value) => {
const data = plainToClass(PipelineTaskEvent, value);
const data = plainToClass(PipelineTaskLogMessage, value);
return data;
},
})
async pipelineTaskEvent(@Args() args: PipelineTaskLogArgs) {
const task = await this.service.findTaskById(args.taskId);
return observableToAsyncIterable<PipelineTaskEvent>(
this.taskLogger.getMessage$(task.id),
);
async pipelineTaskLog(@Args() args: PipelineTaskLogArgs) {
// const task = await this.service.findTaskById(args.taskId);
// const asyncIterator = this.logsService.watchLogs(task);
// return asyncIterator;
}
@Subscription(() => PipelineTask, {

5
src/pipeline-tasks/pipeline-tasks.service.ts
View File

@ -51,10 +51,7 @@ export class PipelineTasksService {
}
async listTasksByCommitHash(hash: string) {
return await this.repository.find({
where: { commit: hash },
order: { createdAt: 'DESC' },
});
return await this.repository.find({ commit: hash });
}
getRedisTokens(pipeline: Pipeline): [string, string] {