CQRS en AWS: Sincronizando los Servicios de Command y Query con Amazon SQS

En la primera publicación de la serie que aborda las diferentes formas de implementar el estándar Command Query Responsibility Segregation (CQRS) en AWS, analizaré la forma más sencilla del estándar, que consiste en sincronizar los servicios de comandos y consultas a través del Amazon Simple Queue Service (Amazon SQS). Cuando se trabaja con aplicaciones que manejan modelos de datos complejos, es posible que este modelo no esté en el formato ideal para la lectura, debido a combinaciones complejas, por ejemplo, que pueden provocar consultas que tarden más de lo esperado o que el usuario esté dispuesto a esperar. En estos casos, se puede utilizar el estándar CQRS.

Por lo tanto, el estándar es relativamente simple: por un lado, tenemos un servicio que recibe scripts (comandos) y, por otro lado, tenemos un servicio que recibe lecturas (consultas), cada uno con una base de datos que facilita cada operación. El desafío de este estándar es: ¿cómo podemos transportar los datos, desde la fase de escritura hasta la lectura, de forma fiable? ¿Qué formas de comunicación se pueden usar en cada escenario? ¿Cuáles son las ventajas y desventajas de cada solución? Estas son las preguntas que se responderán en esta serie sobre el CQRS.

En esta serie de seis publicaciones de blog, explicaré la historia de este estándar, su motivación y cómo se puede implementar con diferentes servicios de AWS. En las cuatro primeras partes se mostrará la edición compatible con Amazon Aurora PostgreSQL-Compatible Edition como la base de datos del servicio de comandos y Amazon Elasticache for Redis como la base de datos del servicio de consultas. En esta primera parte, mostraré el estándar CQRS en su forma más simple. En la segunda parte, empezaremos a utilizar el estándar Transactional Outbox (en español, algo así como «bandeja de salida transaccional») para publicar eventos del servicio de comandos para alimentar el servicio de consultas, utilizando la técnica Polling Publisher. En las partes tercera y cuarta, también utilizaremos el estándar Transactional Outbox, pero utilizando la técnica Transaction Log Tailing (en español, algo parecido a «leer el final o el final del registro de transacciones»).

En la quinta parte, demostraré cómo podemos implementar CQRS con Amazon DynamoDB como base de datos del servicio de comandos y Aurora como base de datos del servicio de consultas. Por último, presentaré una conclusión y una comparación de todas las técnicas en la sexta parte.

Introducción

En 1988, Bertrand Meyer introdujo la idea de Command Query Separation, o CQS, en su libro Object-Oriented Software Construction, para aplicarla al software orientado a objetos. La idea fundamental es que los métodos de un objeto deben dividirse en dos categorías: métodos que solo devuelven valor y no cambian el estado, y métodos que solo cambian el estado y no devuelven ningún valor. Cada método debe pertenecer a una de estas categorías, pero no a ambas. La ventaja es que tenemos una separación clara de los métodos, y los métodos que realizan consultas se pueden invocar de forma idempotente. En la práctica, esta regla se aplica a la mayoría de los casos, pero no a todos. Utilizando los métodos de estructura de datos de la pila como ejemplo, el método pop() devuelve el objeto que está en la parte superior de la pila y también lo elimina, lo que lo convierte en un método que recupera un valor y cambia el estado.

Inspirado por la CQS, Greg Young presentó por primera vez la idea del estándar Command Query Responsibility Segregation (CQRS) en una conferencia que presentó en la QCon de San Francisco en 2006 (lamentablemente, esta es la única edición de la QCon San Francisco que no tiene registros en línea). Sin embargo, la idea no empezó a afianzarse hasta después de una segunda conferencia impartida por él, en la QCon de San Francisco, el 8 de noviembre de 2007. La idea general es que nuestra solución se pueda estructurar de manera que haya dos servicios, uno responsable de las operaciones de escritura y el otro de las operaciones de lectura.

Hay casos en los que la base de datos que se utiliza no favorece las operaciones de lectura, por ejemplo, una base de datos relacional que necesita realizar varias uniones y otras agregaciones para devolver la información necesaria, lo que introduce una latencia no deseada en el tiempo de respuesta de la aplicación. En estos casos, podemos crear un servicio que reciba operaciones de escritura, almacene datos en su base de datos y publique los eventos que captura el servicio responsable de las operaciones de consulta/lectura, que, a su vez, llena su base de datos de forma asincrónica con la nueva información en un formato que facilite la recuperación de esos datos.

Los beneficios de tener dos servicios separados para la escritura y la consulta son los siguientes:

• Escalabilidad independiente. Como los servicios existen de forma independiente, es posible escalarlos según sea necesario. Si la aplicación recibe más solicitudes de cambios de estado, es posible que tengamos más servidores o contenedores que las gestionen. Si la base de datos que se utiliza es una base de datos relacional, como Aurora, tendremos que encontrar el tamaño de instancia de escritura correcto y, para ello, podemos usar métricas como las IOPS, la utilización de la CPU y el rendimiento, además de analizar cómo se escribe en la base de datos. Según el patrón de escritura, por ejemplo, si hay picos durante un período de tiempo determinado y períodos de escritura más largos que son muy bajos, es posible que se utilice la tecnología serverless, que está disponible para Aurora.
• Base de datos para fines especiales. No siempre el modelo de datos o la base de datos que se utilizan pueden servirnos cuando lo necesitamos. Puede ser que el modelo de datos refleje un caso de uso real que, por ejemplo, no favorezca la lectura o que la base de datos utilizada no permita responder fácilmente a preguntas específicas. En estos casos, podemos usar la base de datos más adecuada para escribir y consultar las operaciones por separado.
• Tecnologías y ciclo de vida independientes. Habrá dos servicios independientes que se ocuparán de los comandos y las consultas, y cada servicio se podrá crear con la tecnología que mejor se adapte a cada necesidad. Por ejemplo, podemos tener el servicio de comandos implementado en Java y el servicio de consultas implementado en Python. Del mismo modo, cada servicio puede tener su propio entorno de ejecución, como el servicio de comandos que se ejecuta en AWS Lambda y el servicio de consultas que se ejecuta en Amazon Elastic Kubernetes Service (Amazon EKS). Cada servicio también puede tener su propio ciclo de vida y proceso de implementación.

Tener dos bases de datos separadas y dos servicios separados que atienden diferentes solicitudes introduce más complejidad en un sistema, lo que solo vale la pena cuando necesitamos diferentes escalabilidades o diferentes bases de datos para cumplir con los comandos y las consultas. Si la escalabilidad es el único problema, podemos tener una sola base de datos para diferentes servicios, que se pueden escalar de forma independiente. En ese caso, podríamos simplificar la base de datos para que pudiera gestionar la escritura y la lectura por separado, como ocurre con las réplicas de lectura.

Si la base de datos es un Amazon Relational Database Service (Amazon RDS), el servicio que atiende las consultas puede leer réplicas de lectura. Si la base de datos es una Aurora, también puede utilizar endpoints de lectura, o incluso endpoints personalizados, que le permiten agrupar las réplicas de lectura para diferentes propósitos, como tener dos réplicas de lectura para un informe A y otras dos réplicas de lectura para un informe B. Si el único problema es una base de datos diferente para fines de lectura (como en el caso de que desee mantener la información precalculada en un caché de segundo nivel, como Redis), solo puede tener un servicio que acceda a diferentes bases de datos para cada necesidad, introduciendo, por ejemplo, una cola que recibe un evento después de una inserción y que es leída por un componente que actualiza la base de datos que responde a las consultas.

Si necesitamos dos servicios independientes y dos bases de datos diferentes para ejecutar los comandos y las consultas, hay algunas opciones que podemos usar para implementar CQRS en AWS. Echemos un vistazo a la forma más simple.

Caso de uso: Amazon Aurora PostgreSQL-Compatible Edition como base de datos del servicio de comandos y Amazon Elasticache for Redis como base de datos del servicio de consultas, con Amazon SQS para la sincronización

En el primer caso que analizaré en esta serie de publicaciones de blog, tenemos a Aurora como la base de datos del servicio que recibe los comandos y a Redis como la base de datos del servicio que recibe las consultas. La base de datos Aurora contendrá información relacionada con los clientes, los productos y los pedidos, y Redis contendrá información precalculada relacionada con los clientes. Para probar la arquitectura, utilizaremos dos terminales, uno para guardar los pedidos y otro para recuperar la información relacionada con los clientes.

Resumen de la Solución

La opción más sencilla es publicar un evento en una cola de SQS después de actualizar la base de datos del servicio de comandos. A partir de ahí, un componente que pueda realizar cálculos (como una función de Lambda) recuperará los eventos y la información precalculada relacionada con el cliente se actualizará en Redis. Otra opción sería colocar un evento en un tópico de Amazon SNS en lugar de en una cola de SQS, de modo que pudiera notificar a varios destinatarios acerca de la nueva información introducida en Aurora.

Es importante tener en cuenta que, si bien este ejemplo utiliza Amazon API Gateway, cualquier componente computacional puede implementar el reenvío de mensajes a una función, comando y consulta de Lambda. Podría ser, por ejemplo, un contenedor en Amazon EKS como servicio de comandos y un contenedor en Amazon ECS como servicio de consultas. El aspecto más importante de esta implementación es hacer que el servicio de comandos publique los eventos que alimenten la base de datos del servicio de consultas.

Esta primera solución también considera una dead-letter queue. Cuando una función de Lambda procesa correctamente un mensaje, elimina automáticamente de la cola el mensaje procesado. Si la función Lambda no puede procesar el mensaje debido, por ejemplo, a una excepción, volverá a estar visible en la cola y el número de veces que se recibió el mensaje en la cola aumentará en 1. Cuando este contador alcanza el número máximo de recepciones, configurado cuando se configuró la dead-letter queue en la cola principal, este mensaje se mueve a la dead-letter queue, para que pueda procesarse por otros medios, como activar una alarma, enviar un correo electrónico, insertar un registro en otra base de datos, etc. Para simplificar este ejemplo, se han omitido estos otros mecanismos.

Otra posible solución sería hacer que API Gateway colocara un mensaje directamente en una cola de SQS en lugar de invocar una función de Lambda. En la cola, podríamos hacer que otro componente (como una función Lambda) recupere el mensaje. En este caso, si se produjera una excepción en la función Lambda, el mensaje no se eliminaría de la cola hasta que se alcanzara el número de recibos. Tendríamos que controlar el tratamiento de estos mensajes con estándares de resiliencia, como retry, circuit breaker, exponential backoff y también una dead-letter queue. Si el caso práctico que se está implementando requiere un procesamiento sincrónico, esa idea no sería una opción.

Figura 1. Arquitectura propuesta, con Amazon Aurora PostgreSQL Edition como base de datos del servicio de comandos y Amazon Elasticache para Redis como base de datos del servicio de consultas.

El punto fuerte de esta arquitectura es su sencillez. En pocas palabras, utilizamos una cola para enviar datos desde el lado del servicio de comandos al lado del servicio de consultas. Cuando recibimos un pedido en la función OrderReceiverLambda y lo guardamos en Aurora, publicamos un evento en una cola de SQS. Incluso podríamos preparar la información precalculada relacionada con los clientes en esta función de Lambda, pero estaríamos haciendo más de lo que deberíamos, en contra del Single Responsibility Principle. Además, si por algún motivo no pudiéramos actualizar el caché de Redis (podría ser porque se estuviera actualizando a una nueva versión, por ejemplo), perderíamos esa información. Con una cola, podemos procesar este evento más adelante, incluso si el caché de Redis no está disponible temporalmente.

Nuestro objetivo aquí es llevar los cambios de Aurora a Redis de manera confiable. Al igual que DynamoDB, Amazon DocumentDB o Amazon Neptune, Aurora también tiene una forma de transmisiones, que es Database Activity Streams. El detalle es que esta función de Aurora tiene más fines de auditoría (todo lo que ocurre en la base de datos se registra en formato de registro), por lo que incluso sería posible utilizarla para publicar eventos, pero sería necesario adaptar el formato de registro a un formato más fácil de usar. Además, esta funcionalidad requiere el uso de Amazon Kinesis Data Streams, que no queremos usar todavía. En este escenario, lo más importante es que la transacción (dado que es una base de datos relacional que admite transacciones ACID) esté separada de la publicación del evento.

Hay puntos a tener en cuenta en esta solución. La primera es que el commit de la transacción en el OrderReceiverLambda se separa de la publicación del evento. Consideremos un caso en el que se confirma una transacción y, a continuación, se envía un evento a una cola de SQS. Si por alguna razón hay algún problema al colocar el evento en la cola de SQS, las dos bases de datos no estarán sincronizadas. Si la publicación del evento en la cola de SQS se produce dentro de la transacción, lo que puede ocurrir es que el evento se publique en la cola de SQS y la base de datos de Redis se actualice, pero es posible que la transacción, por algún motivo, no se confirme y las dos bases de datos tampoco estén sincronizadas. En esa misma situación, si hay algún problema al publicar el evento en la cola de SQS, la transacción se revertirá (rolledback) no por un problema con los datos en sí, sino porque hubo un problema con un componente externo.

Esta solución considera una cola SQS standard, lo que significa que podemos recibir casi un número ilimitado de mensajes por segundo. En este caso, no necesitamos que los mensajes estén ordenados, por lo que podemos usar una cola estándar en lugar de una cola FIFO. Si usáramos una cola FIFO, estaríamos limitados a un máximo de 300 mensajes por segundo. Teniendo en cuenta el ejemplo anterior, el mensaje que OrderReceiverLambda colocará en la cola de SQS puede tener el siguiente aspecto:

select_statement = "select name, email from public.client where id = %s"
# Recupera os dados do cliente que está efetuando o pedido
cur.execute(select_statement, str(event["id_client"]))
client = cur.fetchone()
order_event = {
    "messageId": str(uuid.uuid4()),
	"id_client": event["id_client"],
	"name": client[0],
	"email": client[1],
	"order_total": order_total,
	"event_date": now
}

Si bien las colas FIFO tienen una semántica de entrega exactly once, las colas standard tienen una semántica de entrega at-least once, por lo que debemos asegurarnos de que el mismo mensaje no se procese dos veces. Una forma consiste en añadir un campo de identificación al mensaje que se va a colocar en la cola (en el objeto «order_event» de arriba, el campo «messageId») y verificar ese identificador antes de procesarlo. Y como ya estamos usando Redis, podemos añadir una entrada a nuestro caché, cuya clave será el identificador del mensaje y el valor será el identificador de ejecución de la función Lambda (context.aws_request_id), con un tiempo de caducidad de cinco minutos. Además, antes de agregar o actualizar el valor de la clave del cliente, verificamos que el valor de la clave correspondiente al identificador del mensaje sea el mismo que el identificador de ejecución de la función de Lambda. Si es así, procesamos el mensaje; de lo contrario, otro entorno de ejecución ya habrá gestionado el mensaje, por lo que simplemente podemos ignorarlo, ya que será un mensaje duplicado.

Aquí hay un pequeño detalle: dependiendo de cómo realicemos este control, un mensaje determinado aún se puede procesar dos veces. Imagínese el siguiente escenario: cuando recibimos un mensaje de la cola de SQS en la función Lambda, agregamos el identificador del mensaje como clave en Redis y, a continuación, verificamos que el valor de la clave es igual al identificador de ejecución de la función de Lambda en ese contexto. Una vez que la verificación se evalúa como verdadera, otro entorno de ejecución de Lambda recupera el mismo mensaje, sobrescribe la clave con el identificador del mensaje y comprueba que el valor de la clave es el mismo que el identificador de ejecución de la función de Lambda en ese otro contexto. Una vez que la verificación se evalúa como verdadera, el mensaje se procesa por duplicado. Para evitar este escenario, lo primero que hay que hacer es introducir la clave correspondiente al identificador del mensaje en Redis mediante el parámetro «nx» (Not eXists, o «no existe»), lo que garantiza que la clave se introduzca solo si aún no existe. De esta forma, solo podremos procesar los mensajes duplicados una vez.

El código de la función de Lambda que actualiza Redis con los datos del cliente que hizo el pedido puede tener el siguiente aspecto:

from redis.commands.json.path import Path

import json
import os
import redis

redis_host = os.environ['REDIS_HOST']
redis_port = os.environ['REDIS_PORT']
r = redis.Redis(host=redis_host, port=redis_port)

def lambda_handler(event, context):
    body = json.loads(event['Records'][0]['body'])
    r.set(body['messageId'], context.aws_request_id, nx=True, ex=300)

    if r.get(body['messageId']).decode() == context.aws_request_id:
        if not r.exists(body['id_client']):
			json_value = {'name': body['name'],
                          'email': body['email'],
                          'total': body['order_total'],
                          'last_purchase': body['event_date']}
            r.json().set(body['id_client'], Path.root_path(), json_value)
        else:
            current_value = r.json().get(body['id_client'])
            json_value = {'name': body['name'],
                          'email': body['email'],
                          'total': body['order_total'] + current_value['total'],
                          'last_purchase': body['event_date']}
            r.json().set(body['id_client'], Path.root_path(), json_value)

Otra cosa a tener en cuenta es que no podemos volver a ejecutar los eventos, lo que podría ser interesante si quisiéramos reproducir un bug o cargar otra base de datos, por ejemplo. Para resolver estos problemas, el estándar Transactional Outbox puede ayudar. ¡Pero ese es el enfoque que exploraremos en las próximas tres publicaciones de esta serie!

Ejecutando el Ejemplo

Para ejecutar el ejemplo, los lectores deben tener una cuenta de AWS y un usuario con permisos de administrador. A continuación, basta con ejecutar el paso a paso que se proporciona en el repositorio de código para esta serie de entradas de blog sobre CQRS, en AWS Samples, alojadas en Github. Al realizar el proceso paso a paso, los lectores dispondrán de la infraestructura que se presenta aquí en sus propias cuentas.

El ejemplo contiene dos endpoints, uno para recibir información relacionada con los pedidos (que representa nuestro servicio de comando) y el otro para recuperar información relacionada con los clientes (que representa nuestro servicio de consultas). Para comprobar que todo ha funcionado correctamente, ve a la API Gateway y, en la lista de API, introduce la API «OrdersAPI» y, a continuación, Stages. Solo habrá una stage llamada «prod». Recupere el valor del campo Invoke URL y añada «/orders». Este es el endpoint que recibe la información relacionada con los pedidos.

Hagamos una solicitud POST a ese endpoint. Podemos usar cualquier herramienta para realizar solicitudes, como cURL o Postman. Como este endpoint está protegido, también necesitamos añadir basic authentication. Si utilizas Postman, tendrás que recuperar el nombre de usuario y la contraseña generados al crear la infraestructura. En la puerta de enlace de API, vaya a «API Keys» y copie el valor de la columna «API Key» de «admin_key». Este valor contiene el nombre de usuario y la contraseña separados por el carácter «:», pero está codificado en Base64. Decodifique el valor con una herramienta en línea o con el comando «base64» de Linux. El nombre de usuario está a la izquierda del carácter «:» y la contraseña, a la derecha. Añada una «Authorization» del tipo «Basic Auth» y rellene los campos «Username» y «Password» con los valores recuperados. Añade también un header «Content-Type», con el valor «application/json».

El payload para realizar solicitudes a este endpoint es el siguiente:

{
    "id_client": 1,
    "products": [{
        "id_product": 1,
        "quantity": 1
    }, {
        "id_product": 2,
        "quantity": 3
    }]
}

Esto representa un pedido realizado por el cliente con el identificador 1 y que contiene productos con los identificadores 1 y 2. El total de ese pedido es de $3000. Toda esta información se almacenará en Aurora. Al realizar esta solicitud POST, si todo funcionó según lo esperado, debería ver el siguiente resultado:

{
    "statusCode": 200,
    "body": "Order created successfully!"
}

Ahora, verifiquemos que la información relacionada con el cliente se haya enviado a Redis. Al endpoint de API Gateway, que se recuperó anteriormente, añada «/clients/1». Este es el endpoint que recupera la información relacionada con el cliente. Hagamos una solicitud GET para ese endpoint. Al igual que hicimos con el endpoint «/orders», necesitamos añadir basic authentication Siga los pasos explicados anteriormente y realice la solicitud GET. Si todo ha funcionado según lo esperado, verás un resultado similar al siguiente:

{
    "name": "Bob",
    "email": "bob@anemailprovider.com",
    "total": 3000.0,
    "last_purchase": 1700836837
}

Esto significa que pudimos alimentar correctamente a Redis con información lista para ser leída, mientras la misma información está en Aurora, en otro formato.

Limpiando los Recursos

Para borrar la infraestructura creada, en una terminal, en el mismo directorio donde se creó la infraestructura, basta con ejecutar el comando «cdk destroy» y confirmar. La eliminación tarda aproximadamente 10 minutos.

Conclusión

En esta entrada del blog, analizé cómo podemos implementar el estándar CQRS con los servicios de AWS, en su forma más sencilla. La ventaja de este estándar es que podemos tener diferentes servicios que atiendan comandos y consultas, así como diferentes bases de datos, cada una con diferentes propósitos.

Los servicios de comandos y consultas son independientes y, por lo tanto, se pueden escalar por separado, tienen bases de datos diferentes y también reglas diferentes, como la seguridad. Aunque en esta entrada del blog presenté Amazon Aurora como la base de datos para el servicio de comandos y Amazon Elasticache para Redis como la base de datos para el servicio de consultas, la idea es que podamos tener la base de datos que mejor se adapte a cada propósito.

Esta es la forma más simple de CQRS y tiene sus ventajas y desventajas. La ventaja de esta solución es su simplicidad y su desventaja es que los eventos se pueden perder y, además, no se pueden volver a procesar. En la próxima entrada de esta serie, analizaré una técnica que utiliza el estándar Transactional Outbox para enviar información de forma fiable desde el servicio de comandos al servicio de consultas, más específicamente, mediante la técnica Polling Publisher.

Este contenido és una traduccíon del blog original en Portugués (enlace acá)

Acerca del Autor
	Roberto Perillo es un arquitecto de soluciones empresariales en AWS Brasil, especializado en sistemas serverless, que presta servicios a clientes del sector financiero y ha estado en la industria del software desde 2001. Trabajó durante casi 20 años como arquitecto de software y desarrollador Java antes de unirse a AWS en 2022. Es licenciado en Ciencia de la Computación, tiene una especialización en Ingeniería de Software y un máster también en Ciencia de la Computación. Un aprendiz eterno. En su tiempo libre, le gusta estudiar, tocar la guitarra y también ¡jugar a los bolos y al fútbol de mesa con su hijo, Lorenzo!
Acerca de los Colaboradores
	Luiz Santos trabaja actualmente como Technical Account Manager (TAM) en AWS y es un entusiasta de la tecnología, siempre busca nuevos conocimientos y tiene una mayor experiencia en el desarrollo de software, el análisis de datos, la seguridad, la tecnología serverless y DevOps. Anteriormente, tenía experiencia como arquitecto de soluciones de AWS y SDE.
	Maria Mendes es arquitecta de soluciones desde agosto de 2022. Anteriormente, tenía experiencia en el área de ingeniería de datos, trabajando con Python, SQL y arquitectura orientada a eventos. En AWS, Maria forma parte de la comunidad técnica centrada en los servicios de DevOps de AWS.
Acerca de los Revisores
	Gerson Itiro Hidaka trabaja actualmente como arquitecto de soluciones empresariales en AWS y presta servicios a clientes financieros en Brasil. Entusiasta de tecnologías como el Internet de las cosas (IoT), los drones, DevOps y especialista en tecnologías como la virtualización, la tecnología serverless, los contenedores y Kubernetes. Lleva más de 26 años trabajando con soluciones de TI y tiene experiencia en numerosos proyectos de optimización de infraestructuras, redes, migración, recuperación ante desastres y DevOps incluidos en su cartera.
	Peterson Larentis es un arquitecto sénior especializado en soluciones serverless en AWS, profesor y coordinador de cursos de MBA, emprendedor y orador internacional. En AWS, ayuda a los clientes más importantes de América Latina a crear arquitecturas seguras, confiables, elásticas y escalables con excelencia operativa utilizando las mejores prácticas ágiles de ingeniería de software. Fue responsable de las áreas de DevOps en AWS Summit Brazil durante dos ediciones y es el principal arquitecto de la comunidad serverless de AWS en América Latina. Tiene un máster en marketing digital de la Fundación Getúlio Vargas y seis certificaciones de AWS.
	Gonzalo Vásquez es Senior Solutions Architect de AWS Chile para clientes de los segmentos Independent software vendor (ISV) y Digital Native Business (DNB) de Argentina, Paraguay y Uruguay. Antes de sumarse a AWS, se desempeñó como desarrollador de software, arquitecto de sistemas, gerente de investigación y desarrollo y CTO en compañías basadas en Chile.