AWS - Comunicar dos lambdas usando SQS (Queue service)
Enviar un mensaje de una lambda a una SQS que hace de trigger de otra
El contexto es el siguiente:
Tengo dos lambdas que se deben ejecutar de forma sincrona de la siguiente forma:
- La lambda uno hace un proceso de ETL y devuelve un código de cuenta.
- Ese código de cuenta es la entrada de otra ETL (otra lambda) que depende de este dato.
El código de cuenta accountId se transfiere a traves de un un recurso intermedio que sirve al mismo tiempo de trigger de la segunda lambda.
Este trigger es un SQS que es un sistema de colas.
A continuación explico el código hecho en python de las dos lambdas:
- lanbda_que_envia.py
- lambda_que_recibe.py
lambda_que_envia.py
def lambda_handler(event, context):
logger.info("start")
sqs_url = f"https://sqs.{get_config('region')}.amazonaws.com/{get_config('account')}/sqs-trigger-lambda-que-recibe"
sqs_client = boto3.client(
'sqs',
region_name=get_config("region")
)
response = sqs_client.send_message_batch(
QueueUrl=sqs_url,
Entries=[{
"Id": "1",
"MessageBody": json.dumps({
"accountId": "3456"
}),
}]
)
logger.info("response sqs")
pprint(response)
logger.info("end")
Del código anterior podemos esperar que el objeto response, que es de tipo diccionario, tenga estos formatos:
Caso "success"
{
'ResponseMetadata': {
'HTTPHeaders': {
'content-length': '467',
'content-type': 'text/xml',
'date': 'Thu, 22 Apr 2021 12:20:03 GMT',
'x-amzn-requestid': '4da969ec-10d8-51f8-8263-31177ceb3503'
},
'HTTPStatusCode': 200,
'RequestId': '4da969ec-10d8-51f8-8263-31177ceb3503',
'RetryAttempts': 0
},
'Successful': [
{
'Id': '1',
'MD5OfMessageBody': '30840356415d359f33cec2ad1682bd71',
'MessageId': 'f60ac369-85fa-44a4-96de-fa5c4da0bee3'
}
]
}
Caso "error"
{
"errorMessage": "An error occurred (AWS.SimpleQueueService.NonExistentQueue) when calling
the SendMessageBatch operation: The specified queue does not exist for
his wsdl version.",
"errorType": "QueueDoesNotExist",
"stackTrace": [
[
"/var/task/lambda_function.py",
29,
"lambda_handler",
"\"accountId\": account_id"
],
[
"/var/runtime/botocore/client.py",
357,
"_api_call",
"return self._make_api_call(operation_name, kwargs)"
],
[
"/var/runtime/botocore/client.py",
676,
"_make_api_call",
"raise error_class(parsed_response, operation_name)"
]
]
}
Con el código anterior se ha enviado el mensaje a la cola que no es más que un json en string.
Cuando la SQS recibe el mensaje lo inyecta en lambda_que_recibe.py lambda_handler como evento
lambda_que_recibe.py
def lambda_handler(event, context):
logger.info("start recibe")
logger.info("EVENT received")
pprint(event)
logger.info("end EVENT")
account_id = get_account_id_from_event(event)
if not account_id:
error = "Missing accountId"
logger.error(error)
return {"error": error}
logger.info(f"end recibe account-id: {account_id}")
Formato del evento recibido
{
'Records': [
{
'attributes': {
'ApproximateFirstReceiveTimestamp': '1619094828869',
'ApproximateReceiveCount': '1',
'SenderId': 'AROATKX3FEXQZPZAJZR3B:lambda-que-envia',
'SentTimestamp': '1619094828863'
},
'awsRegion': 'us-east-2',
'body': '{"accountId": "3456"}',
'eventSource': 'aws:sqs',
'eventSourceARN': 'arn:aws:sqs:us-east-2:aaaacccc:sqs-lambda-que-recibe',
'md5OfBody': '30840356415d359f33cec2ad1682bd71',
'messageAttributes': {},
'messageId': '0514edc1-2048-4b12-8cc0-d6938d07101c',
'receiptHandle': 'AQEBGxDrjg2DP0ZlWNYqU8BPJw...rUypH+3eCaH52+ie/xzg=='
}
]
}
El código fuente completo lo dejo en mi Github
Autor: Eduardo A. F.
Publicado: 22-04-2021 22:17
Actualizado: 22-04-2021 22:20