Configure consumer to use a SQS client that gets credentials using sts.assumeRole and post refresh

[razzpoker50]

Describe the bug

Created a node service application that is responsible to send messages to a SQS queue. The reason the processing is not done as part of the main API and offloaded to the sqs consumer ( that runs in a separate container ) is to unblock the API and run the processing in the background which will be responsible to call 3rd party API's and transform data and store the transformed data in the DB.

In order to do so I have to call the AWS.sts.assumeRole to get the temporary credentials which will be passed along to the sqs client. This is working as expected when the pod starts running but once the temporary credentials expire we start running into the error.

SQS receive message failed: The security token included in the request is expired, undefined

One way to stop seeing this error is to restart the pod which is not really ideal. What would be the best way to refresh the credentials and pass it to the sqs client so that we don't run into the above error.

async function handleMessage(message: AWS.SQS.Message) {
  const messageId = message.MessageId;

  // process the message 
}

async funtion getServiceObjectsWIthCrossAccountCredentials() {
     // call to AWS.sts.assumeRole to get temporary credentials

    // use the credentials to create the SQS client
}

const main = async () => {
  const { sqsClient } = await getServiceObjectsWIthCrossAccountCredentials();

  logger.info('START SQS consumer for datawarehouse');

  const consumer = Consumer.create({
    queueUrl,
    visibilityTimeout: 10,
    handleMessage: handleMessage,
    sqs: sqsClient,
  });

  consumer.on(
    'message_received',
    (message: AWS.SQS.Message) => {
      logger.info({
        message: `Message received with messageId: ${messageId}`,
      });
    },
  );

  consumer.on(
    'message_processed',
    async (message: AWS.SQS.Message) => {
      const messageId = message.MessageId;
      logger.info({
        message: `Message processed with messageId: ${messageId}`,
      });
    },
  );

  consumer.start();
};

main();

Trying to figure out the cleanest way to keep the application processing the messages without running into the invalid credentials error. Is stopping and restarting the container once the credentials expire the best way to go or are they any other solutions ? Any pointers on this regard would be great.

Your minimal, reproducible example

none

Steps to reproduce

None

Expected behavior

I expect the sqs consumer to process the messages with out any invalid credentials error.

How often does this bug happen?

Every time

Screenshots or Videos

No response

Platform

macOS

Package version

v5.8.0

AWS SDK version

v2.1178.0

Additional context

No response

[nicholasgriffintn]

Hey, I've converted this to a discussion as I'm not sure that there's any particular SQS Consumer.

Happy to discuss options here and raise an issue if we come to that.

I'll have a look through what you wrote and I'll get back with more if I have more to expand on this.

[razzpoker50]

Hello @nicholasgriffintn . Thank you for the reply. When using a single AWS profile or when it uses default configuration AWS SDK takes care of credentials and in case of EKS, establishing a service level account, kublet takes care of the credential management and refreshing the token after 80% of the temporary token creation. Tried using the above package as well, but we can only associate one service level account per pod and I have not really got a good handle on how to make it work with multiple profiles where the access tokens get injected.

Tried a different approach where in I ended up stopping and restarting the consumer on token expiry and passing along the updated SQS client with updated credentials so that we don't run into the expired token error.

let consumer: Consumer;

function main() {
  async function handleMessage(message: AWS.SQS.Message) {
    const messageId = message.MessageId;
  
    // process the message 
  }
  
  async funtion getServiceObjectsWIthCrossAccountCredentials() {
       // call to AWS.sts.assumeRole to get temporary credentials
  
      // use the credentials to create the SQS client
  }

  funtion shouldRefreshCrossAccountCredentials() {
    // check if the token is past expiration
  }

  function checkAndStartConsumerOnCredentialsExpiry() {
    const shouldRefreshToken = shouldRefreshCrossAccountCredentials();

    // stop the consumer and restart in when the `stopped` event is emitted
    if(shouldRefreshToken) {
       consumer.stop();
    }
  }

  /**
   * It is important that we destroy the previous SQS instance when we create a new consumer
   * as the old SQS client instance which was stopped when the credentials expire might still have
   * the sockets open.
  */
  let sqsClient = null;
  
  const main = async () => {
   if (sqsClient) {
   
        /**
          * Destroying the previous instance so that the sockets can be closed
          * when a new SQS consumer is initialized
        */
        sqsClient.destroy();
    }

    const { sqsClient: updatedSqsClient } = await getServiceObjectsWIthCrossAccountCredentials();
  
    logger.info('START SQS consumer for datawarehouse');

   sqsClient = updatedSqsClient;
  
    consumer = Consumer.create({
      queueUrl,
      visibilityTimeout: 10,
      handleMessage: handleMessage,
      sqs: sqsClient,
    });
  
    consumer.on(
      'message_received',
      (message: AWS.SQS.Message) => {
        logger.info({
          message: `Message received with messageId: ${message.MessageId}`,
        });
      },
    );
  
    consumer.on(
      'message_processed',
      async (message: AWS.SQS.Message) => {
        const messageId = message.MessageId;
        logger.info({
          message: `Message processed with messageId: ${message.MessageId}`,
        });
      },
    );

    consumer.on('stopped', async () => {
      logger.info(`SQS queue consumer stopped`);
  
      // Create the consumer again once it stops so that 
      // it uses the sqs client with new credentials
      await main();
    });
  
    consumer.start();
  };
}

main();

Not sure if this is the best way to handle this use case but I stopped running into the token expiry errors after this change, but I am a bit worried about memory leaks. It would be great to know your thoughts around this approach.

[nicholasgriffintn]

Ahh I see, yeah that should work.

You should use the latest version (with AWS SDK v3) as that has a fix in that correctly stops the consumers polling timeout (it wasn't always stopped in older versions).

Alongside that, in the latest version you can also provide a new abort option to stop that will cause it to also abort any ongoing AWS requests when stop is called.

As far as I'm aware, you shouldn't find any leaks on the latest version but we're always happy to take a bug issue if you find one.

[razzpoker50]

I wish I could use the latest version but for that to happen we need to migrate the entire code from AWS sdk v2 -> v3 which is not entirely possible at the moment.

Wondering if there is a good way to take care of the consumers polling timeout in the older version ?

[nicholasgriffintn]

That makes sense, although you would only need to migrate SQS to the V3 SDK for support with SQS Consumer, everything else could be incremented in terms of implementation, which is how other teams have done it.

Outside of that, the 5.8.0 version would have to be forked and the latest changes implemented within that, without AWS SDK V3. It's not possible or maintainable to support both versions of the SDK at the same time.

[razzpoker50]

Hey @nicholasgriffintn . Thank you for the feedback and my next plan was to use multiple versions of the sdk like you suggested.
This would be far more efficient and cleaner than forking the repo and making the changes.

[razzpoker50]

Tried to use the SQS from AWS sdk v3 and the latest sqs-consumer package and I started seeing this error now with the exact same piece of code.

Access to the resource https://sqs.us-west-2.amazonaws.com/ is denied.

Wondering if you have ever run into such an error ?

[razzpoker50]

@nicholasgriffintn Realized that the credentials needs to be passed in as an object to the SQS client instance ( for V3 ) rather than spreading them as we do it in the V2 version. Once this was done I was able to use the latest version of sqs-consumer and aws-sdk-js-v3 in tandem.

One more thing that I realized is that creating the temporary credentials using @aws-sdk/client-sts was throwing an error around Assume Role for the user and I had to instead use the AWS.STS module from the V2 to generate the temporary credentials that would be used to create the client instances. Still trying to figure out the root cause for the behavior when using STS from V3.

V2 working as expected

import AWS from 'aws-sdk';

const stsClient = new AWS.STS({
  region: 'us-west-2'
});
const assumeRoleConfig = {
  RoleArn: 'Role-Arn'
  RoleSessionName: `Test-Session`,
};

async function getCrossAccountCredentials() {
    const assumedRole = await sts.assumeRole(assumeRoleConfig).promise();
    ....
}

V3 which errors out with AssumeRole error

User: arn:aws:sts::111:assumed-role/RoleA/aws-sdk-js-session-1678730405800 is not authorized to perform: sts:AssumeRole on resource: arn:aws:iam::222:role/RoleB

import {AssumeRoleCommand, Credentials, STSClient} from '@aws-sdk/client-sts';

const stsClient = new STSClient({
    region: 'us-west-2'
});
const assumeRoleConfig = {
    RoleArn: 'Role-Arn'
    RoleSessionName: `Test-Session`,
};

async function getCrossAccountCredentials() {
    const command = new AssumeRoleCommand({
        ...assumeRoleConfig,
    });

    const assumedRole = await stsClient.send(command);
    .....
}

[nicholasgriffintn]

Looking at your code, is it possible that AssumeRoleCommand is not a fan of spreading too? Not used it myself to be honest.

[razzpoker50]

I have also tried passing it explicitly but looks like it is not taking that into consideration. Still waiting on a reply back from the aws-sdk-js-v3 contributors on that front.

For now I am using a mix of STS from V2 and the latest sqs-consumer and SQS client from V3.