🥇 Esempio di applicazione event-driven basata su webhook nello storage S3 di Mail.ru Cloud Solutions

Macchina del caffè di Rube Goldberg

L'architettura event-driven aumenta l'efficienza dei costi delle risorse utilizzate, in quanto vengono attivate solo nel momento in cui sono necessarie. Ci sono molti modi per implementarlo senza creare entità cloud aggiuntive come applicazioni worker. Oggi non parlerò di FaaS, ma di webhook. Mostrerò un esempio pratico di elaborazione degli eventi utilizzando webhook nello storage object.

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

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

Creazione di copie di tutti gli oggetti in un altro archivio cloud. Le copie devono essere create "al volo", ogni volta che vengono aggiunti o modificati file.
Creazione automatica di serie di miniature di file grafici, aggiunta di filigrane alle foto e altre modifiche alle immagini.
Notifica all'arrivo di nuovi documenti (ad esempio, un servizio contabile distribuito carica nei cloud i rapporti e il monitoraggio finanziario riceve avvisi sui nuovi rapporti, li verifica e li analizza).
Casi leggermente più complessi implicano, ad esempio, la formulazione di una richiesta a Kubernetes, che crea un pod con i container necessari, trasmette i parametri del task e, dopo l'elaborazione, termina il container.

Come esempio, realizzeremo una variante del compito 1, in cui le modifiche nel bucket di storage oggetti Mail.ru Cloud Solutions (MCS) vengono sincronizzate nello storage oggetti AWS tramite webhook. In un caso reale con un elevato carico, si dovrebbe prevedere un funzionamento asincrono registrando i webhook in una coda, ma per il compito di apprendimento realizzeremo l'implementazione senza questo.

Schema di lavoro

Il protocollo di interazione è descritto in dettaglio nel manuale sui webhook S3 in MCS. Lo schema di lavoro comprende i seguenti elementi:

Servizio di pubblicazione, che si trova sul lato dello storage S3 e pubblica richieste HTTP quando viene attivato un webhook.
Server di ricezione dei webhook, che ascolta le richieste del servizio di pubblicazione tramite HTTP ed esegue le azioni di conseguenza. Il server può essere scritto in qualsiasi linguaggio; nel nostro esempio, scriveremo il server in Go.

Una particolare caratteristica dell'implementazione dei webhook nell'API S3 è la registrazione del server di ricezione dei webhook presso il servizio di pubblicazione. In particolare, il server di ricezione dei webhook deve confermare l'iscrizione ai messaggi del servizio di pubblicazione (in altre implementazioni di 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,
elaborare gli eventi in arrivo.

Installazione del server di ricezione dei webhook

Per avviare il server di ricezione dei webhook è necessaria una macchina Linux. In questo articolo utilizziamo come esempio un'istanza virtuale che distribuiamo 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
Reading package lists... Done
Building dependency tree
Reading state information... Done
The following packages were automatically installed and are no longer required:
  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' to remove them.
Suggested packages:
  git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
The following NEW packages will be installed:
  git
0 upgraded, 1 newly installed, 0 to remove and 46 not upgraded.
Need to get 3915 kB of archives.
After this operation, 32.3 MB of additional disk space will be used.
Get:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Fetched 3915 kB in 1s (5639 kB/s)
Selecting previously unselected package git.
(Reading database ... 53932 files and directories currently installed.)
Preparing to unpack .../git_1%3a2.17.1-1ubuntu0.7_amd64.deb ...
Unpacking git (1:2.17.1-1ubuntu0.7) ...
Setting up git (1:2.17.1-1ubuntu0.7) ...

Stiamo clonando la cartella del server per l'accettazione dei webhook:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Clonazione in 's3-webhook'...
remote: Enumerazione oggetti: 48, fatto.
remote: Conteggio oggetti: 100% (48/48), fatto.
remote: Compressione oggetti: 100% (27/27), fatto.
remote: Totale 114 (delta 20), riutilizzati 45 (delta 18), pacchetto riutilizzato 66
Ricezione 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/
ububuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80

Iscrizione al servizio di pubblicazione

È possibile registrare il proprio server di ricezione dei webhook tramite API o interfaccia web. Per semplificare, procederemo con la registrazione tramite l'interfaccia web:

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

Passiamo alla scheda Webhooks e facciamo clic su Aggiungi:

Compiliamo i campi:

ID — nome del webhook.

Evento — quali eventi trasmettere. Abbiamo impostato la trasmissione di tutti gli eventi che si verificano durante il lavoro con i file (aggiunta e rimozione).

URL — indirizzo del server di ricezione dei webhook.

Filtro prefisso/suffisso — filtro che consente di generare webhook solo per oggetti i cui nomi corrispondono a determinate regole. Ad esempio, per far scattare il webhook solo per i file con estensione .png, in Filtro suffisso è necessario scrivere «png».

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

Facciamo clic su Aggiungi hook e vedremo quanto segue:

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 ottenuto: 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 Firma della risposta generata: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

La registrazione è terminata. Nel prossimo capitolo esamineremo più nel dettaglio l'algoritmo di funzionamento del server per la ricezione dei webhook.

Descrizione del server di ricezione dei webhook

Nel nostro esempio, il server è scritto in Go. Esaminiamo 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 iscrizione
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

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

// Liveness probe
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 argomenti da linea di comando
    bindPort := flag.Int("port", 80, "numero tra 1-65535")
    bindAddr := flag.String("address", "", "indirizzo IP nel 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() — rotta che risponde all'URL /ping, semplice implementazione del liveness probe.
Webhook() — rotta principale, gestore dell'URL /webhook:
- conferma la registrazione al servizio di pubblicazione (passa alla funzione SubscriptionConfirmation),
- gestisce le webhook in arrivo (funzione GotRecords).
Le funzioni HmacSha256 e HmacSha256hex sono implementazioni degli algoritmi di crittografia HMAC-SHA256 e HMAC-SHA256 con output come stringhe di numeri esadecimali per il calcolo della firma.
main — la funzione principale che gestisce i parametri della riga di comando e registra i gestori degli URL.

Parametri della riga di comando accettati dal server:

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

Esaminiamo più nel dettaglio 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 se è arrivata una richiesta di conferma dell'iscrizione o un webhook. Come si evince da documentazione, in caso di conferma dell'iscrizione, la seguente struttura Json viene inviata 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 all'argomento $topic. Per confermare l'iscrizione, devi rispondere con la firma calcolata",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":«RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA»
}

A questa richiesta bisogna rispondere:

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

Dove la firma viene calcolata come:

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

Se invece arriva un webhook, la struttura della richiesta Post appare così:

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 comprendere come elaborare i dati. Ho scelto come indicatore la registrazione "Type":"SubscriptionConfirmation", poiché è presente nella richiesta di conferma dell'abbonamento e non è presente nel webhook. In base alla presenza/assenza di questa registrazione nella richiesta POST, l'esecuzione successiva del programma passa sia alla funzione SubscriptionConfirmation, sia alla funzione GotRecords.

Non approfondiremo la funzione SubscriptionConfirmation, essa è implementata secondo i principi esposti in documentazione. È possibile esaminare il codice sorgente di questa funzione in repository git del progetto.

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

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

Pertanto, se arriva un hook con POST, come descritto in precedenza, e il parametro -script=script.sh lo script sarà invocato nel modo seguente:

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

È importante comprendere che questo server di ricezione webhook non è una soluzione di produzione completa, ma un esempio semplificato di possibile implementazione.

Esempio di funzionamento

Effettueremo la sincronizzazione dei file dal bucket principale in MCS al bucket di backup in AWS. Il bucket principale si chiama myfiles-ash, quello di backup — myfiles-backup (la configurazione del bucket in AWS esula dall'argomento di questo articolo). Di conseguenza, quando un file viene posizionato nel bucket principale, una copia deve apparire in quello di backup; quando viene eliminato dal principale, deve essere eliminato anche dal backup.

Lavoreremo con i bucket utilizzando l'utilità awscli, compatibile sia con il cloud storage MCS che con il cloud storage 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) ...
Configurazione 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 [Nessuno]: hdywEPtuuJTExxxxxxxxxxxxxx
AWS Secret Access Key [Nessuno]: hDz3SgxKwXoxxxxxxxxxxxxxxxxxx
Default region name [Nessuno]:
Default output format [Nessuno]:

Configuriamo l'accesso all'API S3 AWS:

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

Controlliamo 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, quando si esegue il 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 -
scriptt scripts/s3_backup_mcs_aws.sh

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

2020/07/06 09:43:08 [POST] richiesta HTTP in entrata da 
95.163.216.92:56612
download: s3://myfiles-ash/test.txt a ..//..//..//tmp/myfiles-ash/test.txt
upload: ..//..//..//tmp/myfiles-ash/test.txt a 
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, attraverso l'interfaccia web, rimuoviamo il file dal bucket myfiles-ash.

Log del server:

2020/07/06 09:44:46 [POST] richiesta HTTP in entrata 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 rimosso, compito completato.

Conclusione e ToDo

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

Questo codice è solo un esempio di come si possono utilizzare i webhook S3 nelle proprie attività. Come ho detto all'inizio, se si prevede di utilizzare un server di questo tipo in produzione, è necessario almeno riscrivere il server per un funzionamento asincrono: registrare i webhook in arrivo in una coda (RabbitMQ o NATS) e da lì elaborarli con applicazioni worker. Altrimenti, con un afflusso massiccio di webhook, si può incorrere in una carenza di risorse del server per completare i compiti. Avere delle code consente di distribuire il server e i worker, oltre a risolvere i problemi di ripetizione delle attività in caso di guasti. È inoltre consigliabile modificare la registrazione in uno stile più dettagliato e standardizzato.

Buona fortuna!

Altre letture sull'argomento:

Fonte: habr.com