AWS CDK 101 - Ajout d'une file d'attente pour mettre en mémoire tampon notre stepfunction invoquer lambda

AWS CDK 101 – Ajout d’une file d’attente pour mettre en mémoire tampon notre stepfunction invoquer lambda

🔰 Débutants qui découvrent AWS CDK, veuillez consulter mes articles précédents un par un dans cette série.

Si au cas où vous auriez manqué mon article précédent, retrouvez-le avec les liens ci-dessous.

🔁 Message précédent original sur 🔗 Dev Post

🔁 Republié le post précédent à 🔗 dev à @aravindvcyber

Dans cet article, refactorisons notre stepfunction précédente qui invoque directement notre fonction lambda en ajoutant une file d’attente similaire qui déclenche lambda indirectement avec une limite de taille de lot. Cela pourrait nous aider à obtenir une meilleure optimisation de la concurrence lambda comme celle de notre article précédent. traitement évolutif piloté par les événements à l’aide d’eventbridge et de sqs

Flux de courant ☘️

Proposé 🍀

Avantages obtenus dans cette approche 🚣‍♀️

Il y a plusieurs avantages associés à cette approche comme suit.

Nous pouvons détacher l’invocation directe de notre lambda par Stepfunction déclenchant ainsi une invocation indirecte/bufférisée à partir d’une file d’attente SQS standard.
Avec cette approche, nous pourrons avoir la limite de simultanéité lambda au minimum en regroupant les déclencheurs lambda.
De plus, si vous vous souvenez d’une chose, notre lambda est censé effectuer une opération d’écriture dynamodb, qui, si elle s’exécute en haute simultanéité, nécessiterait plus d’unités d’écriture lors de l’utilisation de notre table dynamodb. Avec cette approche, nous pouvons utiliser maintenir et utiliser des unités d’écriture cohérentes dans notre table.
En plus de cela, nous pourrons facilement le mettre à l’échelle dans des environnements plus élevés.
Nous pouvons également utiliser les files d’attente de lettres mortes pour le débogage et l’inspection.

Affiner la file d’attente utilisée dans notre construction précédente 🔖

Modifions notre construction précédente construct/sfn-simple.ts en ajoutant des définitions pour la file d’attente de lettres mortes et la file d’attente de travail comme suit.

File d’attente de lettres mortes 🎾

La file d’attente de lettres mortes doit récupérer les messages ayant échoué, de sorte que la file d’attente de travail a toujours des enregistrements à traiter.

 const sfnCommonEventProcessorQueueDLQ: DeadLetterQueue = {
      queue: new Queue(this, "sfnCommonEventProcessorQueueDLQ", {
        retentionPeriod: Duration.days(14),
        removalPolicy: RemovalPolicy.DESTROY,
        queueName: "sfnCommonEventProcessorQueueDLQ",
      }),
      maxReceiveCount: 100,
};

File d’attente standard 🏊

La file d’attente standard peut également être une file d’attente FIFO basée sur le coût et les exigences commerciales, comme indiqué ci-dessous.


const sfnCommonEventProcessorQueue = new Queue(
      this,
      "sfnCommonEventProcessorQueue",
      {
        retentionPeriod: Duration.days(5),
        removalPolicy: RemovalPolicy.DESTROY,
        deliveryDelay: Duration.seconds(3),
        queueName: "sfnCommonEventProcessorQueue",
        visibilityTimeout: Duration.minutes(100),
        deadLetterQueue: sfnCommonEventProcessorQueueDLQ,
      }
);

Nouveau travail pour envoyer des messages dans notre file d’attente 🎯

Tout comme nous avons défini une nouvelle tâche d’invocation lambda, nous pouvons définir une nouvelle tâche qui poussera un message dans une file d’attente qui sera ensuite interrogée par notre lambda existant de manière asynchrone.

Vous pouvez également constater que nous avons entièrement réutilisé la charge utile, inputPath, resultPath, resultSelector, etc.

const recordingQueue = new tasks.SqsSendMessage(
      this,
      "Record using Queue",
      {
        inputPath: "$",
        messageBody: sfnTaskPayload,
        queue: sfnCommonEventProcessorQueue,
        comment: "Record message into dynamodb using SQS queue buffered lambda",
        integrationPattern: sfn.IntegrationPattern.WAIT_FOR_TASK_TOKEN,
        resultSelector: {
          "Payload.$": "$",
          "StatusCode.$": "$.statusCode",
        },
        resultPath: "$.recordResult",
      }
);

Ajout d’une source d’événement pour le lambda existant ⛺

Nous allons maintenant connecter notre nouvelle file d’attente à notre lambda existant à l’intérieur de la même construction.

Comme vous pouvez le voir, nous avons défini une taille de lot de la file d’attente, même s’il y a n nombre de fonctions d’état exécutées simultanément. Vous pouvez également utiliser un FIFO file d’attente si vous constatez que l’ordre des messages est très important lors de l’enregistrement et du traitement.

triggerFunction.addEventSource(
      new SqsEventSource(sfnCommonEventProcessorQueue, {
        batchSize: 2,
        maxBatchingWindow: Duration.seconds(10)
      })
    );

Ici batchSize est le nombre maximum d’enregistrements à récupérer par invocation et maxBatchingWindow définit le temps d’attente maximal avant de renvoyer les enregistrements à lambda pour traitement. S’il n’est pas défini, le lambda revient immédiatement avec autant d’enregistrements disponibles sous le batchSize.

Modèle de rappel 👓

Encore une fois, si vous remarquez quelque chose, nous avons implémenté le modèle de rappel, dans lequel nous suspendons l’exécution de la fonction step après avoir envoyé le message à SQS et attendons que lambda publie des mises à jour sur la machine d’état une fois qu’elle a effectué le traitement, entraînant le succès ou l’échec.

Avec cette approche, nous pourrions comprendre où se déroule le travail réel et ne pas continuer à interroger les mises à jour encore et encore jusqu’à ce que le résultat soit prêt.

illustration du modèle de rappel ♦️

illustration du modèle de rappel en cas de réussite 💎

Modifications de la fonction lambda 🎼

Les modifications actuelles que nous avons apportées semblent suffisantes à première vue, mais elles échoueront, car le courant lambda de charge utile obtenu sera différent de l’ancienne invocation. En effet, alors que SQS est interrogé par lambda, le message réel sera à l’intérieur d’un objet qui contient un tableau d’enregistrements d’événements. Et à l’intérieur de l’enregistrement de l’événement, la propriété body donnera le message réel, lorsque nous l’avons transmis comme charge utile dans notre article précédent. Nous devons donc certainement ajuster un peu notre lambda comme suit.

{
    "Records": [
        {
            "messageId": "ff11de7f-795a-4c85-8612-9d376e01df0d",
            "receiptHandle": "**********",
            "body": "{\"Record\":{\"createdAt\":\"2022-04-18T07:51:06Z\",\"messageId\":\"538a8436-9987-1f65-478d-815c77f00d0f\",\"event\":{\"message\":\"A secret message\"}},\"MyTaskToken\":\"**********\"}",
            "attributes": {
                "ApproximateReceiveCount": "3",
                "SentTimestamp": "1650268267274",
                "SenderId": "AROAYLZFFS6HZ3ESZEEBH",
                "ApproximateFirstReceiveTimestamp": "1650268270274"
            },
            "messageAttributes": {},
            "md5OfBody": "edddae585d76eb14439f5804dcef24a4",
            "eventSource": "aws:sqs",
            "eventSourceARN": "**********",
            "awsRegion": "ap-south-1"
        }
    ]
}

Réécriture du code lambda 📝

Alors, écrivons la fonction lambda de l’enregistreur de messages comme indiqué ci-dessous en prenant note des changements de charge utile ci-dessus.

Pensons que cela nous donne une étape supplémentaire, cette étape nous aide plutôt à traiter l’écriture sur lambda dans un lot avec des enregistrements de messages pendant le processus d’interrogation.

import { PutItemInput } from "aws-sdk/clients/dynamodb";
import { DynamoDB, StepFunctions } from "aws-sdk";
const sfn = new StepFunctions({ apiVersion: "2016-11-23" });
exports.processor = async function (event: any) {
  const dynamo = new DynamoDB();
  let result: any | undefined = undefined;
  await Promise.all(
    event.Records.map(async (Record: any) => {

      const msg = JSON.parse(Record.body).Record;
      const crt_time: number = new Date(msg.createdAt).getTime();
      const putData: PutItemInput = {
        TableName: process.env.MESSAGES_TABLE_NAME || "",
        Item: {
          messageId: { S: msg.messageId },
          createdAt: { N: `${crt_time}` },
          event: { S: JSON.stringify(msg.event) },
        },
        ReturnConsumedCapacity: "TOTAL",
      }; 
      try {
        result = await dynamo.putItem(putData).promise();
      } catch (err) {
        const sendFailure: StepFunctions.SendTaskFailureInput = {
          error: JSON.stringify(err),
          cause: JSON.stringify({
            statusCode: 500,
            headers: { "Content-Type": "text/json" },
            putStatus: {
              messageId: msg.messageId,
              ProcessorResult: err,
            },
          }),
          taskToken: JSON.parse(Record.body).MyTaskToken,
        };
        console.log(sendFailure);
        await sfn.sendTaskFailure(sendFailure, function (err: any, data: any) {
          if (err) console.log(err, err.stack); 
          else console.log(data); 
        });
        return sendFailure;
      }
      const sendSuccess: StepFunctions.SendTaskSuccessInput = {
        output: JSON.stringify({
          statusCode: 200,
          headers: { "Content-Type": "text/json" },
          putStatus: {
            messageId: msg.messageId,
            ProcessorResult: result,
          },
        }),
        taskToken: JSON.parse(Record.body).MyTaskToken,
      };

      console.log(sendSuccess);

      await sfn
        .sendTaskSuccess(sendSuccess, function (err: any, data: any) {
          if (err) console.log(err, err.stack); 
          else console.log(data); 
        })
        .promise();

      return sendSuccess;
    })
  );
};

résumé des changements lambda 🍋

exports.processor = async function (event: any) {
  const dynamo = new DynamoDB();
  let result: any | undefined = undefined;
  await Promise.all(
    event.Records.map(async (Record: any) => {  
      const msg = JSON.parse(Record.body).Record;
      
    })
  );
};

Changements de TaskToken 🏆

Outre le taskToken a également été modifié, à l’intérieur du lambda de l’exemple précédent à la valeur ci-dessous.

Avant

taskToken: Record.MyTaskToken,

Après

taskToken: JSON.parse(Record.body).MyTaskToken,

Timeout de la fonction pas à pas 🌼

Notez que le délai d’attente défini pour la fonction pas à pas annulera et quittera l’exécution si elle attend plus de temps que la valeur de délai d’attente définie pour la fonction pas à pas.

b799d9d4-5c32-5f42-89bb-830315b023f7    INFO    TaskTimedOut: Task Timed Out: 'Provided task does not exist anymore'
    at Request.extractError (/var/runtime/node_modules/aws-sdk/lib/protocol/json.js:52:27)
    at Request.callListeners (/var/runtime/node_modules/aws-sdk/lib/sequential_executor.js:106:20)
    at Request.emit (/var/runtime/node_modules/aws-sdk/lib/sequential_executor.js:78:10)
    at Request.emit (/var/runtime/node_modules/aws-sdk/lib/request.js:686:14)
    at Request.transition (/var/runtime/node_modules/aws-sdk/lib/request.js:22:10)
    at AcceptorStateMachine.runTo (/var/runtime/node_modules/aws-sdk/lib/state_machine.js:14:12)
    at /var/runtime/node_modules/aws-sdk/lib/state_machine.js:26:10
    at Request.<anonymous> (/var/runtime/node_modules/aws-sdk/lib/request.js:38:9)
    at Request.<anonymous> (/var/runtime/node_modules/aws-sdk/lib/request.js:688:12)
    at Request.callListeners (/var/runtime/node_modules/aws-sdk/lib/sequential_executor.js:116:18) {
  code: 'TaskTimedOut',

Déploiement de nos changements 🐪

Pendant que vous déployez vos modifications, vous pouvez être invité à accepter la modification IAM pour accorder les privilèges nécessaires à sfn pour envoyer des messages à SQS, similaire à ce que nous avons explicitement accordé à lambda dans notre dernier article.

Exemple de journal d’exécution 🌰

Ainsi, nous avons atteint l’objectif de cet article de faire en sorte que la fonction step permette de placer les messages sous forme d’enregistrements dans une file d’attente, qui est ensuite interrogée par le lambda et traitée ultérieurement pour renvoyer l’état à la machine d’état. Cette architecture est donc beaucoup plus évolutive que la précédente avec l’invocation directe de la fonction lambda.

Nous ajouterons plus de connexions à notre pile et la rendrons plus utilisable dans les prochains articles en créant de nouvelles constructions, alors pensez à suivre et à vous abonner à ma newsletter.

⏭ Nous avons notre prochain article en serverless, consultez

🎉 Merci pour votre soutien ! 🙏

Ce serait formidable si vous aimez ☕ Achetez-moi un café, pour aider à stimuler mes efforts.

🔁 Message original sur 🔗 Dev Post

🔁 Republié à 🔗 dev à @aravindvcyber

🍬 AWS CDK 101 – 🔬 Ajout d’une file d’attente pour tamponner notre stepfunction appelant directement lambda#manuscrit #sans serveur #messagequeue #awscdk #aws https://t.co/Fm785cuPf8

— Aravind V (@Aravind_V7) 19 avril 2022