🥇Esempio di applicazione event-driven basata su webhook nell'archiviazione ad oggetti S3 di Mail.ru Cloud Solutions

Macchina da caffè di Rube Goldberg

L'architettura basata su eventi aumenta l'efficienza dei costi delle risorse utilizzate, poiché esse vengono attivate solo nel momento in cui sono necessarie. Ci sono molte soluzioni su come implementarlo senza creare entità cloud aggiuntive come applicazioni worker. E oggi non parlerò di FaaS, ma di webhook. Mostrerò un esempio pratico di elaborazione degli eventi tramite webhook di uno storage oggetti.

Due parole sullo storage oggetti e sui webhook. Gli storage oggetti consentono di memorizzare qualsiasi dato nel cloud sotto forma di oggetti, accessibili tramite S3 o un'altra API (a seconda di come è implementata) tramite HTTP/HTTPS. I webhook (webhook) sono in generale callback personalizzati su HTTP. Di solito vengono attivati da un evento, come l'invio di codice in un repository o un commento pubblicato in un blog. Quando si verifica un evento, il sito di origine invia una richiesta HTTP all'URL specificato per il webhook. Di conseguenza, è possibile fare in modo che eventi su un sito attivino azioni su un altro. Nel caso in cui il sito di origine sia uno storage oggetti, le modifiche al suo contenuto fungono da eventi.wiki.

Esempi di casi semplici in cui è possibile utilizzare tale automazione:

Creazione di copie di tutti gli oggetti in un altro storage cloud. Le copie devono essere create "al volo", a ogni aggiunta o modifica di file.
Creazione automatica di serie di miniature di file grafici, aggiunta di filigrane alle fotografie, altre modifiche delle immagini.
Notifica all'arrivo di nuovi documenti (ad esempio, un servizio contabile distribuito carica report nel cloud e un monitoraggio finanziario riceve notifiche sui nuovi report, li verifica e li analizza).
Casi un po' più complessi prevedono, ad esempio, la generazione di una richiesta a Kubernetes che crea un pod con i contenitori necessari, passa i parametri del task e, dopo l'elaborazione, riduce il contenitore.

Come esempio, realizzeremo una variante del task 1, in cui le modifiche nel bucket dello storage oggetti Mail.ru Cloud Solutions (MCS) vengono sincronizzate nello storage oggetti AWS tramite webhook. In un caso reale ad alto carico, si dovrebbe prevedere il funzionamento asincrono registrando i webhook in coda, ma per questa attività didattica faremo un'implementazione senza di esso.

Schema di lavoro

Il protocollo di interazione è descritto in dettaglio in guida agli webhook S3 su MCS. Nella schema di lavoro ci sono i seguenti elementi:

Servizio di pubblicazione, che si trova lato S3 e pubblica richieste HTTP quando viene attivato il webhook.
Server di ricezione dei webhook, che ascolta le richieste dal servizio di pubblicazione via HTTP e compie le azioni corrispondenti. Il server può essere scritto in qualsiasi linguaggio; nel nostro esempio scriveremo il server in Go.

Una caratteristica dell'implementazione dei webhook nell'S3 API è la registrazione del server di ricezione dei webhook presso il servizio di pubblicazione. In particolare, il server di ricezione deve confermare l'iscrizione ai messaggi del servizio di pubblicazione (in altre implementazioni dei webhook di solito non è richiesta la conferma dell'iscrizione).

Pertanto, il server di ricezione dei webhook deve supportare due operazioni principali:

rispondere alla richiesta del servizio di pubblicazione per confermare la registrazione,
gestire gli eventi in arrivo.

Installazione del server di ricezione dei webhook

Per avviare il server di ricezione dei webhook è necessario un server Linux. In questo articolo, per esempio, utilizziamo un'istanza virtuale che deployiamo su MCS.

Installeremo il software necessario e avvieremo il server di ricezione dei webhook.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Lettura delle liste dei pacchetti... Fatto
Creazione dell'albero delle dipendenze
Lettura delle informazioni di stato... Fatto
I seguenti pacchetti sono stati installati automaticamente e non sono più necessari:
 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
Utilizza 'sudo apt autoremove' per rimuoverli.
Pacchetti suggeriti:
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
I seguenti NUOVI pacchetti verranno installati:
 git
0 aggiornati, 1 appena installato, 0 da rimuovere e 46 non aggiornati.
È necessario scaricare 3915 kB di archivi.
Dopo questa operazione, verranno utilizzati 32.3 MB di spazio su disco aggiuntivo.
Ottieni:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Scaricati 3915 kB in 1s (5639 kB/s)
Selezionando il pacchetto git precedentemente non selezionato.
(Lettura del database... 53932 file e directory attualmente installati.)
Preparazione per estrarre .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Estrazione di git (1:2.17.1-1ubuntu0.7) ...
Configurazione di git (1:2.17.1-1ubuntu0.7) ...

Cloniamo la cartella con il server di ricezione dei webhook:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Clonazione in 's3-webhook'...
remoto: Enumerazione degli oggetti: 48, fatto.
remoto: Conteggio degli oggetti: 100% (48/48), fatto.
remoto: Compressione degli oggetti: 100% (27/27), fatto.
remoto: Totale 114 (delta 20), riutilizzati 45 (delta 18), pacchetto riutilizzato 66
Ricezione degli oggetti: 100% (114/114), 23.77 MiB | 20.25 MiB/s, fatto.
Risoluzione dei delta: 100% (49/49), fatto.

Avviamo il server:

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

Abbonamento al servizio di pubblicazione

È possibile registrare il proprio server di ricezione dei webhook tramite API o interfaccia web. Per semplicità, registriamo tramite l'interfaccia web:

Andiamo nella sezione dei bucket nel pannello di controllo.
Accediamo al bucket per il quale configureremo i webhook e premiamo sull'ingranaggio:

Passiamo alla scheda Webhooks e clicchiamo Aggiungi:

Compiliamo i campi:

ID — nome del webhook.

Event — quali eventi trasmettere. Abbiamo impostato la trasmissione di tutti gli eventi che si verificano durante l'interazione con i file (aggiunta e rimozione).

URL — indirizzo del server di ricezione dei webhook.

Filter prefix/suffix — filtro che consente di generare webhook solo per oggetti i cui nomi corrispondono a determinate regole. Ad esempio, affinché il webhook scatti solo per i file con estensione .png, in Filter suffix deve essere scritto "png".

Attualmente, sono supportate solo le porte 80 e 443 per la connessione al server di ricezione dei webhook.

Clicchiamo Aggiungi hook e vedremo il seguente messaggio:

Hook aggiunto.

Il server di ricezione dei webhook nei log mostra il processo di registrazione dell'hook:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] richiesta HTTP in arrivo da 
95.163.216.92:42530
2020/06/15 12:01:14 Timestamp ricevuto: 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 Genera la firma di risposta: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Registrazione completata. Nella prossima sezione esamineremo più nel dettaglio l'algoritmo di funzionamento del server di ricezione dei webhook.

Descrizione del server di ricezione dei webhook

Nel nostro esempio, il server è scritto in Go. Analizziamo i principi fondamentali del suo funzionamento.

package main

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

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

// Invia conferma di abbonamento
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Invia conferma di abbonamento
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Prova di vitalità
func Ping(w http.ResponseWriter, req *http.Request) {
    // registra la richiesta
    log.Printf("[%s] richiesta HTTP Ping in arrivo da %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

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

func main() {

    // ottieni gli argomenti della riga di comando
    bindPort := flag.Int("port", 80, "numero tra 1-65535")
    bindAddr := flag.String("address", "", "indirizzo IP in formato dot")
    flag.StringVar(&actionScript, "script", "", "script esterno da eseguire")
    flag.Parse()

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

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

Esaminiamo le funzioni principali:

Ping() — router che risponde all'URL/ping, implementazione più semplice del liveness probe.
Webhook() — router principale, gestore dell'URL/webhook:
- conferma la registrazione sul servizio di pubblicazione (passaggio alla funzione SubscriptionConfirmation),
- gestisce i webhook in arrivo (funzione Gotrecords).
Le funzioni HmacSha256 e HmacSha256hex — implementazioni degli algoritmi di crittografia HMAC-SHA256 e HMAC-SHA256 con output in forma di stringa di numeri esadecimali per il calcolo della firma.
main — funzione principale, gestisce i parametri della riga di comando e registra i gestori dell'URL.

Parametri della riga di comando accettati dal server:

-port — la porta su cui il server ascolterà.
-address — l'indirizzo IP che il server ascolterà.
-script — programma esterno che viene chiamato per ogni webhook in arrivo.

Esaminiamo più da vicino alcune funzioni:

//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)
    }

}

Questa funzione determina cosa è arrivato — una richiesta di conferma di registrazione oppure un webhook. Come indicato da documentazione, nel caso di conferma di registrazione, la seguente struttura Json arriva nella richiesta 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":"Hai scelto di iscriverti al tema $topic. Per confermare l'iscrizione devi rispondere con la firma calcolata",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":"RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA"
}

A questa richiesta è necessario rispondere:

content-type: application/json

{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}

Dove la firma è calcolata come:

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

Se arriva un webhook, la struttura della richiesta Post sarà la seguente:

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"
            }
        }
    ]
}

Pertanto, a seconda della richiesta è necessario capire come elaborare i dati. Ho scelto come indicatore la registrazione "Type":"SubscriptionConfirmation", poiché è presente nella richiesta di conferma dell'iscrizione e non è presente nel webhook. In base alla presenza/assenza di questa registrazione nella richiesta POST, l'esecuzione successiva del programma passa o alla funzione SubscriptionConfirmation, oppure alla funzione GotRecords.

Non approfondiremo la funzione SubscriptionConfirmation, essa è implementata secondo i principi esposti in documentazione. Il codice sorgente di questa funzione può essere esaminato in repository git del progetto.

La funzione GotRecords analizza la richiesta in arrivo e per ogni oggetto Record chiama uno script esterno (il nome del quale è stato passato come parametro -script) con i parametri:

nome del bucket
chiave dell'oggetto
azione:
- copy — se nella richiesta originale EventName = ObjectCreated | PutObject | PutObjectCopy
- delete — se nella richiesta originale EventName = ObjectRemoved | DeleteObject

Pertanto, se viene ricevuto un hook con una richiesta Post, come descritto sopra, e il parametro -script=script.sh, allora lo script sarà chiamato nel seguente modo:

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

Va compreso che questo server di ricezione webhook non è una soluzione di produzione definitiva, ma un esempio semplificato di una possibile implementazione.

Esempio di funzionamento

Eseguiamo la sincronizzazione dei file dal bucket principale in MCS nel bucket di backup in AWS. Il bucket principale si chiama myfiles-ash, il backup è myfiles-backup (la configurazione del bucket in AWS esula da questo articolo). Pertanto, quando un file viene caricato nel bucket principale, la sua copia dovrebbe apparire nel backup, quando viene rimosso dal principale — deve essere rimosso anche dal backup.

Lavoreremo con i bucket utilizzando l'utility awscli, compatibile sia con lo storage cloud MCS che con AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Lettura delle liste dei pacchetti... Fatto
Costruzione dell'albero delle dipendenze
Lettura delle informazioni di stato... Fatto
Dopo questa operazione, verranno utilizzati 34,4 MB di spazio su disco aggiuntivo.
Estrazione di awscli (1.14.44-1ubuntu1) ...
Impostazione di awscli (1.14.44-1ubuntu1) ...

Configuriamo l'accesso all'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]:

Configuriamo l'accesso all'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]:

Verifichiamo gli accessi:

A AWS:

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

Per MCS, durante il funzionamento del comando è necessario aggiungere —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

Accesso ottenuto.

Ora scriviamo uno script per gestire il webhook in arrivo, lo chiameremo 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

Avviamo il server:

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

Controlliamo come funzionerà. Tramite l'interfaccia web MCS aggiungeremo il file test.txt al bucket myfiles-ash. Nei log sulla console si vede che è stata fatta una richiesta al server dei webhook:

2020/07/06 09:43:08 [POST] richiesta HTTP in arrivo da 
95.163.216.92:56612
download: s3://myfiles-ash/test.txt to ../../../../tmp/myfiles-ash/test.txt
upload: ../../../../tmp/myfiles-ash/test.txt to 
s3://myfiles-backup/test.txt

Controlliamo il contenuto del bucket myfiles-backup in 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

Ora tramite l'interfaccia web eliminiamo il file dal bucket myfiles-ash.

Log del server:

2020/07/06 09:44:46 [POST] richiesta HTTP in arrivo da 
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Contenuto del bucket:

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

File eliminato, compito completato.

Conclusione e ToDo

Tutto il codice utilizzato in questo articolo si trova nel mio repository. Ci sono anche esempi di script e esempi di calcolo delle firme per la registrazione dei webhook.

Questo codice è solo un esempio di come utilizzare i webhook S3 nella propria attività. Come ho detto in precedenza, se si prevede di utilizzare un server di questo tipo in produzione, è necessario almeno riscrivere il server per il lavoro asincrono: registrare i webhook in arrivo in una coda (RabbitMQ o NATS), da cui verranno poi elaborati dalle applicazioni worker. Altrimenti, in caso di arrivi massicci di webhook, si potrebbero incontrare carenze nelle risorse del server per l'esecuzione dei compiti. La presenza di code consente di distribuire il server e i worker, nonché di risolvere questioni relative alla ripetizione dei compiti in caso di errori. È anche consigliabile cambiare la registrazione in qualcosa di più dettagliato e standardizzato.

Buona fortuna!

Ulteriori letture sull'argomento:

Fonte: habr.com