feat_the_progress_of_tasks #5

Merged
Ivan merged 20 commits from feat_the_progress_of_tasks into master 2021-06-27 19:57:14 +08:00
14 changed files with 201 additions and 24 deletions
Showing only changes of commit a231a02c28 - Show all commits

View File

@ -1,5 +1,6 @@
{
"cSpell.words": [
"Mutex",
"Repos",
"amqp",
"boardcat",

View File

@ -1,8 +1,10 @@
import { Module } from '@nestjs/common';
import { PasswordConverter } from './services/password-converter';
import { RedisMutexModule } from './redis-mutex/redis-mutex.module';
@Module({
providers: [PasswordConverter],
exports: [PasswordConverter],
exports: [PasswordConverter, RedisMutexModule],
imports: [RedisMutexModule],
})
export class CommonsModule {}

View File

@ -0,0 +1,10 @@
import { Module } from '@nestjs/common';
import { RedisMutexService } from './redis-mutex.service';
import { RedisModule } from 'nestjs-redis';
@Module({
imports: [RedisModule],
providers: [RedisMutexService],
exports: [RedisMutexService],
})
export class RedisMutexModule {}

View File

@ -0,0 +1,18 @@
import { Test, TestingModule } from '@nestjs/testing';
import { RedisMutexService } from './redis-mutex.service';
describe('RedisMutexService', () => {
let service: RedisMutexService;
beforeEach(async () => {
const module: TestingModule = await Test.createTestingModule({
providers: [RedisMutexService],
}).compile();
service = module.get<RedisMutexService>(RedisMutexService);
});
it('should be defined', () => {
expect(service).toBeDefined();
});
});

View File

@ -0,0 +1,71 @@
import { Injectable } from '@nestjs/common';
import { RedisService } from 'nestjs-redis';
import * as uuid from 'uuid';
import { ApplicationException } from '../exceptions/application.exception';
export interface RedisMutexOption {
/**
* seconds
*/
expires?: number;
/**
* seconds
*/
timeout?: number | null;
/**
* milliseconds
*/
retryDelay?: number;
}
@Injectable()
export class RedisMutexService {
constructor(private readonly redisClient: RedisService) {}
public async lock(
key: string,
{ expires = 100, timeout = 10, retryDelay = 100 }: RedisMutexOption = {
expires: 100,
timeout: 10,
retryDelay: 100,
},
) {
const redisKey = `${'mutex-lock'}:${key}`;
const redis = this.redisClient.getClient();
const value = uuid.v4();
const timeoutAt = timeout ? Date.now() + timeout * 1000 : null;
while (
!(await redis
.set(redisKey, value, 'EX', expires, 'NX')
.then(() => true)
.catch(() => false))
) {
if (timeoutAt && timeoutAt > Date.now()) {
throw new ApplicationException('lock timeout');
}
await new Promise((resolve) => setTimeout(resolve, retryDelay));
}
const renewTimer = setInterval(() => {
redis.expire(redisKey, expires);
}, (expires * 1000) / 2);
return async () => {
clearInterval(renewTimer);
await redis.eval(
`
if redis.call("get", KEYS[1]) == ARGV[1]
then
return redis.call("del", KEYS[1])
else
return 0
end
`,
1,
redisKey,
value,
);
};
}
}

View File

@ -22,7 +22,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
TypeOrmModule.forFeature([PipelineTask, Pipeline]),
RedisModule,
ReposModule,
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
@ -36,22 +35,6 @@ import { PipelineTaskFlushService } from './pipeline-task-flush.service';
autoDelete: true,
},
},
{
name: 'stop-pipeline-task',
type: 'fanout',
options: {
durable: true,
autoDelete: true,
},
},
{
name: 'update-pipeline-task',
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
{
name: EXCHANGE_PIPELINE_TASK_FANOUT,
type: 'fanout',

View File

@ -57,7 +57,7 @@ export class PipelinesService extends BaseDbService<Pipeline> {
exchange: EXCHANGE_REPO,
routingKey: getAppInstanceRouteKey(ROUTE_FETCH, appInstance),
payload: pipeline,
timeout: 30_000,
timeout: 120_000,
});
}
async listCommits(pipeline: Pipeline) {

View File

@ -0,0 +1,3 @@
export const EXCHANGE_PROJECT_TOPIC = 'project.topic';
export const EXCHANGE_PROJECT_FANOUT = 'project.fanout';
export const ROUTE_PROJECT_CHANGE = 'project-change';

View File

@ -3,9 +3,31 @@ import { ProjectsService } from './projects.service';
import { ProjectsResolver } from './projects.resolver';
import { TypeOrmModule } from '@nestjs/typeorm';
import { Project } from './project.entity';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { EXCHANGE_PROJECT_FANOUT } from './projects.constants';
@Module({
imports: [TypeOrmModule.forFeature([Project])],
imports: [
TypeOrmModule.forFeature([Project]),
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({
uri: configService.get<string>('db.rabbitmq.uri'),
exchanges: [
{
name: EXCHANGE_PROJECT_FANOUT,
type: 'fanout',
options: {
durable: false,
autoDelete: true,
},
},
],
}),
inject: [ConfigService],
}),
],
providers: [ProjectsService, ProjectsResolver],
exports: [ProjectsService],
})

View File

@ -2,6 +2,7 @@ import { Test, TestingModule } from '@nestjs/testing';
import { ProjectsService } from './projects.service';
import { getRepositoryToken } from '@nestjs/typeorm';
import { Project } from './project.entity';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
describe('ProjectsService', () => {
let service: ProjectsService;
@ -14,6 +15,10 @@ describe('ProjectsService', () => {
provide: getRepositoryToken(Project),
useValue: {},
},
{
provide: AmqpConnection,
useValue: {},
},
],
}).compile();

View File

@ -5,6 +5,11 @@ import { Repository } from 'typeorm';
import { CreateProjectInput } from './dtos/create-project.input';
import { Project } from './project.entity';
import { UpdateProjectInput } from './dtos/update-project.input';
import { AmqpConnection } from '@golevelup/nestjs-rabbitmq';
import {
EXCHANGE_PROJECT_FANOUT,
ROUTE_PROJECT_CHANGE,
} from './projects.constants';
@Injectable()
export class ProjectsService extends BaseDbService<Project> {
@ -12,6 +17,7 @@ export class ProjectsService extends BaseDbService<Project> {
constructor(
@InjectRepository(Project)
readonly repository: Repository<Project>,
private readonly amqpConnection: AmqpConnection,
) {
super(repository);
}
@ -28,7 +34,12 @@ export class ProjectsService extends BaseDbService<Project> {
async update(dto: UpdateProjectInput) {
await this.isDuplicateEntityForUpdate(dto.id, dto);
const old = await this.findOne(dto.id);
return await this.repository.save(this.repository.merge(old, dto));
const project = await this.repository.save(this.repository.merge(old, dto));
this.amqpConnection.publish(EXCHANGE_PROJECT_FANOUT, ROUTE_PROJECT_CHANGE, [
project,
old,
]);
return project;
}
async remove(id: string) {

View File

@ -3,3 +3,4 @@ export const ROUTE_FETCH = 'fetch';
export const ROUTE_LIST_COMMITS = 'list-commits';
export const QUEUE_LIST_COMMITS = 'list-commits';
export const QUEUE_FETCH = 'repo-fetch';
export const QUEUE_REFRESH_REPO = 'refresh-repo';

View File

@ -7,12 +7,14 @@ import { ConfigModule, ConfigService } from '@nestjs/config';
import { ProjectsModule } from '../projects/projects.module';
import { EXCHANGE_REPO } from './repos.constants';
import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { CommonsModule } from '../commons/commons.module';
@Module({
imports: [
TypeOrmModule.forFeature([Project]),
ConfigModule,
ProjectsModule,
CommonsModule,
RabbitMQModule.forRootAsync(RabbitMQModule, {
imports: [ConfigModule],
useFactory: (configService: ConfigService) => ({

View File

@ -11,13 +11,14 @@ import { Project } from '../projects/project.entity';
import { ListBranchesArgs } from './dtos/list-branches.args';
import { ConfigService } from '@nestjs/config';
import { Commit } from './dtos/log-list.model';
import { Nack, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
import { Nack, RabbitRPC, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
import { Pipeline } from '../pipelines/pipeline.entity';
import { InjectPinoLogger, Logger } from 'nestjs-pino';
import { InjectPinoLogger, PinoLogger } from 'nestjs-pino';
import {
EXCHANGE_REPO,
QUEUE_FETCH,
QUEUE_LIST_COMMITS,
QUEUE_REFRESH_REPO,
ROUTE_FETCH,
ROUTE_LIST_COMMITS,
} from './repos.constants';
@ -27,6 +28,12 @@ import {
getSelfInstanceRouteKey,
} from '../commons/utils/rabbit-mq';
import { ApplicationException } from '../commons/exceptions/application.exception';
import {
EXCHANGE_PROJECT_FANOUT,
ROUTE_PROJECT_CHANGE,
} from '../projects/projects.constants';
import { RedisMutexService } from '../commons/redis-mutex/redis-mutex.service';
import { rm } from 'fs/promises';
const DEFAULT_REMOTE_NAME = 'origin';
const INFO_PATH = '@info';
@ -37,7 +44,8 @@ export class ReposService {
private readonly projectRepository: Repository<Project>,
private readonly configService: ConfigService,
@InjectPinoLogger(ReposService.name)
private readonly logger: Logger,
private readonly logger: PinoLogger,
private readonly redisMutexService: RedisMutexService,
) {}
getWorkspaceRoot(project: Project): string {
@ -170,6 +178,9 @@ export class ReposService {
},
})
async fetch(pipeline: Pipeline): Promise<string | null | Nack> {
const unlock = await this.redisMutexService.lock(
`repo-project-${pipeline.projectId}`,
);
try {
const git = await this.getGit(pipeline.project, undefined, {
fetch: false,
@ -179,6 +190,43 @@ export class ReposService {
} catch (error) {
this.logger.error({ error, pipeline }, '[fetch] %s', error?.message);
return new Nack();
} finally {
await unlock();
}
}
@RabbitSubscribe({
exchange: EXCHANGE_PROJECT_FANOUT,
routingKey: ROUTE_PROJECT_CHANGE,
queue: QUEUE_REFRESH_REPO,
queueOptions: {
autoDelete: true,
durable: true,
},
})
async refreshRepo([project]: [Project]) {
this.logger.info({ project }, '[refreshRepo] start');
const unlock = await this.redisMutexService.lock(
`repo-project-${project.id}`,
{
timeout: null,
},
);
try {
const path = join(
this.configService.get<string>('workspaces.root'),
encodeURIComponent(project.name),
);
await rm(path, { recursive: true });
this.logger.info({ project }, '[refreshRepo] success');
} catch (error) {
this.logger.error(
{ project, error },
'[refreshRepo] failed. $s',
error.message,
);
} finally {
await unlock();
}
}
}