Docs Menu
Docs Home
/
Atlas
/
Atlas Stream Processing
/
Aggregation Pipelines

$externalFunction

On this page

Definition

$externalFunction

The $externalFunction stage triggers processes in a specific AWS Lambda resource. Your request to the AWS Lambda process can either be synchronous or asynchronous.

Create an AWS Lambda and Authenticate with Unified AWS Access

In order to call an AWS Lambda resource from within your Atlas Stream Processing pipeline, your AWS Lambda must be deployed to the same AWS region in which your Atlas Stream Processing is deployed. To learn more about deploying an AWS Lambda resource, see the AWS documentation.

1

Create an AWS Lambda Function.

With either the AWS CLI or through the AWS UI, create a lambda function.

Note

  • AWS Lambda resources must be created with an AWS_IAM authentication type, which requires that you create an AWS IAM Role and Policy.

  • Only the buffered, not streaming, response type is supported.

2

Configure Unified AWS Access.

Note

The procedure described here only covers the basic setup flow in the Atlas UI. To learn more, see the the Set Up Unified AWS Access documentation.

Required Access

To set up unified AWS access, you must have Organization Owner or Project Owner access to the project.

Prerequisites

Note

  • Your AWS IAM policy must include the lambda:InvokeFunction action.

  • You must replace the placeholder ExternalId and Resource values with your own, which are available through the Unified AWS Access configuration process. Note that the ExternalId in this example includes a wildcard that matches any Lambda Function with a name that begins with function-.

3

Add Trust Relationships to an Existing Role in the Atlas UI

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. Navigate to the AWS IAM integration page in your Atlas project and click on the Authorize an AWS IAM role button.

  2. Create a new role (or modify an existing role) with the role-trust-policy.json that is shown in the modal.

  3. Once the role is created (or the existing role is updated with the new trust policy), paste the role's ARN in the modal.

  4. In the AWS console, go to IAM > Roles and select your role.

  5. In the permissions tab, add a new "inline permission" to allow this role to invoke your lambda(s). The example permission-policy.json provided above adds the permission to run any lambda with the name <function-name>.

  6. Finally, navigate to your Atlas Stream Processing Instance, add a new AWS Lambda connection and choose the AWS IAM Role ARN that you configured in the previous step.

Connect your Atlas Stream Processing Instance to your AWS Lambda Function

In order to send a request to your AWS Lambda resource from within your Atlas Stream Processing pipeline, you must first add your AWS Lambda resource as a connection in your Atlas Stream Processing resource.

You can add your AWS Lambda resource as a connection through the Atlas UI, the MongoDB Terraform provider, the Atlas CLI, or the Atlas API, as shown in the following example:

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"

Note

You can update the roleArn placeholder in the above example with the arn from your AWS IAM configuration.

Syntax

Minimal request

The following example shows the required fields for a minimal request.

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

Customized request

The following, customized example specifies error handling, synchronous execution and a preprocessed Atlas Stream Processing document as a payload in addition to the required fields illustrated above.

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

Note

The onError field defines behavior for API level errors for both synchronous and asynchronous requests to your AWS Lambda resource as well as AWS Lambda function errors for synchronous requests.

The $externalFunction stage takes a document with the following fields:

Field
Type
Necessity
Description

connectionName

string

Required

Label that identifies the connection in the Connection Registry, to which the request is sent.

functionName

string

Required

The full AWS ARN or the name of the AWS Lambda function to be triggered.

execution

enum

Optional

Parameter that specifies whether the AWS Lambda function should be called synchronously or asynchronously. Accepted values are:

  • sync

  • async

Defaults to sync, which is required if remaining stages in your Atlas Stream Processing pipeline require the output from your AWS Lambda function.

as

string

Optional

Name of the field for the REST API response.

If the endpoint returns 0 bytes, the operator doesn't set the as field.

onError

string

Optional

Behavior when the operator encounters an HTTPS Status Code or Lambda Runtime-related failure. Must be one of the following values:

  • "dlq" : Pass the affected document to the dead letter queue.

  • "ignore" : Do nothing with the affected document.

  • "fail" : Terminate the stream processor on error.

onError does not trigger on errors arising from incorrect configuration of the $externalFunction operator itself, such as invalid expressions.

Defaults to "dlq".

payload

array

Optional

Custom inner pipeline that allows you to customize the request body sent to the API endpoint. payload supports the following expressions:

  • $project

  • $addFields

  • $replaceRoot

  • $set

Back

$merge

Next

Monitoring