[Bug]: High CPU usage

[dungvv-qsoft]

Describe the bug

- Issue: High cpu usage

Testing 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

Testing 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%

Your minimal, reproducible example

None

Steps to reproduce

Try to run a stress test on the api that using sqs-consumer lib -> The CPU usage increase significantly (It's not that high if we turn off the queue)

Expected behavior

CPU usage should be not high compared when turn off queue (with the same traffic)

How often does this bug happen?

Every time

Screenshots or Videos

No response

Platform

• Node.js: v16.20.1

Package version

v5.7.0

AWS SDK version

v2.1106.0

Additional context

No response

[nicholasgriffintn]

I'm going to convert this to a discussion as without a proper reproduction, there's no task for us here.

For a bug ticket, we'd need exactly the area that SQS Consumer is actually responsible and there are a few other variables at play here including how you are using SQS Consumer, what your infrastructure setup is, the aws-sdk library, etc, etc.

Feel free to raise an issue if you ever pin point the issue and it is with SQS Consumer.

[andrew-the-drawer]

code snippet to initialize the 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());
  }

[nicholasgriffintn]

Hey, sorry I'm late replying, there's a lot going on there and not much to go from honestly.

But I'd still suggest that this is potentially an issue outside of the library, might be something to raise with the AWS SDK or other things that you are using, unless you can grab a report that directly links to SQS Consumer?