[SQS performance] High CPU usage

[dungvv-qsoft]

We are facing with an issue when turn on SQS on our server which is they use 100% of the available CPU compared around 60% when turn off queue (with the same traffic).

We run a profiler:

On the local server to be able to generate flame graph

With Queue enabled, It shows the initial spike is too high on resources occupied

This is non queue test, the initial spike is not too high compared with queue turning on:

Evidence: .cpuprofile and .text report files for 2 cases (turn on/off queue when do stress test)
https://github.com/dungvv-qsoft/cpu-profile

On our production server

Queue on: (CPU/Ram usage graph - From 3:20 - 3:30)
Maximum average CPU usage is almost 100%

Queue off: (CPU/Ram usage graph - From 4:15 to 4:30)
Maximum average CPU usage is around 60%

Environment
Node version: v16.20.1
AWS-SDK version: v2.1106.0
SQS-Consumer version: v5.7.0

Snippet to initialize SQS consumer

export const sqsProvider = {
  provide: SQS_PROVIDER,
  useFactory: (): SQS => {
    const sqs = new SQS(getOptionsForAWSService(EAWSService.SQS));
    return sqs;
  },
};

....

export const getOptionsForAWSService = (service: EAWSService) => {
  return {
    credentials: new ChainableTemporaryCredentials({
      params: {
        RoleArn: config.get<string>(`aws.${service}.roleArn`),
        RoleSessionName: `zaapi-chat-${service}-${uuidV4()}`,
      },
      masterCredentials: new Credentials({
        accessKeyId: config.get('aws.accessKeyId'),
        secretAccessKey: config.get('aws.secretAccessKey'),
      }),
    }),
    region: config.get<string>('aws.region'),
    httpOptions: {
      agent: new https.Agent({
        keepAlive: true,
        maxSockets: config.get<number>('aws.maxSockets'),
      }),
    },
  };
};

constructor(
    @Inject(SQS_PROVIDER)
    private readonly sqs: SQS,
    private readonly sagaConsumer: SagaConsumer,
  ) {
    const queueConfigMap: Record<EChatChannel | string, string> = {
      [EChatChannel.LINE]: 'aws.sqs.queues.lineWebhookQueueFifo.url',
      [EChatChannel.FACEBOOK]: 'aws.sqs.queues.fbWebhookQueueFifo.url',
      [EChatChannel.INSTAGRAM]: 'aws.sqs.queues.instaWebhookQueueFifo.url',
      [EChatChannel.SHOPEE]: 'aws.sqs.queues.shopeeWebhookQueueFifo.url',
      [EChatChannel.LAZADA]: 'aws.sqs.queues.lazadaWebhookQueueFifo.url',
      [INACTIVE_STORE_SQS_INSTANCE_NAME]:
        'aws.sqs.queues.inactiveStoreWebhookQueue.url',
    };
    Object.entries(queueConfigMap).forEach(([channel, configKey]) => {
      this.queueUrls.set(channel, config.get<string>(configKey));
    });

    const baseSqsConfig = {
      handleMessageBatch: async (messages: SQS.Message[]) => {
        const jobGroups = _.groupBy(
          messages,
          (msg) => msg.MessageAttributes?.['Name'].StringValue || '',
        );
        await Promise.all(
          Object.entries(jobGroups).map(async ([jobName, events]) => {
            await this.sagaConsumer.consume(
              {
                name: WEBHOOK_EVENTS_RECEIVER_SAGA_NAME,
                jobName,
              },
              {
                events: events.map((event) => tryParseStringToJson(event.Body)),
              },
            );
          }),
        );
      },
      messageAttributeNames: ['Name'],
      batchSize: 10,
    };
    if (WebhookReceiverSqsDuplex.webhookReceiverQueueEnabled) {
      this.queueUrls.forEach((queueUrl, name) => {
        const sqsConsumer = this.createSqsConsumer(queueUrl, baseSqsConfig);
        this.sqsConsumers.set(name, sqsConsumer);
      });
    }
  }

  private createSqsConsumer(queueUrl: string, baseConfig: any): SQSConsumer {
    const sqsConsumer = SQSConsumer.create({
      queueUrl,
      sqs: this.sqs,
      ...baseConfig,
      pollingWaitTimeMs: config.get<number>('aws.sqs.pollingWaitTimeMs'),
    });

    sqsConsumer.on('error', (error) => {
      errorLog(error, { queueUrl }, `Error on Webhook Receiver SQS`);
    });

    return sqsConsumer;
  }

  onApplicationBootstrap() {
    this.sqsConsumers.forEach((sqsConsumer) => sqsConsumer.staqrt());
  }

  beforeApplicationShutdown() {
    this.sqsConsumers.forEach((sqsConsumer) => sqsConsumer.stop());
  }