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 registros. 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.

La asignación de orígenes 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 Configurar la simultaneidad reservada para una función.

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, el origen de eventos 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

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.

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.

La práctica recomendada para utilizar Amazon MQ con Lambda consiste en utilizar puntos de conexión de VPC de AWS PrivateLink y permitir que la función de Lambda acceda a la VPC de su agente. Implemente un punto de conexión para Lambda y, solo para ActiveMQ, un punto de conexión para AWS Security Token Service (AWS STS). Si el agente utiliza autenticación, implemente también un punto de conexión de VPC para AWS Secrets Manager. Para obtener más información, consulte Uso de los puntos de enlace de la VPC.

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 Habilitación del acceso a Internet para funciones de Lambda conectadas a VPC.

Al crear una asignación de orígenes de eventos para un agente de Amazon MQ, Lambda comprueba si las interfaces de red elásticas (ENI) ya están presentes en las subredes y los grupos de seguridad de la VPC de su agente. Si Lambda encuentra ENI existentes, intenta reutilizarlos. De lo contrario, Lambda crea nuevos ENI para conectarse al origen de eventos e invocar la función.

nota

Las funciones de Lambda siempre se ejecutan dentro de VPC propiedad del servicio de Lambda. El servicio mantiene estas VPC de forma automática, y no son visibles para los clientes. También puede conectar su función a una Amazon VPC. En cualquier caso, la configuración de VPC de la función no afecta la asignación de orígenes de eventos. Solo la configuración de la VPC del origen de eventos determina cómo Lambda se conecta al origen de eventos.

Reglas de los grupos de seguridad de VPC

Configure los grupos de seguridad de Amazon VPC que contienen su clúster con las siguientes reglas (como mínimo):

• 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.

Uso de los puntos de enlace de la VPC

Cuando utiliza puntos de conexión de VPC, las llamadas a la API para invocar su función se enrutan a través de estos puntos de conexión mediante los ENI. La entidad principal del servicio de Lambda debe llamar a lambda:InvokeFunction para cualquier función que utilice esos ENI. Además, en el caso de ActiveMQ, la entidad principal del servicio de Lambda debe llamar a sts:AssumeRole en roles que utilicen los ENI.

De forma predeterminada, los puntos de conexión de VPC tienen políticas de IAM abiertas. La práctica recomendada es restringir estas políticas para permitir que solo entidades principales realicen las acciones necesarias mediante ese punto de conexión. Para garantizar que la asignación de orígenes de eventos pueda invocar la función de Lambda, la política de puntos de conexión de VPC debe permitir que el principio del servicio de Lambda llame a lambda:InvokeFunction y, para ActiveMQ, a sts:AssumeRole. Restringir las políticas de puntos de conexión de VPC para permitir únicamente las llamadas a la API que se originen en su organización impide que la asignación de orígenes de eventos funcione correctamente.

En el siguiente ejemplo de políticas de puntos de conexión de VPC, se muestra cómo conceder el acceso necesario a AWS STS y los puntos de conexión de Lambda.

ejemplo Política de puntos de conexión de VPC: punto de conexión de AWS STS (solo ActiveMQ)

{
    "Statement": [
        {
            "Action": "sts:AssumeRole",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}

ejemplo Política del punto de conexión de VPC: punto de conexión de Lambda

{
    "Statement": [
        {
            "Action": "lambda:InvokeFunction",
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com"
                ]
            },
            "Resource": "*"
        }
    ]
}

Si su agente de Amazon MQ utiliza la autenticación, también puede restringir la política de puntos de conexión de VPC para el punto de conexión de Secrets Manager. Para llamar a la API de Secrets Manager, Lambda usa su rol de función, no la entidad principal de servicio de Lambda. En el siguiente ejemplo, se muestra una política del punto de conexión de Secrets Manager.

ejemplo Política de punto de conexión de VPC: punto de conexión de Secrets Manager

{
    "Statement": [
        {
            "Action": "secretsmanager:GetSecretValue",
            "Effect": "Allow",
            "Principal": {
                "AWS": [
                    "customer_function_execution_role_arn"
                ]
            },
            "Resource": "customer_secret_arn"
        }
    ]
}

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 varios orígenes 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 los orígenes 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 al origen: 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.

Actualización de la asignación de orígenes de eventos

Utilice el comando update-event-source-mapping para actualizar una asignación de orígenes de eventos. 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 el origen del evento.

EVENT_SOURCE_AUTHZ_ERROR

Lambda no tiene los permisos necesarios para acceder al origen 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 origen 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.