$externalFunction

Definición

$externalFunction

La $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.

Crea una función AWS Lambda y autentícate con Unified AWS Access

Para llamar a un recurso de AWS Lambda desde dentro de tu pipeline de Atlas Stream Processing, tu AWS Lambda debe estar implementado en la misma región de AWS en la que tu Atlas Stream Processing esté implementado. Para obtener más información sobre cómo implementar un recurso de AWS Lambda, consulta la documentación de AWS.

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

Conecta tu espacio de trabajo de Atlas Stream Processing a tu función AWS Lambda

Para enviar una solicitud a tu recurso de AWS Lambda desde tu pipeline de Atlas Stream Processing, primero debes agregar tu recurso de AWS Lambda como una conexión en tu 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 --header "Authorization: Bearer {ACCESS-TOKEN}" \
   --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

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

{ $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 }}],
}}

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 una solicitud síncrona falla, el comportamiento de manejo de errores del pipeline está determinado por el campo onError. Si la solicitud es asíncrona, el campo onError se aplica solo a errores de la API de AWS, en lugar de errores de función de AWS Lambda. Si el campo onError no se especifica, el comportamiento por defecto es enviar el documento afectado a la fila de letra muerta.

Si la canalización no puede convertir el documento a un JSON correcto 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 fila de letra muerta independientemente del ajuste onError.

Los errores que surgen de una configuración incorrecta del operador $externalFunction en sí, como expresiones no válidas, no desencadenan el comportamiento onError. En cambio, el procesador de flujo falla con un mensaje de error. La pipeline de Atlas Stream Processing reintenta las solicitudes fallidas que pueden ser resultado de errores transitorios.