From 0c3310d3a5bbb6fad50c14d7757a58e9758eb5fd Mon Sep 17 00:00:00 2001 From: Ivan Date: Tue, 8 Jun 2021 14:52:10 +0800 Subject: [PATCH] =?UTF-8?q?feat(pipeline-task):=E9=80=9A=E8=BF=87=E4=BB=BB?= =?UTF-8?q?=E5=8A=A1=E5=AE=8C=E6=88=90=E6=B6=88=E6=81=AF=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=BA=93=E4=B8=AD=E7=9A=84=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E4=BF=A1=E6=81=AF=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../pipeline-task-flush.service.ts | 7 +- .../pipeline-tasks.constants.ts | 1 + src/pipeline-tasks/pipeline-tasks.service.ts | 78 ++++++++++++++++++- 3 files changed, 82 insertions(+), 4 deletions(-) diff --git a/src/pipeline-tasks/pipeline-task-flush.service.ts b/src/pipeline-tasks/pipeline-task-flush.service.ts index b06675f..18a0e6d 100644 --- a/src/pipeline-tasks/pipeline-task-flush.service.ts +++ b/src/pipeline-tasks/pipeline-task-flush.service.ts @@ -29,12 +29,13 @@ export class PipelineTaskFlushService { queue: getSelfInstanceQueueKey(QUEUE_WRITE_PIPELINE_TASK_LOG), queueOptions: { autoDelete: true, + durable: true, }, }) async write(message: PipelineTaskEvent) { - await this.redisService - .getClient() - .rpush(this.getKey(message.taskId), JSON.stringify(message)); + const client = this.redisService.getClient(); + await client.rpush(this.getKey(message.taskId), JSON.stringify(message)); + await client.expire(this.getKey(message.taskId), 600); // ten minutes if (isNil(message.unit) && terminalTaskStatuses.includes(message.status)) { this.amqpConnection.request({ exchange: EXCHANGE_PIPELINE_TASK_TOPIC, diff --git a/src/pipeline-tasks/pipeline-tasks.constants.ts b/src/pipeline-tasks/pipeline-tasks.constants.ts index 49101c3..42d10f9 100644 --- a/src/pipeline-tasks/pipeline-tasks.constants.ts +++ b/src/pipeline-tasks/pipeline-tasks.constants.ts @@ -4,3 +4,4 @@ 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'; +export const QUEUE_PIPELINE_TASK_DONE = 'pipeline-task-done'; diff --git a/src/pipeline-tasks/pipeline-tasks.service.ts b/src/pipeline-tasks/pipeline-tasks.service.ts index 812a3c3..fdc1aa2 100644 --- a/src/pipeline-tasks/pipeline-tasks.service.ts +++ b/src/pipeline-tasks/pipeline-tasks.service.ts @@ -5,7 +5,21 @@ import { Repository } from 'typeorm'; import { CreatePipelineTaskInput } from './dtos/create-pipeline-task.input'; import { Pipeline } from '../pipelines/pipeline.entity'; import debug from 'debug'; -import { AmqpConnection } from '@golevelup/nestjs-rabbitmq'; +import { + AmqpConnection, + RabbitRPC, + RabbitSubscribe, +} from '@golevelup/nestjs-rabbitmq'; +import { + EXCHANGE_PIPELINE_TASK_FANOUT, + QUEUE_PIPELINE_TASK_DONE, + ROUTE_PIPELINE_TASK_DONE, +} from './pipeline-tasks.constants'; +import { PipelineTaskFlushService } from './pipeline-task-flush.service'; +import { find, isNil, propEq } from 'ramda'; +import { PipelineTaskLogs } from './models/pipeline-task-logs.model'; +import { TaskStatuses, terminalTaskStatuses } from './enums/task-statuses.enum'; +import { InjectPinoLogger, PinoLogger } from 'nestjs-pino'; const log = debug('fennec:pipeline-tasks:service'); @@ -17,6 +31,9 @@ export class PipelineTasksService { @InjectRepository(Pipeline) private readonly pipelineRepository: Repository, private readonly amqpConnection: AmqpConnection, + private readonly eventFlushService: PipelineTaskFlushService, + @InjectPinoLogger(PipelineTasksService.name) + private readonly logger: PinoLogger, ) {} async addTask(dto: CreatePipelineTaskInput) { const pipeline = await this.pipelineRepository.findOneOrFail({ @@ -60,4 +77,63 @@ export class PipelineTasksService { getRedisTokens(pipeline: Pipeline): [string, string] { return [`pipeline-${pipeline.id}:lck`, `pipeline-${pipeline.id}:tasks`]; } + + @RabbitRPC({ + exchange: EXCHANGE_PIPELINE_TASK_FANOUT, + routingKey: ROUTE_PIPELINE_TASK_DONE, + queue: QUEUE_PIPELINE_TASK_DONE, + queueOptions: { + autoDelete: true, + durable: true, + }, + }) + async updateByEvent({ id }: { id: string }) { + try { + const [events, task] = await Promise.all([ + this.eventFlushService.read(id), + this.findTaskById(id), + ]); + this.logger.info('[updateByEvent] start. taskId: %s', id); + + for (const event of events) { + if (isNil(event.unit)) { + if ( + event.status !== TaskStatuses.pending && + task.status === TaskStatuses.pending + ) { + task.startedAt = event.emittedAt; + } else if (terminalTaskStatuses.includes(event.status)) { + task.endedAt = event.emittedAt; + } + task.status = event.status; + } else { + let l: PipelineTaskLogs = find( + propEq('unit', event.unit), + task.logs, + ); + + if (isNil(l)) { + l = { + unit: event.unit, + startedAt: event.emittedAt, + endedAt: null, + logs: event.message, + status: event.status, + }; + + task.logs.push(l); + } + + if (terminalTaskStatuses.includes(event.status)) { + l.endedAt = event.emittedAt; + } + l.status = event.status; + } + } + await this.repository.update({ id }, task); + this.logger.info('[updateByEvent] success. taskId: %s', id); + } catch (error) { + this.logger.error({ error }, '[updateByEvent] failed. taskId: %s', id); + } + } }