Jeżeli wszystko zadziałało, powinniście zobaczyć log z wywołania funkcji. Na pewno się domyslacie, że każda nowa wiadomość w kolejce uruchomi funkcję, a ta zapisze dane do logów. My chcemy osiągnąć to, aby funkcja przetwarzała wiadomość z określonym body tylko jeden raz. Oczywiście w ramach konkretnego okna czasowego, które domyslnie ustawione jest na 3600 sekund, czyli 1 godzinę.

Taką funkcjinalność można oprogramować samemu, można też sięgnąć po gotowe rozwiązania. My wykorzystamy AWS Lambda Powertools.

import logging
import os
from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer, IdempotencyConfig, idempotent_function)
logger = logging.getLogger()
logger.setLevel(logging.INFO)

TABLE_NAME = os.environ['IDEMPOTENCY_TABLE']

dynamodb = DynamoDBPersistenceLayer(table_name=TABLE_NAME)
config =  IdempotencyConfig(
    event_key_jmespath="body"
)

@idempotent_function(data_keyword_argument="message",config=config, persistence_store=dynamodb)
def message_consumer(message):
    logger.info(message)

def handler(event, context):
    for m in event['Records']:
        message_consumer(message=m)

Całość dostępna jest oczywiście w GitHub.

Działanie

Sprobujmy teraz przesłać do kolejki kilka ponumerowanych wiadomości:

aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test1"
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test2"
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test3"
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test4"
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test1"
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test2"

Zwróćcie uwagę, że ostatnie dwa polecenia wysyłają do kolejki zduplikowane wiadomości 1 i 2.

Sprawdźmy jeszcze raz logi zapisane przez Lambdę. Przypominam, że logi zapisywane są w metodzie message_consumer, nie powinnismy więc zobaczyć duplikatów.

Tak też się dzieje. Nasza metoda zadziałała dla każdego payloadu tylko jeden raz.

Sprawdźmy jeszcze jak wyglądają dane zapisane w tabeli DynamoDB:

Jak widać, mamy zapisane poszczególne wywołania, nie ma jednak danych. Spróbujmy więc coś z naszej metody zwrócić.
Zmieńmy wielkość liter na duze i zwróćmy je z metody

@idempotent_function(data_keyword_argument="message",config=config, persistence_store=dynamodb)
def message_consumer(message):
    logger.info(message)
    m = message['body'].upper()
    return m

a w handlerze zapiszmy zwróconą wartość do CloudWatcha

def handler(event, context):
    for m in event['Records']:
        ret = message_consumer(message=m)
        logger.info("Consumer returned: {0}".format(ret))

Zobaczny co się teraz wydarzy. Na początek wyslijmy kilka nowych wiadomości do kolejki:

aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test danych 1" 
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test danych 2" 
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test danych 3"

Zawartości logów w CloudWatch się pewnie takie, jakich się spodziewacie

Funkcja zostala uruchomiona trzy razy i za każdym razem pracowały obie metody.

Jak teraz wyglądają dane w DynamoDB? Otóż tak:

Oprócz samych wywołań mamy zapisane także zwrócone dane. Wyślijmy więc jeszcze raz pierwsze dwie wiadomości:

aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test danych 1" 
aws sqs send-message --queue-url <QUEUE-URL> --region eu-central-1 --message-body "Test danych 2"

i sprawdźmy logi

Jak widać, ponowne uruchomienie funkcji nie spowodowało zadziałania metody message_consumer, ale zostały zwrócone poprawne dane. O to chodziło. AWS Lambda i idempotentność? Załatwione.

Wyjątki

Ważne jest zrozumienie jak zachowają się AWS Power Tools w przypadku gdy wystąpią jakieś wyjątki.

Jeżeli wyjątek będzie miał miejsce w metodzie, która nie jest oznaczona dekoratorem idempotencji i została ona wykonana później to dane pozostaną zapisane i powtórzona wiadomość nie zostanie przetworzona.

Natomiast gdy nie przechwycimy wyjątku w metodzie objętej idempotentnością to dane zostaną skasowane. Jeżeli zależy nam na zachowaniu danych, to należy wyjącek przechwycić i zwrócić z metody poprawne dane.

Podsumowanie

Według dokumentacji AWS „Standard queues provide at-least-once delivery, which means that each message is delivered at least once”. Czyli mamy gwarancję, że  konsument otrzyma wiadomość ze standardowej kolejki CO NAJMNIEJ jeden raz. Co najmniej, czyli SQS może dostarczyć konkretną wiadomość więcej niż jeden raz.

Czasami nie będzie to miało dużego znaczenia, ale może to też być krytyczne dla naszego systemu. Możemy sami próbować poradzić sobie z tym problemem, możemy także wspomóc się działającym, uzywanym przez wiele aplikacji AWS Lambda Powertools, które pomoże w załatwieniu problemu AWS Lambdy i idempotentności.

Powertoolsy potrafią zresztą o wiele więcej. Nawet odnośnie implementacji idempotentności. Poszczegóły odsyłam do dokumentacji, ale mam nadzieję, że zainteresowałem tych z Was, którzy jeszcze tego rozwiązania nie używali.

Przypominam, że całość kodu dostępna jest na moim GitHubie.