package-lock.json

@@ -16,7 +16,6 @@
         "@nestjs/graphql": "^7.9.8",
         "@nestjs/platform-express": "^7.5.1",
         "@nestjs/typeorm": "^7.1.5",
-        "@types/amqplib": "^0.8.0",
         "@types/bull": "^3.15.0",
         "@types/ramda": "^0.27.38",
         "apollo-server-express": "^2.19.2",
@@ -2759,15 +2758,6 @@
         "@types/node": "*"
       }
     },
-    "node_modules/@types/amqplib": {
-      "version": "0.8.0",
-      "resolved": "https://registry.npmjs.org/@types/amqplib/-/amqplib-0.8.0.tgz",
-      "integrity": "sha512-RDojJ8WACs43HIfWSQGnAVwgNzjMGx4YMNeW7jptgAFgkG1EpNQqts+cND5HYWdYgTM58b+RHe675b0i4A9WpQ==",
-      "dependencies": {
-        "@types/bluebird": "*",
-        "@types/node": "*"
-      }
-    },
     "node_modules/@types/babel__core": {
       "version": "7.1.14",
       "resolved": "https://registry.npmjs.org/@types/babel__core/-/babel__core-7.1.14.tgz",
@@ -2809,11 +2799,6 @@
         "@babel/types": "^7.3.0"
       }
     },
-    "node_modules/@types/bluebird": {
-      "version": "3.5.35",
-      "resolved": "https://registry.npmjs.org/@types/bluebird/-/bluebird-3.5.35.tgz",
-      "integrity": "sha512-2WeeXK7BuQo7yPI4WGOBum90SzF/f8rqlvpaXx4rjeTmNssGRDHWf7fgDUH90xMB3sUOu716fUK5d+OVx0+ncQ=="
-    },
     "node_modules/@types/body-parser": {
       "version": "1.19.0",
       "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.0.tgz",
@@ -17568,15 +17553,6 @@
         "@types/node": "*"
       }
     },
-    "@types/amqplib": {
-      "version": "0.8.0",
-      "resolved": "https://registry.npmjs.org/@types/amqplib/-/amqplib-0.8.0.tgz",
-      "integrity": "sha512-RDojJ8WACs43HIfWSQGnAVwgNzjMGx4YMNeW7jptgAFgkG1EpNQqts+cND5HYWdYgTM58b+RHe675b0i4A9WpQ==",
-      "requires": {
-        "@types/bluebird": "*",
-        "@types/node": "*"
-      }
-    },
     "@types/babel__core": {
       "version": "7.1.14",
       "resolved": "https://registry.npmjs.org/@types/babel__core/-/babel__core-7.1.14.tgz",
@@ -17618,11 +17594,6 @@
         "@babel/types": "^7.3.0"
       }
     },
-    "@types/bluebird": {
-      "version": "3.5.35",
-      "resolved": "https://registry.npmjs.org/@types/bluebird/-/bluebird-3.5.35.tgz",
-      "integrity": "sha512-2WeeXK7BuQo7yPI4WGOBum90SzF/f8rqlvpaXx4rjeTmNssGRDHWf7fgDUH90xMB3sUOu716fUK5d+OVx0+ncQ=="
-    },
     "@types/body-parser": {
       "version": "1.19.0",
       "resolved": "https://registry.npmjs.org/@types/body-parser/-/body-parser-1.19.0.tgz",

package.json

@@ -29,7 +29,6 @@
     "@nestjs/graphql": "^7.9.8",
     "@nestjs/platform-express": "^7.5.1",
     "@nestjs/typeorm": "^7.1.5",
-    "@types/amqplib": "^0.8.0",
     "@types/bull": "^3.15.0",
     "@types/ramda": "^0.27.38",
     "apollo-server-express": "^2.19.2",

src/pipeline-tasks/pipeline-task-flush.service.spec.ts

@@ -50,20 +50,17 @@ describe('PipelineTaskFlushService', () => {
   });
 
   describe('write', () => {
-    const amqpMsg = {
-      properties: { headers: { sender: 'test' } },
-    } as any;
     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, amqpMsg);
+      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(1);
+      expect(request).toBeCalledTimes(0);
     });
     it('event for which task done', async () => {
       const testEvent = new PipelineTaskEvent();
@@ -71,17 +68,13 @@ describe('PipelineTaskFlushService', () => {
       testEvent.status = TaskStatuses.success;
       const rpush = jest.spyOn(redisService.getClient(), 'rpush');
       const request = jest.spyOn(amqpConnection, 'request');
-      await service.write(testEvent, amqpMsg);
+      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: {
+        payload: { taskId: 'test', status: TaskStatuses.success },
-          taskId: 'test',
-          status: TaskStatuses.success,
-          runOn: 'test',
-        },
       });
     });
   });

src/pipeline-tasks/pipeline-task-flush.service.ts

@@ -1,10 +1,10 @@
 import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
 import { Injectable } from '@nestjs/common';
-import { ConsumeMessage } from 'amqplib';
 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,
@@ -32,20 +32,16 @@ export class PipelineTaskFlushService {
       durable: true,
     },
   })
-  async write(message: PipelineTaskEvent, amqpMsg: ConsumeMessage) {
+  async write(message: PipelineTaskEvent) {
     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)) {
+    if (isNil(message.unit) && terminalTaskStatuses.includes(message.status)) {
       try {
         await this.amqpConnection.request({
           exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
           routingKey: ROUTE_PIPELINE_TASK_DONE,
-          payload: {
+          payload: { taskId: message.taskId, status: message.status },
-            taskId: message.taskId,
-            status: message.status,
-            runOn: amqpMsg.properties.headers.sender,
-          },
         });
       } catch (error) {
         console.log(error);

src/pipeline-tasks/pipeline-task.entity.ts

@@ -36,7 +36,4 @@ export class PipelineTask extends AppBaseEntity {
 
   @Column({ nullable: true })
   endedAt?: Date;
-
-  @Column({ nullable: true })
-  runOn: string;
 }

src/pipeline-tasks/pipeline-task.runner.ts

@@ -5,34 +5,19 @@ 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 {
+import { AmqpConnection, RabbitSubscribe } from '@golevelup/nestjs-rabbitmq';
-  AmqpConnection,
-  RabbitRPC,
-  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_TOPIC,
-  QUEUE_PIPELINE_TASK_KILL,
-  ROUTE_PIPELINE_TASK_KILL,
-} from './pipeline-tasks.constants';
 import {
   EXCHANGE_PIPELINE_TASK_FANOUT,
   ROUTE_PIPELINE_TASK_LOG,
 } from './pipeline-tasks.constants';
-import {
-  getInstanceName,
-  getSelfInstanceQueueKey,
-  getSelfInstanceRouteKey,
-} from '../commons/utils/rabbit-mq';
 
 type Spawn = typeof spawn;
 
 export class PipelineTaskRunner {
   readonly processes = new Map<string, ChildProcessWithoutNullStreams>();
-  readonly stopTaskIds = new Set<string>();
 
   constructor(
     private readonly reposService: ReposService,
@@ -55,27 +40,19 @@ export class PipelineTaskRunner {
       this.logger.error({ task, err }, err.message);
     }
   }
-  @RabbitRPC({
+  @RabbitSubscribe({
-    exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
+    exchange: 'stop-pipeline-task',
-    routingKey: getSelfInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL),
+    routingKey: 'mac',
-    queue: getSelfInstanceQueueKey(QUEUE_PIPELINE_TASK_KILL),
+    queue: 'mac.stop-pipeline-task',
-    queueOptions: {
-      autoDelete: true,
-      durable: true,
-    },
   })
   async onStopTask(task: PipelineTask) {
     this.logger.info({ task }, 'on stop task [%s].', task.id);
-    this.stopTaskIds.add(task.id);
     const process = this.processes.get(task.id);
     if (process) {
       this.logger.info({ task }, 'send signal SIGINT to child process.');
       process.kill('SIGINT');
 
       setTimeout(() => {
-        setTimeout(() => {
-          this.stopTaskIds.delete(task.id);
-        }, 10_000);
         if (process === this.processes.get(task.id)) {
           this.logger.info({ task }, 'send signal SIGKILL to child process.');
           process.kill('SIGKILL');
@@ -91,7 +68,6 @@ export class PipelineTaskRunner {
     } else {
       this.logger.info({ task }, 'child process is not running.');
     }
-    return true;
   }
 
   async doTask(task: PipelineTask) {
@@ -161,9 +137,6 @@ export class PipelineTaskRunner {
     try {
       for (const script of scripts) {
         this.logger.debug('begin runScript %s', script);
-        if (this.stopTaskIds.has(task.id)) {
-          throw new ApplicationException('Task is be KILLED');
-        }
         await this.runScript(script, workspaceRoot, task, unit);
         this.logger.debug('end runScript %s', script);
       }
@@ -234,11 +207,7 @@ export class PipelineTaskRunner {
       status,
     };
     this.amqpConnection
-      .publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event, {
+      .publish(EXCHANGE_PIPELINE_TASK_FANOUT, ROUTE_PIPELINE_TASK_LOG, event)
-        headers: {
-          sender: getInstanceName(),
-        },
-      })
       .catch((error) => {
         this.logger.error(
           { error, event },
@@ -291,9 +260,6 @@ export class PipelineTaskRunner {
         if (code === 0) {
           return resolve();
         }
-        if (this.stopTaskIds.has(task.id)) {
-          throw reject(new ApplicationException('Task is be KILLED'));
-        }
         return reject(new ApplicationException('exec script failed'));
       });
     });

src/pipeline-tasks/pipeline-tasks.constants.ts

@@ -5,5 +5,3 @@ 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';
-export const ROUTE_PIPELINE_TASK_KILL = 'pipeline-task-kill';
-export const QUEUE_PIPELINE_TASK_KILL = 'pipeline-task-kill';

src/pipeline-tasks/pipeline-tasks.resolver.ts

@@ -55,7 +55,5 @@ export class PipelineTasksResolver {
   @Mutation(() => Boolean)
   async stopPipelineTask(@Args('id') id: string) {
     const task = await this.service.findTaskById(id);
-    await this.service.stopTask(task);
-    return true;
   }
 }

src/pipeline-tasks/pipeline-tasks.service.ts

@@ -1,4 +1,4 @@
-import { BadRequestException, Injectable } from '@nestjs/common';
+import { Injectable } from '@nestjs/common';
 import { InjectRepository } from '@nestjs/typeorm';
 import { PipelineTask } from './pipeline-task.entity';
 import { Repository } from 'typeorm';
@@ -7,6 +7,7 @@ import { Pipeline } from '../pipelines/pipeline.entity';
 import debug from 'debug';
 import { AmqpConnection, RabbitRPC } from '@golevelup/nestjs-rabbitmq';
 import {
+  EXCHANGE_PIPELINE_TASK_FANOUT,
   EXCHANGE_PIPELINE_TASK_TOPIC,
   QUEUE_PIPELINE_TASK_DONE,
   ROUTE_PIPELINE_TASK_DONE,
@@ -16,8 +17,6 @@ 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';
-import { getAppInstanceRouteKey } from '../commons/utils/rabbit-mq';
-import { ROUTE_PIPELINE_TASK_KILL } from './pipeline-tasks.constants';
 
 const log = debug('fennec:pipeline-tasks:service');
 
@@ -85,7 +84,7 @@ export class PipelineTasksService {
       durable: true,
     },
   })
-  async updateByEvent({ taskId, runOn }: { taskId: string; runOn: string }) {
+  async updateByEvent({ taskId }: { taskId: string }) {
     try {
       const [events, task] = await Promise.all([
         this.eventFlushService.read(taskId),
@@ -128,10 +127,9 @@ export class PipelineTasksService {
           l.status = event.status;
         }
       }
-      task.runOn = runOn;
       await this.repository.update({ id: taskId }, task);
-      this.logger.info('[updateByEvent] success. taskId: %s', taskId);
       return task;
+      this.logger.info('[updateByEvent] success. taskId: %s', taskId);
     } catch (error) {
       this.logger.error(
         { error },
@@ -140,17 +138,4 @@ export class PipelineTasksService {
       );
     }
   }
-
-  async stopTask(task: PipelineTask) {
-    if (isNil(task.runOn)) {
-      throw new BadRequestException(
-        "the task have not running instance on database. field 'runOn' is nil",
-      );
-    }
-    await this.amqpConnection.request({
-      exchange: EXCHANGE_PIPELINE_TASK_TOPIC,
-      routingKey: getAppInstanceRouteKey(ROUTE_PIPELINE_TASK_KILL, task.runOn),
-      payload: task,
-    });
-  }
 }