Menu Docs
Página inicial do Docs
/ /
Estágios de aggregation
Página inicial do Docs
/
Desenvolvimento
/
Atlas Stream Processing
/
Estágios de aggregation
Página inicial do Docs
/
Desenvolvimento
/
Atlas Stream Processing
/
Estágios de aggregation

$externalFunction

Definição

$externalFunction

A fase $externalFunction aciona processos em um recurso específico do AWS Lambda. Sua solicitação para o processo do AWS Lambda pode ser síncrona ou assíncrona.

Crie um AWS Lambda e autentique com o AWS Access Unificado

Para chamar um recurso do Amazon Web Services Lambda de dentro do pipeline do Atlas Stream Processing , o Amazon Web Services Lambda deve ser implantado na mesma região do Amazon Web Services em que o Atlas Stream Processing está implantado. Para saber mais sobre como implantar um recurso do Amazon Web Services Lambda, consulte a documentação do Amazon Web Services.

1

Crie uma função de Lambda do Amazon Web Services.

Com a Amazon Web Services CLI ou a interface do usuário do Amazon Web Services, crie uma função Lambda.

Observação

  • Os recursos do Amazon Web Services Lambda devem ser criados com um tipo de autenticação AWS_IAM, que exige que você crie uma função e uma política.do Amazon Web Services IAM.

  • Apenas o tipo de resposta em buffer, e não o de streaming, é compatível.

2

Configure o acesso unificado do AWS.

Observação

O procedimento descrito aqui cobre apenas o fluxo básico de configurar na IU do Atlas. Para aprender mais, consulte a documentação Configurar o Acesso Unificado da AWS.

Acesso necessário

Para configurar o acesso unificado AWS, você deve ter o acesso Organization Owner ou Project Owner ao projeto.

Pré-requisitos

Observação

  • Sua política do AWS IAM deve incluir a ação lambda:InvokeFunction.

  • Substitua os valores de ExternalId e Resource de espaço reservado pelos seus, que estão disponíveis por meio do processo de configuração do Unified AWS Access. Observe que o ExternalId neste exemplo inclui um curinga, correspondente a qualquer função do Lambda com um nome que começa com function-.

3

Adicionar relações de confiança a uma função existente na IU do Atlas

Em seguida, você deve habilitar sua função autogerenciada do AWS IAM para executar seu recurso do 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. Navegue até a página de integração do AWS IAM no seu projeto Atlas e clique no botão Authorize an AWS IAM role.

  2. Crie uma nova função (ou modifique uma função existente) com o role-trust-policy.json mostrado no modal.

  3. Assim que a função for criada (ou a função existente for atualizada com a nova política de confiança), cole o ARN da função no modal.

  4. No console da AWS, acesse IAM > Roles e selecione sua função.

  5. Na aba permissions, adicione uma nova "permissão em linha" para permitir que esta função invoque seu(s) lambda(s). O exemplo permission-policy.json fornecido acima adiciona a permissão para executar qualquer lambda com o nome <function-name>.

  6. Por fim, navegue até o Atlas Stream Processing Workspace, adicione um novo AWS Lambda connection e escolha o AWS IAM Role ARN que você configurou na etapa anterior.

Conecte seu Atlas Stream Processing à sua AWS Lambda Function

Para enviar uma solicitação ao recurso AWS Lambda de dentro do pipeline do Atlas Stream Processing, você deve primeiro adicionar o recurso AWS Lambda como uma conexão no recurso Atlas Stream Processing.

Você pode adicionar seu recurso AWS Lambda como uma conexão por meio da IU do Atlas, da Atlas CLI ou da API do Atlas, conforme mostrado no exemplo a seguir. Você pode atualizar o espaço reservado roleArn no exemplo com o arn da sua configuração do 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"

Sintaxe

Solicitação mínima

O exemplo a seguir mostra os campos obrigatórios para uma solicitação mínima.

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

Solicitação customizada

O exemplo personalizado a seguir especifica o tratamento de erros, a execução síncrona e um documento pré-processado do Atlas Stream Processing como carga útil, além dos campos obrigatórios ilustrados acima.

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

Observação

O campo onError define o comportamento para erros de nível da API em solicitações síncronas e assíncronas para seu recurso AWS Lambda, assim como para erros de função do AWS Lambda em solicitações síncronas.

O estágio $externalFunction recebe um documento com os seguintes campos:

Campo
Tipo
necessidade
Descrição

connectionName

string

Obrigatório

Rótulo que identifica a conexão no registro de conexões, para o qual a solicitação é enviada.

functionName

string

Obrigatório

O ARN completo da AWS ou o nome da função AWS Lambda a ser acionada.

execution

enum

Opcional

Parâmetro que especifica se a função do Amazon Web Services Lambda deve ser chamada de forma síncrona ou assíncrona. Os valores aceitos são:

  • sync

  • async

O padrão é sync, que é necessário se os estágios restantes no pipeline do Atlas Stream Processing precisarem da saída da função do AWS Lambda.

as

string

Opcional

Nome do campo para a resposta da REST API.

Se o ponto de extremidade retornar 0 bytes, o operador não configurará o campo as.

onError

string

Opcional

Comportamento quando o operador encontra uma falha relacionada a HTTPS Status Code ou ao tempo de execução do Lambda. Deve ser um dos seguintes valores:

  • "dlq" : passe o documento afetado para a fila de mensagens não entregues.

  • "ignore" : Não faça nada com o documento afetado.

  • "fail" : Finalizar o processador de fluxo em caso de erro.

onError não é acionado por erros decorrentes da configuração incorreta do operador $externalFunction em si, como expressões inválidas.

Padrão é "dlq".

payload

array

Opcional

Pipeline interno personalizado que permite personalizar o corpo da solicitação enviado ao ponto de extremidade da API. payload suporta as seguintes expressões:

  • $project

  • $addFields

  • $replaceRoot

  • $set

Comportamento

O estágio $externalFunction envia uma solicitação para o recurso Lambda do Amazon Web Services especificado. Se a solicitação for síncrona, a resposta será retornada no campo especificado e o pipeline continuará processando. Se a solicitação for assíncrona, o pipeline continuará processando sem aguardar a resposta.

Se uma solicitação síncrona falhar, o comportamento de tratamento de erros do pipeline será determinado pelo campo onError. Se a solicitação for assíncrona, o campo onError será aplicado somente a erros da API do Amazon Web Services, em oposição aos erros de função do Amazon Web Services Lambda. Se o campo onError não for especificado, o comportamento padrão será enviar o documento afetado para a fila de mensagens não entregues (DLQ).

onError Valor
Comportamento

"dlq"

O documento afetado é enviado para a fila de mensagens não entregues (DLQ).

"ignore"

O documento afetado é ignorado.

"fail"

O processador de fluxo é encerrado em caso de erro.

Se o pipeline não puder converter o documento no JSON adequado ou se o pipeline configurado não criar um objeto BSON válido como produto do estágio final, o documento será enviado para a fila de mensagens não entregues (DLQ), independentemente da configuração onError.

Erros resultantes da configuração incorreta do próprio operador $externalFunction, como expressões inválidas, não acionam o comportamento onError. Em vez disso, o processador de fluxo falha com uma mensagem de erro. O pipeline do Atlas Stream Processing tenta novamente solicitações com falha que podem ser o resultado de erros transitórios.

Voltar

$merge

Próximo

$sessionWindow

Nesta página