O blog da AWS

Invocando recursos locais de forma interativa usando o AWS Step Functions e o MQTT

Esta publicação foi escrita por Alex Paramonov, arquiteto sênior de soluções, ISV, e Pieter Prinsloo, gerente de soluções para clientes.

Às vezes, as cargas de trabalho na AWS exigem acesso a dados armazenados em bancos de dados e armazenamento locais. As soluções tradicionais para estabelecer conectividade com eses recursos locais exigem regras de entrada para firewalls, um túnel VPN ou endpoints públicos.

Esta postagem do blog demonstra como usar o protocolo MQTT (AWS IoT Core) com o AWS Step Functions para enviar trabalhos a workers locais para acessar ou recuperar dados armazenados localmente. A máquina de estado pode se comunicar com os workers locais sem abrir portas de entrada ou precisar de endpoints públicos em recursos locais. Os workers podem trabalhar com roteadores de Network Address Translation (NAT) enquanto mantêm a conectividade bidirecional com a nuvem da AWS. Isso fornece uma maneira mais segura e econômica de acessar dados armazenados localmente.

Visão geral

Ao usar o Step Functions com o AWS Lambda e o AWS IoT Core, você pode acessar dados armazenados localmente de forma segura sem alterar a configuração de rede existente.

O AWS IoT Core permite conectar dispositivos de IoT e rotear mensagens para os serviços da AWS sem gerenciar a infraestrutura. Ao usar uma imagem de contêiner do Docker executada localmente como um proxy IoT Thing, você pode aproveitar o agente de mensagens MQTT totalmente gerenciado do AWS IoT Core para casos de uso não relacionados à IoT.

Os subscribers do MQTT recebem informações por meio dos tópicos do MQTT. Um tópico do MQTT atua como um mecanismo de correspondência entre publishers e subscribers. Conceitualmente, um tópico do MQTT se comporta como um canal de notificação efêmero. Você pode criar tópicos em grande escala sem praticamente nenhum limite para o número de tópicos. Em aplicativos SaaS, por exemplo, você pode criar tópicos por tenant. Saiba mais sobre o design de tópicos do MQTT aqui.

A arquitetura de referência a seguir mostrada usa o AWS Serverless Application Model (AWS SAM) para implantação, o Step Functions para orquestrar o fluxo de trabalho, o AWS Lambda para enviar e receber mensagens locais e o AWS IoT Core para fornecer o agente de mensagens MQTT, gerenciamento de certificados e políticas e tópicos de publish/subscribe.

Reference architecture

  1. Inicie a máquina de estado, “on demand” ou de acordo com um cronograma.
  2. O estado: “Lambda: Invoke Dispatch Job to On-Premises” publica uma mensagem para um agente de mensagens MQTT no AWS IoT Core.
  3. O agente de mensagens envia a mensagem para o tópico correspondente ao worker (tenant) no contêiner local que executa o trabalho.
  4. O contêiner local recebe a mensagem e inicia a execução do trabalho. A autenticação é feita usando certificados de cliente e a política anexada limita o acesso do worker somente ao tópico do tenant.
  5. O worker no contêiner local pode acessar recursos locais, como bancos de dados ou locais de armazenamento.
  6. O contêiner local envia os resultados e o status do trabalho de volta para outro tópico do MQTT.
  7. A regra do AWS IoT Core invoca a função Lambda “TaskToken Done”.
  8. A função Lambda envia os resultados para o Step Functions por meio da API SendTaskSuccess ou SendTaskFailure.

Implantando e testando o exemplo

Garanta que você possa gerenciar os recursos da AWS a partir do seu terminal e que:

  • As versões mais recentes do AWS CLI e do AWS SAM CLI estão instaladas.
  • Você tem uma conta da AWS. Se não, visite esta página.
  • Seu usuário tem permissões suficientes para gerenciar os recursos da AWS.
  • O Git está instalado.
  • A versão 3.11 ou superior do Python está instalada.
  • O Docker está instalado.

Você pode acessar o repositório do GitHub aqui e seguir estas etapas para implantar o exemplo.

O diretório aws-resources contém os recursos necessários da AWS, incluindo a máquina de estado, funções, tópicos e políticas do Lambda. O diretório on-prem-worker contém os artefatos da imagem do contêiner Docker. Use-o para executar o worker localmente.

Neste exemplo, o contêiner de trabalho adiciona dois números, fornecidos como entrada no seguinte formato:

{
"a": 15,
"b": 42
}

Em um cenário real, você pode substituir essa operação pela lógica de negócios. Por exemplo, recuperar dados de bancos de dados locais, gerar agregados e, em seguida, enviar os resultados de volta para sua máquina de estado.

Siga estas etapas para testar o exemplo de ponta a ponta.

Usando o AWS IoT Core sem dispositivos de IoT

Não há dispositivos de IoT no caso de uso do exemplo. No entanto, o agente de mensagens MQTT totalmente gerenciado no AWS IoT Core permite rotear mensagens para os serviços da AWS sem gerenciar a infraestrutura.

O AWS IoT Core autentica clientes usando certificados de cliente X.509. Você pode anexar uma política a um certificado de cliente, permitindo que o cliente publique e assine somente determinados tópicos. Essa abordagem não exige credenciais do IAM dentro do contêiner de trabalho local.

A segurança, a eficiência de custos, a infraestrutura gerenciada e a escalabilidade do AWS IoT Core o tornam ideal para muitos aplicativos híbridos, além dos casos de uso típicos de IoT.

Despachando tarefas do Step Functions e aguardando uma resposta

Quando uma máquina de estado atinge o estado de despachar o trabalho para um worker local, a execução é pausada e aguarda até que o trabalho seja concluído. O Step Functions suporta três padrões de integração: request-response, sync run a job e wait for a callback with task token. O exemplo usa a integração “wait for a callback with task token”. Ele permite que a máquina de estado faça uma pausa e aguarde um retorno de chamada por até 1 ano.

Quando o worker local conclui o trabalho, ele publica uma mensagem sobre o tópico no AWS IoT Core. Uma regra no AWS IoT Core então invoca uma função Lambda, que envia o resultado de volta para a máquina de estado chamando a API SendTaskSuccess ou SendTaskFailure em Step Functions.

Você pode evitar que a máquina de estado atinja o tempo limite adicionando HeartbeatSeconds à tarefa no Amazon States Language (ASL). Os tempos limite acontecem se a tarefa congelar e a API SendTaskFailure não for chamada. HeartbeatSeconds envia pulsos do worker por meio da chamada da API SendTaskHeartbeat e deve ser menor que o timeoutSeconds especificado.

Para criar uma tarefa em ASL para sua máquina de estado, que espera por um token de retorno de chamada, use o código a seguir:

{
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke.waitForTaskToken",
      "Parameters": {
        "FunctionName": "${LambdaNotifierToWorkerArn}",
        "Payload": {
          "Input.$": "$",
          "TaskToken.$": "$$.Task.Token"
        }
}

O sufixo .waitForTaskToken  indica que a tarefa deve aguardar o retorno de chamada. A máquina de estado gera um token de retorno de chamada exclusivo, acessível por meio da variável embutida $$.Task.Token, e o passa como uma entrada para a função Lambda definida em functionName.

Em seguida, a função Lambda envia o token para o worker local por meio de um tópico do AWS IoT Core.

O Lambda não é o único serviço que oferece suporte à integração Wait for Callback — veja a lista completa de serviços compatíveis aqui.

Além de enviar tarefas e recuperar o resultado, você pode implementar mecanismos de rastreamento de progresso e desligar. Para acompanhar o progresso, o worker envia métricas por meio de um tópico separado.

Dependendo da sua implementação atual, você tem várias opções:

  1. Armazenar dados de progresso do worker no Amazon DynamoDB e visualizá-los por meio de chamadas da API REST para uma função Lambda, que lê a tabela do DynamoDB. Consulte este tutorial sobre como armazenar dados no DynamoDB diretamente do tópico.
  2. Para uma experiência de usuário reativa, crie uma regra para invocar uma função do Lambda quando novos dados de progresso chegarem. Abra uma conexão WebSocket com seu back-end. A função Lambda envia dados de progresso via WebSocket diretamente para o frontend.

Para implementar um mecanismo de desligamento, você pode executar trabalhos em threads separadas em seu worker e assinar o tópico, no qual sua máquina de estado publica as mensagens de desligamento. Se uma mensagem de desligamento chegar, encerre a thread no worker e devolva o status, incluindo o token de retorno de chamada da tarefa.

Usando as Core Rulers do AWS IoT e as funções do Lambda

Uma mensagem com os resultados do trabalho do worker não chega diretamente à API do Step Functions. Em vez disso, uma regra do AWS IoT Core e uma função Lambda dedicada encaminham a mensagem de status para o Step Functions. Isso permite permissões mais granulares nas políticas do AWS IoT Core, o que resulta em maior segurança pois o contêiner de trabalho só pode publicar e assinar tópicos específicos. Não existem credenciais do IAM no local.

O execution role da função Lambda contém as permissões somente para chamadas de API SendTaskSuccess, SendTaskHeartbeat e SendTaskFailure.

Como alternativa, um worker pode executar chamadas de API diretamente nos fluxos de trabalho do Step Functions, o que substitui a necessidade de um tópico no AWS IoT Core, uma regra e uma função do Lambda para invocar a API do Step Functions. Essa abordagem requer credenciais do IAM dentro do contêiner do worker. Você pode usar o AWS Identity and Access Management Roles Anywhere para obter credenciais de segurança temporárias. À medida que a funcionalidade do seu worker evolui com o tempo, você pode adicionar mais chamadas de API da AWS e, ao mesmo tempo, adicionar permissões à função de execução do IAM.

Limpando

Os serviços usados nessa solução são elegíveis para o nível gratuito da AWS. Para limpar os recursos no diretório aws-resources/ do repositório, execute:

sam delete

Isso remove todos os recursos provisionados pelo arquivo template.yml.

Para remover o certificado do cliente da AWS, navegue até os certificados principais do AWS IoT e exclua o certificado, que você adicionou durante as etapas de implantação manual.

Por fim, interrompa o contêiner Docker no local e remova-o:

docker rm --force mqtt-local-client

Por fim, remova a imagem do contêiner:

docker rmi mqtt-client-waitfortoken

Conclusão

Acessar recursos locais com workers controlados por meio do Step Functions usando o MQTT e o AWS IoT Core é uma forma segura, reativa e econômica de executar trabalhos locais. Considere atualizar suas cargas de trabalho híbridas do uso de pesquisas ou agendadores ineficientes para a abordagem reativa descrita nesta postagem. Isso oferece uma experiência de usuário aprimorada com envio e rastreamento rápidos de trabalhos fora da nuvem.

Para obter mais recursos de aprendizado sem servidor, visite Serverless Land.

Este blog é uma tradução do conteúdo original em inglês (link aqui).

Autores

Alex Paramonov, Arquiteto Sênior de Soluções, ISV.
Pieter Prinsloo, Gerente de Soluções para Clientes.

Biografia do Tradutor

Rodrigo Peres é Arquiteto de Soluções na AWS, com mais de 20 anos de experiência trabalhando com arquitetura de soluções, desenvolvimento de sistemas e modernização de sistemas legados.