Uso de Lambda con Amazon MQ

nota

Si desea enviar datos a un destino que no sea una función de Lambda o enriquecer los datos antes de enviarlos, consulte Amazon EventBridge Pipes (Canalizaciones de Amazon EventBridge).

Amazon MQ es un servicio de agente de mensajes administrado para Apache ActiveMQ y RabbitMQ. Un agente de mensajes permite que las aplicaciones de software y los componentes se comuniquen mediante varios lenguajes de programación, sistemas operativos y protocolos de mensajería formales a través de destinos de eventos de tema o de cola.

Amazon MQ también puede administrar instancias de Amazon Elastic Compute Cloud (Amazon EC2) en su nombre instalando agentes de ActiveMQ o RabbitMQ proporcionando diferentes topologías de red y otras necesidades de infraestructura.

Puede utilizar una función de Lambda para procesar registros de su agente de mensajes de Amazon MQ. Lambda invoca su función a través de una asignación de orígenes de eventos, un recurso de Lambda que lee los mensajes de su agente e invoca la función sincrónicamente.

aviso

Las asignaciones de orígenes de eventos de Lambda procesan cada evento al menos una vez y puede producirse un procesamiento duplicado de lotes. Para evitar posibles problemas relacionados con la duplicación de eventos, le recomendamos encarecidamente que haga que el código de la función sea idempotente. Para obtener más información, consulte ¿Cómo puedo hacer que mi función de Lambda sea idempotente? en el Centro de conocimientos de AWS.

El mapeo de fuente de eventos de Amazon MQ tiene las siguientes restricciones de configuración:

• Simultaneidad: las funciones de Lambda que utilizan una asignación de orígenes de eventos de Amazon MQ tienen una configuración de simultaneidad máxima predeterminada. Para ActiveMQ, el servicio Lambda limita la cantidad de entornos de ejecución simultánea a cinco. Para RabbitMQ, la cantidad de entornos de ejecución simultánea está limitada a 1. Incluso si cambia la configuración de simultaneidad reservada o aprovisionada de la función, el servicio Lambda no ofrecerá más entornos de ejecución disponibles. Para solicitar un aumento de la simultaneidad máxima predeterminada, póngase en contacto con AWS Support.

• Cuentas cruzadas: Lambda no admite el procesamiento de cuentas cruzadas. No se puede utilizar Lambda para procesar registros de un agente de mensajes de Amazon MQ que se encuentra en una Cuenta de AWS diferente.

• Autenticación: para ActiveMQm solo se admite ActiveMQ SimpleAuthenticationPlugin. Para RabbitMQ, solo se admite el mecanismo de autenticación PLAIN. Los usuarios deben utilizar AWS Secrets Manager para administrar sus credenciales. Para obtener más información acerca de la autenticación de ActiveMQ, consulte Integración de agentes de ActiveMQ con LDAP en la Guía para desarrolladores de Amazon MQ.

• Cuota de conexión: los agentes de cuota de conexión tienen un número máximo de conexiones permitidas por protocolo de nivel de cable. Esta cuota se basa en el tipo de instancia del broker. Para obtener más información, consulte la sección de Agentes de Cuotas en Amazon MQ en la Guía para desarrolladores de Amazon MQ.

• Conectividad: puede crear agentes en la nube privada virtual (VPC) pública o privada. En el caso de las VPC privadas, su función de Lambda necesita acceso a la VPC para recibir mensajes. Para obtener más información, consulte Configuración de red más adelante en este tema.

• Destinos de eventos: solo se admiten los destinos de cola. Sin embargo, puede utilizar un tema virtual, que se comporta como un tema internamente mientras interactúa con Lambda como una cola. Para obtener más información, consulte Destinos virtuales en el sitio web de Apache ActiveMQ y Virtual Hosts en el sitio web RabbitMQ.

• Topología de red: para ActiveMQ, solo se admite una instancia única o agente en espera por asignación de orígenes de eventos. Para RabbitMQ, solo se admite una implementación de agente o clúster de una instancia única por asignación de orígenes de eventos. Los agentes de instancia única requieren un punto de conexión de conmutación por error. Para obtener más información acerca de estos modos de implementación de agente, consulte Arquitectura de agente Active MQ y Arquitectura de agente de Rabbit MQ en la Guía para desarrolladores de Amazon MQ.

• Protocolos: los protocolos compatibles dependen del tipo de integración de Amazon MQ.

  • Para las integraciones de ActiveMQ, Lambda consume mensajes mediante el protocolo OpenWire/Servicio de mensajes de Java (JMS). No se admiten otros protocolos para consumir mensajes. Dentro del protocolo JMS, solo TextMessage y BytesMessage son compatibles. Lambda también admite las propiedades personalizadas de JMS. Para obtener más información acerca del protocolo OpenWire, consulte OpenWire en el sitio web de Apache ActiveMQ.

  • Para las integraciones de RabbitMQ, Lambda consume mensajes utilizando el protocolo AMQP 0-9-1. No se admiten otros protocolos para consumir mensajes. Para obtener más información acerca de la implementación de RabbitMQ del protocolo AMQP 0-9-1, consulte la Guía de referencia completa de AMQP 0-9-1 en el sitio web de RabbitMQ.

Lambda admite automáticamente las últimas versiones de ActiveMQ y RabbitMQ que admite Amazon MQ. Para obtener las últimas versiones compatibles, consulte Notas de la versión de Amazon MQ en la Guía para desarrolladores de Amazon MQ.

nota

De forma predeterminada, Amazon MQ tiene una ventana de mantenimiento semanal para los agentes. Durante esa ventana de tiempo, los corredores no están disponibles. Para los agentes sin espera, Lambda no puede procesar ningún mensaje durante esa ventana.

Grupo de consumidores de Lambda

Para interactuar con Amazon MQ, Lambda crea un grupo de consumidores que puede leer de sus agentes de Amazon MQ. El grupo de consumidores se crea con el mismo ID que el UUID de asignación de orígenes de eventos.

Para los orígenes de eventos de Amazon MQ, Lambda agrupa los registros y los envía a su función en una sola carga. Para controlar el comportamiento, puede configurar el plazo de procesamiento por lotes y el tamaño del lote. Lambda extrae mensajes hasta que procesa el tamaño de carga máximo de 6 MB, el plazo de procesamiento por lotes vence o el número de registros alcanza el tamaño completo del lote. Para obtener más información, consulte Comportamiento de procesamiento por lotes.

El grupo de consumidores recupera los mensajes como un BLOB de bytes, los codifica con base64 en una sola carga JSON y luego invoca la función. Si su función devuelve un error para cualquiera de los mensajes de un lote, Lambda reintenta todo el lote de mensajes hasta que el procesamiento sea correcto o los mensajes caduquen.

nota

Si bien las funciones de Lambda suelen tener un límite de tiempo de espera máximo de 15 minutos, las asignaciones de orígenes de eventos para Amazon MSK, Apache Kafka autoadministrado, Amazon DocumentDB y Amazon MQ para ActiveMQ y RabbitMQ solo admiten funciones con límites de tiempo de espera máximos de 14 minutos. Esta restricción garantiza que la asignación de orígenes de eventos pueda gestionar correctamente los errores y reintentos de las funciones.

Puede supervisar el uso de concurrencia de una función determinada utilizando la métrica ConcurrentExecutions en Amazon CloudWatch. Para obtener más información acerca de la simultaneidad, consulte Configuración de la simultaneidad reservada.

ejemplo Eventos de registro de Amazon MQ

ActiveMQ

{
   "eventSource": "aws:mq",
   "eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:test:b-9bcfa592-423a-4942-879d-eb284b418fc8",
   "messages": [
      { 
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1", 
        "messageType": "jms/text-message",
        "deliveryMode": 1,
        "replyTo": null,
        "type": null,
        "expiration": "60000",
        "priority": 1,
        "correlationId": "myJMSCoID",
        "redelivered": false,
        "destination": { 
          "physicalName": "testQueue" 
        },
        "data":"QUJDOkFBQUE=",
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958, 
        "brokerOutTime": 1598827811959, 
        "properties": {
          "index": "1",
          "doAlarm": "false",
          "myCustomProperty": "value"
        }
      },
      { 
        "messageID": "ID:b-9bcfa592-423a-4942-879d-eb284b418fc8-1.mq.us-west-2.amazonaws.com-37557-1234520418293-4:1:1:1:1",
        "messageType": "jms/bytes-message",
        "deliveryMode": 1,
        "replyTo": null,
        "type": null,
        "expiration": "60000",
        "priority": 2,
        "correlationId": "myJMSCoID1",
        "redelivered": false,
        "destination": { 
          "physicalName": "testQueue" 
        },
        "data":"LQaGQ82S48k=",
        "timestamp": 1598827811958,
        "brokerInTime": 1598827811958, 
        "brokerOutTime": 1598827811959, 
        "properties": {
          "index": "1",
          "doAlarm": "false",
          "myCustomProperty": "value"
        }
      }
   ]
}

RabbitMQ

{
  "eventSource": "aws:rmq",
  "eventSourceArn": "arn:aws:mq:us-west-2:111122223333:broker:pizzaBroker:b-9bcfa592-423a-4942-879d-eb284b418fc8",
  "rmqMessagesByQueue": {
    "pizzaQueue::/": [
      {
        "basicProperties": {
          "contentType": "text/plain",
          "contentEncoding": null,
          "headers": {
            "header1": {
              "bytes": [
                118,
                97,
                108,
                117,
                101,
                49
              ]
            },
            "header2": {
              "bytes": [
                118,
                97,
                108,
                117,
                101,
                50
              ]
            },
            "numberInHeader": 10
          },
          "deliveryMode": 1,
          "priority": 34,
          "correlationId": null,
          "replyTo": null,
          "expiration": "60000",
          "messageId": null,
          "timestamp": "Jan 1, 1970, 12:33:41 AM",
          "type": null,
          "userId": "AIDACKCEVSQ6C2EXAMPLE",
          "appId": null,
          "clusterId": null,
          "bodySize": 80
        },
        "redelivered": false,
        "data": "eyJ0aW1lb3V0IjowLCJkYXRhIjoiQ1pybWYwR3c4T3Y0YnFMUXhENEUifQ=="
      }
    ]
  }
}
            

nota

En el ejemplo de RabbitMQ, pizzaQueue es el nombre de la cola de RabbitMQ y / es el nombre del host virtual. Al recibir mensajes, la fuente del evento muestra los mensajes enpizzaQueue::/.

Permisos de rol de ejecución

Para leer registros de un agente de Amazon MQ, su función de Lambda necesita los siguientes permisos agregados a su rol de ejecución:

• mq:DescribeBroker

• secretsmanager:GetSecretValue

• ec2:CreateNetworkInterface

• ec2:DeleteNetworkInterface

• ec2:DescribeNetworkInterfaces

• ec2:DescribeSecurityGroups

• ec2:DescribeSubnets

• ec2:DescribeVpcs

• logs:CreateLogGroup

• logs:CreateLogStream

• logs:PutLogEvents

nota

Cuando utilice una clave administrada por el cliente cifrada, agregue también el permiso kms:Decrypt.

Configuración de red

De forma predeterminada, al crear un agente de Amazon MQ, la marca PubliclyAccessible está establecida en falso. Para que su agente reciba una dirección IP pública, debe establecer la marca PubliclyAccessible en verdadero.

Para que Lambda tenga acceso completo a su agente a través de su asignación de orígenes de eventos, el agente debe utilizar un punto de conexión público (dirección IP pública) o bien debe proporcionar acceso a la instancia de Amazon VPC en la que creó el agente. La práctica recomendada para utilizar Amazon MQ con Lambda consiste en utilizar puntos de conexión privados y permitir que la función de Lambda acceda a la VPC de su agente. Consulte Preferir agentes sin accesibilidad pública en la Guía para desarrolladores de Amazon MQ para obtener más información.

Tenga en cuenta que cuando agrega Amazon MQ como desencadenador, Lambda asume la configuración de VPC del agente de Amazon MQ, no la configuración de VPC de la función de Lambda. Para que la función de Lambda acceda a la VPC con el agente de Amazon MQ, realice una de las siguientes acciones:

• Configure una puerta de enlace de NAT en cada subred pública de la VPC que contenga su agente de Amazon MQ. Para obtener más información, consulte Acceso a Internet y a los servicios para funciones conectadas a la VPC.

• Cree una conexión entre la instancia de Amazon VPC que contiene su agente de Amazon MQ y un punto de conexión de VPC de Lambda. Su Amazon VPC también debe conectarse a AWS Security Token Service (AWS STS) y los puntos de conexión de Secrets Manager. Para obtener más información, consulte Conexión de puntos de conexión de VPC de la interfaz de entrada para Lambda.

También debe configurar las siguientes reglas de grupos de seguridad de Amazon VPC en la VPC:

• Reglas de entrada: permiten todo el tráfico en el puerto del agente para el grupo de seguridad especificado para el origen de eventos desde dentro del propio grupo de seguridad. ActiveMQ usa el puerto 61 617 de forma predeterminada y RabbitMQ usa el puerto 5671 de forma predeterminada.

• Reglas de salida: permiten todo el tráfico en el puerto 443 para todos los destinos. Permita todo el tráfico en el puerto del agente para dentro de su propio grupo de seguridad. ActiveMQ usa el puerto 61 617 de forma predeterminada y RabbitMQ usa el puerto 5671 de forma predeterminada.

Si utiliza puntos de conexión de VPC en lugar de una puerta de enlace de NAT, los grupos de seguridad asociados a los puntos de conexión de VPC deben permitir todo el tráfico entrante en el puerto 443 desde los grupos de seguridad del origen de eventos.

Adición de permisos y creación de la asignación de orígenes de eventos

Cree una asignación de orígenes de eventos para indicar que envíe registros de un agente de Amazon MQ a una función de Lambda. Puede crear varias asignaciones de orígenes de eventos para procesar los mismos datos con distintas funciones o para procesar elementos de varias fuentes con una sola función.

Para configurar la función para leer de Amazon MQ, agregue los permisos requeridos y cree un desencadenador de MQ en la consola de Lambda.

Cómo agregar permisos y crear un desencadenador

1. Abra la página de Funciones en la consola de Lambda.

2. Elija el nombre de una función.

3. Elija la pestaña Configuración y, a continuación, elija Permisos.

4. En Nombre del rol, elija el enlace al rol de ejecución. Este enlace abre el rol en la consola de IAM.

   

5. Seleccione Agregar permisos y, a continuación, Crear política insertada.

   

6. En Editor de políticas, elija JSON. Escriba la siguiente política. La función necesita estos permisos de lectura desde un agente de Amazon MQ.

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "mq:DescribeBroker",
           "secretsmanager:GetSecretValue",
           "ec2:CreateNetworkInterface",
           "ec2:DeleteNetworkInterface",
           "ec2:DescribeNetworkInterfaces", 
           "ec2:DescribeSecurityGroups",
           "ec2:DescribeSubnets",
           "ec2:DescribeVpcs",
           "logs:CreateLogGroup",
           "logs:CreateLogStream", 
           "logs:PutLogEvents"		
         ],
         "Resource": "*"
       }
     ]
   }

   nota

   Cuando utilice una clave cifrada y administrada por el cliente, agregue también el permiso kms:Decrypt.

7. Elija Siguiente. Introduzca un nombre de política y, a continuación, elija Crear política.

8. Regrese a la función en la consola de Lambda. En Descripción general de la función, elija Agregar desencadenador.

   

9. Elija el tipo de desencadenador MQ.

10. Configure las opciones requeridas y luego elija Agregar.

Lambda admite las siguientes opciones para las fuentes de eventos de Amazon MQ.

• Agente de MQ: seleccione un agente de Amazon MQ.

• Tamaño del lote; establezca el número máximo de mensajes que se recuperarán en un solo lote.

• Nombre de la cola: escriba la cola de Amazon MQ que se va a consumir.

• Configuración del acceso a la fuente: introduzca la información del host virtual y el secreto de Secrets Manager que almacena las credenciales del agente.

• Activar desencadenador: desactive el desencadenador para detener el procesamiento de registros.

Para habilitar o desactivar el desencadenador (o eliminarlo), elija el desencadenador de MQ en el diseñador. Para volver a configurar el desencadenador, utilice las operaciones de API de asignación de origen de eventos.

API de asignación de orígenes de eventos

Para administrar un origen de eventos con la AWS Command Line Interface (AWS CLI) o AWS SDK, puede utilizar las siguientes operaciones de la API:

• CreateEventSourceMapping

• ListEventSourceMappings

• GetEventSourceMapping

• UpdateEventSourceMapping

• DeleteEventSourceMapping

Para crear la asignación de orígenes de eventos con la AWS Command Line Interface (AWS CLI), use el comando create-event-source-mapping.

El siguiente comando AWS CLI de ejemplo crea un origen de eventos que asigna una función de Lambda denominada MQ-Example-Function a un agente basado en RabbitMQ de Amazon MQ denominado ExampleMQBroker. El comando también proporciona un nombre de host virtual y ARN secreto de Secrets Manager que almacena las credenciales del agente.

aws lambda create-event-source-mapping \
--event-source-arn arn:aws:mq:us-east-1:123456789012:broker:ExampleMQBroker:b-24cacbb4-b295-49b7-8543-7ce7ce9dfb98 \
--function-name arn:aws:lambda:us-east-1:123456789012:function:MQ-Example-Function \
--queues ExampleQueue \
--source-access-configuration Type=VIRTUAL_HOST,URI="/" Type=BASIC_AUTH,URI=arn:aws:secretsmanager:us-east-1:123456789012:secret:ExampleMQBrokerUserPassword-xPBMTt \

Debería ver los siguientes datos de salida:

{
    "UUID": "91eaeb7e-c976-1234-9451-8709db01f137",
    "BatchSize": 100,
    "EventSourceArn": "arn:aws:mq:us-east-1:123456789012:broker:ExampleMQBroker:b-b4d492ef-bdc3-45e3-a781-cd1a3102ecca",
    "FunctionArn": "arn:aws:lambda:us-east-1:123456789012:function:MQ-Example-Function",
    "LastModified": 1601927898.741,
    "LastProcessingResult": "No records processed",
    "State": "Creating",
    "StateTransitionReason": "USER_INITIATED",
    "Queues": [
        "ExampleQueue"
    ],
    "SourceAccessConfigurations": [
        {
            "Type": "BASIC_AUTH",
            "URI": "arn:aws:secretsmanager:us-east-1:123456789012:secret:ExampleMQBrokerUserPassword-xPBMTt"
        }
    ]
}

Con el comando update-event-source-mapping, puede configurar opciones adicionales, tales como la forma en que Lambda procesa los lotes y especifica cuándo descartar los registros que no se pueden procesar. El siguiente comando de ejemplo actualiza una asignación de origen de eventos para que tenga un tamaño de lote de 2.

aws lambda update-event-source-mapping \
--uuid 91eaeb7e-c976-1234-9451-8709db01f137 \
--batch-size 2

Debería ver los siguientes datos de salida:

{
    "UUID": "91eaeb7e-c976-1234-9451-8709db01f137",
    "BatchSize": 2,
    "EventSourceArn": "arn:aws:mq:us-east-1:123456789012:broker:ExampleMQBroker:b-b4d492ef-bdc3-45e3-a781-cd1a3102ecca",
    "FunctionArn": "arn:aws:lambda:us-east-1:123456789012:function:MQ-Example-Function",
    "LastModified": 1601928393.531,
    "LastProcessingResult": "No records processed",
    "State": "Updating",
    "StateTransitionReason": "USER_INITIATED"
} 

Lambda actualiza esta configuración de forma asíncrona. La salida no reflejará los cambios hasta que se complete este proceso. Para consultar el estado actual del recurso, utilice el comando get-event-source-mapping.

aws lambda get-event-source-mapping \
--uuid 91eaeb7e-c976-4939-9451-8709db01f137

Debería ver los siguientes datos de salida:

{
    "UUID": "91eaeb7e-c976-4939-9451-8709db01f137",
    "BatchSize": 2,
    "EventSourceArn": "arn:aws:mq:us-east-1:123456789012:broker:ExampleMQBroker:b-b4d492ef-bdc3-45e3-a781-cd1a3102ecca",
    "FunctionArn": "arn:aws:lambda:us-east-1:123456789012:function:MQ-Example-Function",
    "LastModified": 1601928393.531,
    "LastProcessingResult": "No records processed",
    "State": "Enabled",
    "StateTransitionReason": "USER_INITIATED"
}

Errores de asignación de orígenes de eventos

Cuando una función de Lambda encuentra un error irrecuperable, el consumidor de Amazon MQ dejará de procesar registros. Cualquier otro consumidor puede continuar procesando, siempre que no encuentre el mismo error. Para determinar la causa potencial de un consumidor detenido, marque el campo StateTransitionReason en los detalles de devolución de su EventSourceMapping para obtener uno de los siguientes códigos:

ESM_CONFIG_NOT_VALID

La configuración de asignación de orígenes de eventos no es válida.

EVENT_SOURCE_AUTHN_ERROR

Lambda ha producido un error al autenticar la fuente del evento.

EVENT_SOURCE_AUTHZ_ERROR

Lambda no tiene los permisos necesarios para acceder a la fuente de eventos.

FUNCTION_CONFIG_NOT_VALID

La configuración de la función no es válida.

Los registros también quedan sin procesar si Lambda los elimina debido a su tamaño. El límite de tamaño para los registros de Lambda es de 6 MB. Para volver a entregar mensajes en caso de error de función, puede utilizar una política de cola de mensajes fallidos (DLQ). Para obtener más información, consulte Reentrega de mensajes y administración de DLQ en el sitio web de Apache ActiveMQ y Guía de la fiabilidad en el sitio web de RabbitMQ.

nota

Lambda no admite políticas de reenvío personalizadas. En su lugar, Lambda utiliza una política con los valores predeterminados de la página de Política de reentrega en el sitio web de Apache ActiveMQ, con maximumRedeliveries establecido en 6.

Parámetros de configuración de Amazon MQ y RabbitMQ

Todos los tipos de fuente de eventos Lambda comparten las mismas operaciones CreateEventSourceMapping y UpdateEventSourceMapping de la API. Sin embargo, solo algunos de los parámetros se aplican a Amazon MQ y RabbitMQ.

Parámetros de fuente de eventos que se aplican a Amazon MQ y RabbitMQ
Parámetro	Obligatoria	Predeterminado	Notas
BatchSize	N	100	Máximo: 10 000
Habilitado	N	true
FunctionName	Y
FilterCriteria	N		Filtrado de eventos de Lambda
MaximumBatchingWindowInSeconds	N	500 ms	Comportamiento de procesamiento por lotes
Queues	N		Nombre de la cola de destino del agente de Amazon MQ que se va a consumir.
SourceAccessConfigurations	N		Para ActiveMQ, credenciales BASIC_AUTH. Para RabbitMQ, puede contener tanto credenciales BASIC_AUTH como información de VIRTUAL_HOST.