Carregamento de dados de transmissão do Amazon S3

Você pode usar o Lambda para enviar dados para o domínio do OpenSearch Service do Amazon S3. Os novos dados recebidos em um bucket do S3 acionam uma notificação de evento para o Lambda, que executa seu código personalizado para realizar a indexação.

Esse método de streaming de dados é extremamente flexível. Você pode indexar metadados de objeto, ou se o objeto for texto simples, analisar e indexar alguns elementos do corpo do objeto. Esta seção inclui alguns códigos de exemplo Python simples que usam expressões regulares para analisar um arquivo de log e indexar as correspondências.

Pré-requisitos

Para continuar, você deve ter os recursos a seguir.

Pré-requisito	Descrição
Bucket do Amazon S3.	Para obter mais informações, consulte Criar seu primeiro bucket do S3 no Manual do usuário do Amazon Simple Storage Service. O bucket deve residir na mesma região do domínio do OpenSearch Service.
Domínio do OpenSearch Service	O destino dos dados depois que a função do Lambda os processa. Para ter mais informações, consulte Criação OpenSearch de domínios de serviço.

Criar o pacote de implantação do Lambda

Os pacotes de implantação são arquivos ZIP ou JAR que contêm o código e as dependências. Esta seção inclui código de exemplo Python. Para outras linguagens de programação, consulte Pacotes de implantação do Lambda no Guia do desenvolvedor do AWS Lambda.

1. Crie um diretório. Neste exemplo, usamos o nome s3-to-opensearch.

2. Crie um arquivo no diretório chamado sample.py:

   import boto3
   import re
   import requests
   from requests_aws4auth import AWS4Auth
   
   region = '' # e.g. us-west-1
   service = 'es'
   credentials = boto3.Session().get_credentials()
   awsauth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)
   
   host = '' # the OpenSearch Service domain, e.g. https://search-mydomain.us-west-1.es.amazonaws.com
   index = 'lambda-s3-index'
   datatype = '_doc'
   url = host + '/' + index + '/' + datatype
   
   headers = { "Content-Type": "application/json" }
   
   s3 = boto3.client('s3')
   
   # Regular expressions used to parse some simple log lines
   ip_pattern = re.compile('(\d+\.\d+\.\d+\.\d+)')
   time_pattern = re.compile('\[(\d+\/\w\w\w\/\d\d\d\d:\d\d:\d\d:\d\d\s-\d\d\d\d)\]')
   message_pattern = re.compile('\"(.+)\"')
   
   # Lambda execution starts here
   def handler(event, context):
       for record in event['Records']:
   
           # Get the bucket name and key for the new file
           bucket = record['s3']['bucket']['name']
           key = record['s3']['object']['key']
   
           # Get, read, and split the file into lines
           obj = s3.get_object(Bucket=bucket, Key=key)
           body = obj['Body'].read()
           lines = body.splitlines()
   
           # Match the regular expressions to each line and index the JSON
           for line in lines:
               line = line.decode("utf-8")
               ip = ip_pattern.search(line).group(1)
               timestamp = time_pattern.search(line).group(1)
               message = message_pattern.search(line).group(1)
   
               document = { "ip": ip, "timestamp": timestamp, "message": message }
               r = requests.post(url, auth=awsauth, json=document, headers=headers)

   Edite as variáveis de region e host.

3. Se ainda não o fez, instale o pip. Em seguida, instale as dependências em um novo diretório package:

   cd s3-to-opensearch
   
   pip install --target ./package requests
   pip install --target ./package requests_aws4auth

   Como todos os ambientes de execução do Lambda têm o Boto3 instalado, você não precisa incluí-lo no pacote de implantação.

4. Empacote o código do aplicativo e as dependências:

   cd package
   zip -r ../lambda.zip .
   
   cd ..
   zip -g lambda.zip sample.py

Criar a função do Lambda

Depois de criar o pacote de implantação, você poderá criar a função do Lambda. Ao criar uma função, escolha um nome, o tempo de execução (por exemplo, Python 3.8) e a função do IAM. A função do IAM define as permissões para a função. Para obter instruções detalhadas, consulte Criar uma função Lambda com o console no Guia do desenvolvedor do AWS Lambda.

Esse exemplo pressupõe que você está usando o console. Escolha o Python 3.9 e uma função que tenha permissões de leitura do S3 e permissões de gravação do OpenSearch Service, conforme mostrado na captura de tela a seguir:

Depois de criar a função, você deverá adicionar um gatilho. Neste exemplo, queremos que o código seja executado sempre que um arquivo de log chegue em um bucket do S3:

1. Escolha Adicionar acionador e selecione S3.

2. Escolha o bucket.

3. Em Tipo de evento, selecione PUT.

4. Em Prefixo, digite logs/.

5. Em Sufixo, digite .log.

6. Confirme o aviso de invocação recursiva e escolha Adicionar.

Por fim, você pode carregar o pacote de implantação:

1. Escolha Carregar de e arquivo .zip e siga os avisos para carregar do pacote de implantação.

2. Depois que o carregamento terminar, edite as Configurações do tempo de execução e altere o Manipulador para sample.handler. Essa configuração informa ao Lambda o arquivo (sample.py) e o método (handler) que deverão ser executados depois de um acionador.

A esta altura, você tem um conjunto completo de recursos: um bucket para arquivos de log, uma função executada sempre que um arquivo de log é adicionado ao bucket, o código que realiza a análise e a indexação e um domínio do OpenSearch Service para pesquisa e visualização.

Teste da função do Lambda

Após criar a função, você poderá testá-la carregando de um arquivo no bucket do Amazon S3. Crie um arquivo chamado sample.log usando as seguintes linhas de log de exemplo:

12.345.678.90 - [10/Oct/2000:13:55:36 -0700] "PUT /some-file.jpg"
12.345.678.91 - [10/Oct/2000:14:56:14 -0700] "GET /some-file.jpg"

Carregue o arquivo na pasta logs do bucket do S3. Para obter instruções, consulte Carregar um objeto para o seu bucket no Manual do usuário do Amazon Simple Storage Service.

Em seguida, use o console do OpenSearch Service ou o OpenSearch Dashboards para verificar se o índice lambda-s3-index contém dois documentos. Você também pode fazer uma solicitação de pesquisa padrão:

GET https://domain-name/lambda-s3-index/_search?pretty
{
  "hits" : {
    "total" : 2,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "lambda-s3-index",
        "_type" : "_doc",
        "_id" : "vTYXaWIBJWV_TTkEuSDg",
        "_score" : 1.0,
        "_source" : {
          "ip" : "12.345.678.91",
          "message" : "GET /some-file.jpg",
          "timestamp" : "10/Oct/2000:14:56:14 -0700"
        }
      },
      {
        "_index" : "lambda-s3-index",
        "_type" : "_doc",
        "_id" : "vjYmaWIBJWV_TTkEuCAB",
        "_score" : 1.0,
        "_source" : {
          "ip" : "12.345.678.90",
          "message" : "PUT /some-file.jpg",
          "timestamp" : "10/Oct/2000:13:55:36 -0700"
        }
      }
    ]
  }
}