Docs Menu
Docs Home
/ /
Etapas de agregación
Docs Home
/
Desarrollo
/
Atlas Stream Processing
/
Etapas de agregación
Docs Home
/
Desarrollo
/
Atlas Stream Processing
/
Etapas de agregación

$externalFunction

Definición

$externalFunction

El $externalFunction La etapa activa procesos en un recurso específico de AWS Lambda. Su solicitud al proceso de AWS Lambda puede ser síncrona o asíncrona.

Cree una AWS Lambda y autentíquese con Unified AWS Access

Para llamar a un recurso de AWS Lambda desde su canalización de Atlas Stream Processing, su AWS Lambda debe estar implementado en la misma región de AWS donde se implementa Atlas Stream Processing. Para obtener más información sobre la implementación de un recurso de AWS Lambda, consulte la documentación de AWS.

1

Cree una función AWS Lambda.

Con la CLI de AWS o mediante la interfaz de usuario de AWS, cree una función lambda.

Nota

  • Los recursos de AWS Lambda deben crearse con un AWS_IAM tipo de autenticación, lo que requiere la creación de una función y una política de AWS IAM.

  • Solo se admite el tipo de respuesta almacenada en búfer, no en streaming.

2

Configurar el acceso unificado a AWS.

Nota

El procedimiento descrito aquí solo cubre el flujo de configuración básico en la interfaz de usuario de Atlas. Para obtener más información, consulte Configurar la documentación de Unified AWS Access.

Acceso requerido

Para configurar unificado Acceso a AWS, debe Organization Owner tener acceso o Project Owner al proyecto.

Requisitos previos

Nota

  • Su política de AWS IAM debe incluir la acción lambda:InvokeFunction.

  • Debe reemplazar los valores de los marcadores de posición ExternalId y Resource por los suyos propios, disponibles mediante el proceso de configuración de Unified AWS Access. Tenga en cuenta que el ExternalId de este ejemplo incluye un comodín que coincide con cualquier función Lambda cuyo nombre comience por function-.

3

Añade relaciones de confianza a un rol existente en la interfaz de usuario de Atlas

A continuación, debe habilitar su rol de AWS IAM autoadministrado para ejecutar su recurso de AWS Lambda.

permission-policy.json

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "lambda:InvokeFunction"
            ],
            "Resource": "arn:aws:lambda:us-east-1:257394458927:function:<function-name>"
        }
    ]
}

  1. Vaya a la página de integración de AWS IAM en su proyecto Atlas y haga clic en el icono Authorize an AWS IAM role botón.

  2. Crea un nuevo rol (o modifica un rol existente) con el role-trust-policy.json que se muestra en el modal.

  3. Una vez creado el rol (o actualizado el rol existente con la nueva política de confianza), pegue el ARN del rol en el modal.

  4. En la consola de AWS, vaya a IAM > Roles y seleccione su rol.

  5. En la pestaña permissions, agregue un nuevo "permiso en línea" para permitir que este rol invoque sus lambdas. El ejemplo permission-policy.json anterior agrega el permiso para ejecutar cualquier lambda con el nombre <function-name>.

  6. Por último, navegue a su espacio de trabajo de procesamiento de secuencias Atlas, agregue un nuevo AWS Lambda connection y elija el AWS IAM Role ARN que configuró en el paso anterior.

Conecte su espacio de trabajo de procesamiento de flujos de Atlas a su función AWS Lambda

Para enviar una solicitud a su recurso de AWS Lambda desde su canalización de Atlas Stream Processing, primero debe agregar su recurso de AWS Lambda como una conexión en su recurso de Atlas Stream Processing.

Puede agregar su recurso de AWS Lambda como una conexión a través de la interfaz de usuario de Atlas, la CLI de Atlas o la API de Atlas, como se muestra en el siguiente ejemplo. Puede actualizar el roleArn marcador de posición en el ejemplo con el arn de su configuración de AWS IAM.

curl --user "username:password" --digest \
   --header "Content-Type: application/json" \
   --header "Accept: application/vnd.atlas.2023-02-01+json" \
   --include \
   --data '{"name": "TestAWSLambdaConnection","type": "AWSLambda","aws": {"roleArn": "arn:aws:iam::<aws_account>:role/<role_name>"}}' \
   --request POST "https://cloud.mongodb.com/api/atlas/v2/groups/<group_id>/streams/<tenant_name>/connections"

Sintaxis

Solicitud mínima

El siguiente ejemplo muestra los campos obligatorios para una solicitud mínima.

{ $externalFunction: {
  connectionName: "myLambdaConnection",
  functionName: "arn:aws:lambda:region:account-id:function:function-name",
  as: "response",
}}

Solicitud personalizada

El siguiente ejemplo personalizado especifica el manejo de errores, la ejecución sincrónica y un documento de procesamiento de flujo Atlas preprocesado como carga útil, además de los campos obligatorios ilustrados anteriormente.

{ $externalFunction: {
        connectionName: "myLambdaConnection",
        functionName: "arn:aws:lambda:region:account-id:function:function-name",
        execution: "sync"
        as: "response",
        onError: "fail",
        payload: [{$replaceRoot: { newRoot: "$fullDocument.payloadToSend" } }, { $addFields: { sum: { $sum: "$randomArray" }}}, { $project: { success: 1, sum: 1 }}],
}}

Nota

El campo onError define el comportamiento de los errores de nivel de API para solicitudes sincrónicas y asincrónicas a su recurso de AWS Lambda, así como errores de función de AWS Lambda para solicitudes sincrónicas.

La etapa $externalFunction procesa un documento con los siguientes campos:

Campo
Tipo
Necesidad
Descripción

connectionName

string

Requerido

Etiqueta que identifica la conexión en el Registro de Conexiones, a la que se envía la solicitud.

functionName

string

Requerido

El ARN de AWS completo o el nombre de la función de AWS Lambda que se activará.

execution

enum

Opcional

Parámetro que especifica si la función AWS Lambda debe llamarse de forma síncrona o asíncrona. Los valores aceptados son:

  • sync

  • async

El valor predeterminado es sync, que es necesario si las etapas restantes en su canalización de procesamiento de secuencias de Atlas requieren la salida de su función de AWS Lambda.

as

string

Opcional

Nombre del campo para la respuesta de la API REST.

Si el punto final devuelve 0 bytes, el operador no establece el campo as.

onError

string

Opcional

Comportamiento cuando el operador detecta un error HTTPS Status Code o relacionado con el tiempo de ejecución de Lambda. Debe ser uno de los siguientes valores:

  • "dlq" :Pase el documento afectado a la cola de mensajes no entregados.

  • "ignore" :No hacer nada con el documento afectado.

  • "fail" :Terminar el procesador de flujo en caso de error.

onError no se activa ante errores que surgen de una configuración incorrecta del propio operador $externalFunction, como expresiones no válidas.

Se establece por defecto en "dlq".

payload

arreglo

Opcional

Pipeline interno personalizado que permite personalizar el cuerpo de la solicitud enviado al endpoint de la API. payload soporta las siguientes expresiones:

  • $project

  • $addFields

  • $replaceRoot

  • $set

Comportamiento

La etapa envía una solicitud al recurso de AWS Lambda especificado. Si la solicitud es síncrona, la respuesta se devuelve en el campo especificado y la canalización continúa el procesamiento. Si la solicitud es asíncrona, la canalización continúa el procesamiento sin esperar la $externalFunction respuesta.

Si falla una solicitud síncrona, el comportamiento de gestión de errores de la canalización se determina mediante el onError campo. Si la solicitud es asíncrona, el onError campo solo se aplica a los errores de la API de AWS, a diferencia de los errores de la función AWS Lambda. Si onError no se especifica el campo, el comportamiento predeterminado es enviar el documento afectado a la cola de mensajes fallidos.

onError Valor
Comportamiento

"dlq"

El documento afectado se envía a la cola de mensajes no entregados.

"ignore"

El documento afectado se ignora.

"fail"

El procesador de flujo finaliza en caso de error.

Si la canalización no puede convertir el documento a JSON adecuado, o la canalización configurada no crea un objeto bson válido como producto de la etapa final, el documento se envía a la cola de mensajes no entregados independientemente de la onError configuración.

Los errores derivados de una configuración incorrecta del operador $externalFunction, como expresiones no válidas, no activan el comportamiento onError. En su lugar, el procesador de flujos falla con un mensaje de error. La canalización de procesamiento de flujos de Atlas reintenta las solicitudes fallidas que pueden deberse a errores transitorios.

Volver

$merge

Next

Seguridad

En esta página