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 un mapero de fuente de eventos, un recurso de Lambda que lee los mensajes de su agente e invoca la función sincrónicamente.

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

• 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 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 Virtual Private Cloud (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 asmapeo de fuente de eventos. Para RabbitMQ, solo se admite una implementación de agente o clúster de una instancia única por mapeo de fuente de eventos. Los agentes de instancia única requieren un punto de enlace 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/Java Message Service (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 mapeo de origen 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 del origen del evento 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 enlace 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.

Configuración de un agente como origen de eventos

Cree un mapeo de fuente de eventos para indicar que envíe registros de un agente de Amazon MQ a una función Lambda. Puede crear varios mapeos 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, cree un desencadenador de MQ en la consola de Lambda.

Para crear un desencadenador

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

2. Elija el nombre de una función.

3. En Function overview (Descripción general de la función), elija Add trigger (Agregar desencadenador).

4. Elija un tipo de desencadenador.

5. Configure las opciones requeridas y luego elija Add (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 mapeo de origen 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

nota

Cuando actualice, deshabilite o elimine una asignación de orígenes de eventos para Amazon MQ, los cambios pueden tardar hasta 15 minutos en surtir efecto. Antes de que haya transcurrido este periodo, la asignación de orígenes de eventos puede continuar con el procesamiento de los eventos y la invocación de la función con la configuración anterior. Esto es cierto incluso cuando el estado de la asignación de orígenes de eventos que se muestra en la consola indica que se aplicaron los cambios.

Para crear el mapeo de origen del evento 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 mapeo de origen 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 mapeo de origen del evento 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 Política de reentrega en el sitio web de Apache ActiveMQ, con maximumRedeliveries establecido en 5.

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	Obligatorio	Valor 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.