🥇Exemplo de um aplicativo baseado em eventos usando webhooks no armazenamento de objetos S3 da Mail.ru Cloud Solutions

Máquina de café de Rube Goldberg

A arquitetura orientada a eventos aumenta a eficiência de custo dos recursos utilizados, pois eles são acionados apenas no momento em que são necessários. Existem várias maneiras de implementar isso sem criar entidades em nuvem adicionais como aplicativos worker. E hoje eu não vou falar sobre FaaS, mas sobre webhooks. Vou mostrar um exemplo prático de processamento de eventos usando webhooks de armazenamento de objetos.

Algumas palavras sobre armazenamento de objetos e webhooks. Os armazenamentos de objetos permitem armazenar quaisquer dados na nuvem na forma de objetos acessíveis pelo S3 ou outra API (dependendo da implementação) via HTTP/HTTPS. Webhooks, em geral, são callbacks personalizados via HTTP. Normalmente, eles são acionados por um evento, como o envio de código para um repositório ou um comentário postado em um blog. Quando um evento ocorre, o site de origem envia uma solicitação HTTP para a URL especificada para o webhook. Como resultado, é possível fazer com que eventos em um site acionem ações em outro. No caso em que o site de origem é um armazenamento de objetos, as mudanças em seu conteúdo atuam como eventos.wiki). Quando o site de origem é um armazenamento de objetos, as alterações em seu conteúdo atuam como eventos.

Exemplos de casos simples onde essa automação pode ser utilizada:

Criação de cópias de todos os objetos em outro armazenamento em nuvem. As cópias devem ser feitas "em tempo real", a cada adição ou modificação de arquivos.
Criação automática de séries de miniaturas de arquivos gráficos, adição de marcas d'água a fotos, outras modificações de imagens.
Notificação sobre a chegada de novos documentos (por exemplo, um serviço contábil distribuído publica relatórios na nuvem, e a supervisão financeira recebe notificações sobre novos relatórios, verifica e analisa-os).
Casos um pouco mais complexos envolvem, por exemplo, a formação de uma solicitação ao Kubernetes, que cria um pod com os recipientes necessários, transfere os parâmetros da tarefa para ele e, após o processamento, desativa o recipiente.

Como exemplo, vamos criar a tarefa 1, onde as mudanças no bucket do armazenamento de objetos Mail.ru Cloud Solutions (MCS) são sincronizadas com o armazenamento de objetos AWS através de webhooks. Em um caso real de alta carga, deve-se prever o trabalho assíncrono, registrando os webhooks em uma fila, mas para a tarefa de aprendizado faremos a implementação sem isso.

Esquema de operação

O protocolo de interação está detalhadamente descrito em o guia de webhooks S3 no MCS. A esquematização contém os seguintes elementos:

Serviço de publicação, que está localizado no lado do armazenamento S3 e publica solicitações HTTP quando o webhook é acionado.
Servidor de recebimento de webhooks, que escuta as solicitações do serviço de publicação via HTTP e executa as ações correspondentes. O servidor pode ser escrito em qualquer linguagem; em nosso exemplo, escreveremos o servidor em Go.

Uma característica da implementação de webhooks na API S3 é o registro do servidor de recebimento de webhooks no serviço de publicação. Em particular, o servidor de recebimento de webhooks deve confirmar a assinatura das mensagens do serviço de publicação (em outras implementações de webhooks, geralmente não é necessário confirmar a assinatura).

Consequentemente, o servidor de recebimento de webhooks deve suportar duas operações principais:

responder à solicitação do serviço de publicação para confirmar o registro,
processar os eventos recebidos.

Instalação do servidor de recebimento de webhooks

Para iniciar o servidor de recebimento de webhooks, é necessário um servidor Linux. Neste artigo, usaremos como exemplo uma instância virtual que estaremos configurando no MCS.

Vamos instalar o software necessário e iniciar o servidor de recebimento de webhooks.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Lendo listas de pacotes... Concluído
Construindo árvore de dependências
Lendo informações de estado... Concluído
Os seguintes pacotes foram instalados automaticamente e não são mais necessários:
 bc dns-root-data dnsmasq-base ebtables landscape-common liblxc-common 
liblxc1 libuv1 lxcfs lxd lxd-client python3-attr python3-automat 
python3-click python3-constantly python3-hyperlink
 python3-incremental python3-pam python3-pyasn1-modules 
python3-service-identity python3-twisted python3-twisted-bin 
python3-zope.interface uidmap xdelta3
Use 'sudo apt autoremove' para removê-los.
Pacotes sugeridos:
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
Os seguintes NOVOS pacotes serão instalados:
 git
0 atualizados, 1 recém-instalado, 0 para remover e 46 não atualizados.
Necessário baixar 3915 kB de arquivos.
Após esta operação, 32.3 MB de espaço em disco adicional serão utilizados.
Get:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Baixados 3915 kB em 1s (5639 kB/s)
Selecionando pacote git anteriormente não selecionado.
(Lendo banco de dados ... 53932 arquivos e diretórios atualmente instalados.)
Preparando para desempacotar .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Desempacotando git (1:2.17.1-1ubuntu0.7) ...
Configurando git (1:2.17.1-1ubuntu0.7) ...

Clonando a pasta com o servidor de recebimento de webhooks:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Clonando para 's3-webhook'...
remoto: Enumerando objetos: 48, feito.
remoto: Contando objetos: 100% (48/48), feito.
remoto: Comprimindo objetos: 100% (27/27), feito.
remoto: Total 114 (delta 20), reutilizado 45 (delta 18), pack-reutilizado 66
Recebendo objetos: 100% (114/114), 23.77 MiB | 20.25 MiB/s, feito.
Resolvendo deltas: 100% (49/49), feito.

Vamos iniciar o servidor:

ubuntu@ubuntu-basic-1-2-10gb:~$ cd s3-webhook/
ubantu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80

Assinatura para o serviço de publicação

Você pode registrar seu servidor de recebimento de webhooks via API ou pela interface web. Para simplificar, vamos registrar pela interface web:

Vamos para a seção de buckets no painel de controle.
Entramos no bucket para o qual vamos configurar webhooks e clicamos na engrenagem:

Vamos para a aba Webhooks e clicamos em Adicionar:

Preencha os campos:

ID — nome do webhook.

Evento — quais eventos transmitir. Nós configuramos a transmissão de todos os eventos que ocorrem ao trabalhar com arquivos (adição e remoção).

URL — endereço do servidor de recebimento de webhooks.

Prefixo/sufixo de filtro — filtro que permite gerar webhooks apenas para objetos cujos nomes correspondem a regras específicas. Por exemplo, para que o webhook dispare apenas para arquivos com a extensão .png, em Filtro sufixo você deve escrever 'png'.

No momento, apenas as portas 80 e 443 são suportadas para se conectar ao servidor de recebimento de webhooks.

Clique em Adicionar hook e veremos o seguinte:

Hook adicionado.

O servidor de recebimento de webhooks nos logs mostra o processo de registro do hook:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] requisição HTTP recebida de 
95.163.216.92:42530
2020/06/15 12:01:14 Recebido timestamp: 2020-06-15T15:01:13+03:00 TopicArn: 
mcs5259999770|myfiles-ash|s3:ObjectCreated:*,s3:ObjectRemoved:* Token: 
E2itMqAMUVVZc51pUhFWSp13DoxezvRxkUh5P7LEuk1dEe9y URL: 
http://89.208.199.220/webhook
2020/06/15 12:01:14 Gerar assinatura de resposta: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Registro concluído. Na próxima seção, vamos examinar mais detalhadamente o algoritmo de funcionamento do servidor de recebimento de webhooks.

Descrição do servidor de recebimento de webhooks

No nosso exemplo, o servidor foi escrito em Go. Vamos analisar os princípios básicos de seu funcionamento.

package main

// Gerar hmac_sha256_hex
func HmacSha256hex(message string, secret string) string {
}

// Gerar hmac_sha256
func HmacSha256(message string, secret string) string {
}

// Enviar confirmação de assinatura
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Enviar confirmação de assinatura
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Verificação de vivência
func Ping(w http.ResponseWriter, req *http.Request) {
    // log da requisição
    log.Printf("[%s] requisição HTTP Ping recebida de %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

//Webhook
func Webhook(w http.ResponseWriter, req *http.Request) {
}

func main() {

    // obter argumentos de linha de comando
    bindPort := flag.Int("port", 80, "número entre 1-65535")
    bindAddr := flag.String("address", "", "endereço IP no formato decimal")
    flag.StringVar(&actionScript, "script", "", "script externo para executar")
    flag.Parse()

    http.HandleFunc("/ping", Ping)
    http.HandleFunc("/webhook", Webhook)

log.Fatal(http.ListenAndServe(*bindAddr+":"+strconv.Itoa(*bindPort), nil))
}

Vamos considerar as principais funções:

Ping() — rota que responde na URL /ping, uma implementação simples de liveness probe.
Webhook() — rota principal, manipulador da URL /webhook:
- confirma o registro no serviço de publicação (passando para a função SubscriptionConfirmation),
- processa os webhooks recebidos (função Gotrecords).
As funções HmacSha256 e HmacSha256hex — implementações dos algoritmos de criptografia HMAC-SHA256 e HMAC-SHA256 com saída em formato de string hexadecimal para cálculo da assinatura.
main — a função principal, que processa os parâmetros da linha de comando e registra os manipuladores de URL.

Parâmetros da linha de comando aceitos pelo servidor:

-port — a porta na qual o servidor estará escutando.
-address — o endereço IP que o servidor estará escutando.
-script — programa externo que é chamado para cada webhook recebido.

Vamos analisar algumas funções em mais detalhes:

//Webhook
func Webhook(w http.ResponseWriter, req *http.Request) {

    // Read body
    body, err := ioutil.ReadAll(req.Body)
    defer req.Body.Close()
    if err != nil {
        http.Error(w, err.Error(), 500)
        return
    }

    // log request
    log.Printf("[%s] incoming HTTP request from %sn", req.Method, req.RemoteAddr)
    // check if we got subscription confirmation request
    if strings.Contains(string(body), 
""Type":"SubscriptionConfirmation"") {
        SubscriptionConfirmation(w, req, body)
    } else {
        GotRecords(w, req, body)
    }

}

Essa função determina o que foi recebido — um pedido de confirmação de registro ou um webhook. Como pode ser visto, documentação, no caso de confirmação de registro, a seguinte estrutura JSON chega na solicitação Post:

POST http://test.com HTTP/1.1
x-amz-sns-messages-type: SubscriptionConfirmation
content-type: application/json

{
    "Timestamp":"2019-12-26T19:29:12+03:00",
    "Type":"SubscriptionConfirmation",
    "Message":"Você escolheu se inscrever no tópico $topic. Para confirmar a inscrição, você precisa responder com a assinatura calculada",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":«RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA»
}

A esta solicitação, deve-se responder:

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

Onde a assinatura é calculada como:

signature = hmac_sha256(url, hmac_sha256(TopicArn, 
hmac_sha256(Timestamp, Token)))

Se um webhook chegar, então a estrutura da solicitação Post será assim:

POST  HTTP/1.1
x-amz-sns-messages-type: SubscriptionConfirmation

{ "Records":
    [
        {
            "s3": {
                "object": {
                    "eTag":"aed563ecafb4bcc5654c597a421547b2",
                    "sequencer":1577453615,
                    "key":"some-file-to-bucket",
                    "size":100
                },
            "configurationId":"1",
            "bucket": {
                "name": "bucketA",
                "ownerIdentity": {
                    "principalId":"mcs2883541269"}
                },
                "s3SchemaVersion":"1.0"
            },
            "eventVersion":"1.0",
            "requestParameters":{
                "sourceIPAddress":"185.6.245.156"
            },
            "userIdentity": {
                "principalId":"2407013e-cbc1-415f-9102-16fb9bd6946b"
            },
            "eventName":"s3:ObjectCreated:Put",
            "awsRegion":"ru-msk",
            "eventSource":"aws:s3",
            "responseElements": {
                "x-amz-request-id":"VGJR5rtJ"
            }
        }
    ]
}

Portanto, dependendo da solicitação, é necessário entender como processar os dados. Eu escolhi como indicador a gravação "Type":"SubscriptionConfirmation", pois está presente na solicitação de confirmação de assinatura e não está presente no webhook. Com base na presença/ausência deste registro na solicitação POST, a execução do programa prossegue para a função SubscriptionConfirmation, ou para a função GotRecords.

Não vamos analisar a função SubscriptionConfirmation em detalhes, ela é implementada com os princípios descritos em documentação. O código fonte desta função pode ser encontrado em repositório git do projeto.

A função GotRecords analisa a solicitação recebida e para cada objeto Record chama um script externo (cujo nome foi passado no parâmetro -script) com os seguintes parâmetros:

nome do bucket
chave do objeto
ação:
- copy — se na solicitação original EventName = ObjectCreated | PutObject | PutObjectCopy
- delete — se na solicitação original EventName = ObjectRemoved | DeleteObject

Assim, se um webhook chegou com uma solicitação POST, conforme descrito acima, e o parâmetro -script=script.sh então o script será chamado da seguinte forma:

script.sh  bucketA some-file-to-bucket copy

Deve-se entender que este servidor de recebimento de webhooks não é uma solução de produção finalizada, mas um exemplo simplificado de uma possível implementação.

Um exemplo de funcionamento

Vamos fazer a sincronização de arquivos do bucket principal no MCS para o bucket de backup na AWS. O bucket principal é chamado myfiles-ash, e o backup — myfiles-backup (a configuração do bucket na AWS está além do escopo deste artigo). Assim, quando um arquivo é colocado no bucket principal, sua cópia deve aparecer no backup, e quando é removido do principal — deve ser removido do backup.

Trabalharemos com os buckets utilizando a ferramenta awscli, que é compatível com o armazenamento em nuvem MCS e o armazenamento em nuvem AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Reading package lists... Done
Building dependency tree
Reading state information... Done
After this operation, 34.4 MB of additional disk space will be used.
Unpacking awscli (1.14.44-1ubuntu1) ...
Setting up awscli (1.14.44-1ubuntu1) ...

Vamos configurar o acesso à API S3 MCS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile mcs
AWS Access Key ID [None]: hdywEPtuuJTExxxxxxxxxxxxxx
AWS Secret Access Key [None]: hDz3SgxKwXoxxxxxxxxxxxxxxxxxx
Default region name [None]:
Default output format [None]:

Vamos configurar o acesso à API S3 AWS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile aws
AWS Access Key ID [None]: AKIAJXXXXXXXXXXXX
AWS Secret Access Key [None]: dfuerphOLQwu0CreP5Z8l5fuXXXXXXXXXXXXXXXX
Default region name [None]:
Default output format [None]:

Vamos verificar os acessos:

Para AWS:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws s3 ls --profile aws
2020-07-06 08:44:11 myfiles-backup

Para MCS, ao executar o comando deve-se adicionar —endpoint-url:

ubuntu@ubuntu-basic-1-2-10gb:~$ aws s3 ls --profile mcs --endpoint-url 
https://hb.bizmrg.com
2020-02-04 06:38:05 databasebackups-0cdaaa6402d4424e9676c75a720afa85
2020-05-27 10:08:33 myfiles-ash

Acesso obtido.

Agora vamos escrever um script para processar o webhook de entrada, chamaremos de s3_backup_mcs_aws.sh

#!/bin/bash
# Require aws cli
# if file added — copy it to backup bucket
# if file removed — remove it from backup bucket
# Variables
ENDPOINT_MCS="https://hb.bizmrg.com"
AWSCLI_MCS=`which aws`" --endpoint-url ${ENDPOINT_MCS} --profile mcs s3"
AWSCLI_AWS=`which aws`" --profile aws s3"
BACKUP_BUCKET="myfiles-backup"

SOURCE_BUCKET="${1}"
SOURCE_FILE="${2}"
ACTION="${3}"

SOURCE="s3://${SOURCE_BUCKET}/${SOURCE_FILE}"
TARGET="s3://${BACKUP_BUCKET}/${SOURCE_FILE}"
TEMP="/tmp/${SOURCE_BUCKET}/${SOURCE_FILE}"

case ${ACTION} in
    "copy")
    ${AWSCLI_MCS} cp "${SOURCE}" "${TEMP}"
    ${AWSCLI_AWS} cp "${TEMP}" "${TARGET}"
    rm ${TEMP}
    ;;

    "delete")
    ${AWSCLI_AWS} rm ${TARGET}
    ;;

    *)
    echo "Usage: ${0} sourcebucket sourcefile copy/delete"
    exit 1
    ;;
esac

Iniciando o servidor:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80 -
scripts/scripts/s3_backup_mcs_aws.sh

Vamos verificar como isso funcionará. Através do interface web do MCS vamos adicionar o arquivo test.txt ao bucket myfiles-ash. Nos logs do console, podemos ver que uma solicitação foi feita ao servidor de webhooks:

2020/07/06 09:43:08 [POST] requisição HTTP recebida de
95.163.216.92:56612
download: s3://myfiles-ash/test.txt para ../..//tmp/myfiles-ash/test.txt
upload: ../..//tmp/myfiles-ash/test.txt para
s3://myfiles-backup/test.txt

Vamos verificar o conteúdo do bucket myfiles-backup na AWS:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ aws s3 --profile aws ls
myfiles-backup
2020-07-06 09:43:10       1104 test.txt

Agora, através da interface web, vamos remover o arquivo do bucket myfiles-ash.

Logs do servidor:

2020/07/06 09:44:46 [POST] requisição HTTP recebida de
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Conteúdo do bucket:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ aws s3 --profile aws ls
myfiles-backup
ubuntu@ubuntu-basic-1-2-10gb:~$

Arquivo removido, tarefa concluída.

Conclusão e ToDo

Todo o código utilizado neste artigo está no meu repositório. Também há exemplos de scripts e exemplos de cálculo de assinaturas para registro de webhooks.

Este código é apenas um exemplo de como utilizar webhooks S3 em suas atividades. Como mencionei no início, se você pretende utilizar um servidor desse tipo em produção, é necessário, no mínimo, reescrever o servidor para operação assíncrona: registrar os webhooks recebidos em uma fila (RabbitMQ ou NATS) e, a partir dali, processá-los com aplicativos worker. Caso contrário, ao receber um grande volume de webhooks, você pode se deparar com a falta de recursos do servidor para executar as tarefas. A presença de filas permite distribuir o servidor e os workers, além de resolver questões com a repetição de tarefas em caso de falhas. Também é desejável mudar a gravação de logs para algo mais detalhado e padronizado.

Sucesso!