Exemple d'application événementielle basée sur des webhooks dans le stockage d'objet S3 de Mail.ru Cloud Solutions

Machine à café de Rube Goldberg

L'architecture orientée événements améliore l'efficacité des ressources utilisées, car elles ne sont mobilisées que lorsque c'est nécessaire. Il existe de nombreuses façons de le réaliser sans créer d'entités cloud supplémentaires comme des applications worker. Aujourd'hui, je ne vais pas parler de FaaS, mais des webhooks. Je vais montrer un exemple d'apprentissage du traitement des événements à l'aide de webhooks de stockage d'objets.

Quelques mots sur le stockage d'objets et les webhooks. Les stockages d'objets permettent de conserver des données dans le cloud sous forme d'objets, accessibles via S3 ou une autre API (selon l'implémentation) par HTTP/HTTPS. Les webhooks, en général, sont des appels de retour personnalisés par HTTP. Ils sont généralement déclenchés par un événement, comme l'envoi de code dans un dépôt ou un commentaire publié sur un blog. Lorsque l'événement se produit, le site source envoie une requête HTTP à l'URL spécifiée pour le webhook. En conséquence, il est possible de faire en sorte que des événements sur un site entraînent des actions sur un autre site. Dans le cas où le site source est un stockage d'objets, les changements de son contenu jouent le rôle d'événements.wiki). Dans le cas où le site source est un stockage objet, les événements sont les modifications de son contenu.

Exemples de cas simples où l'on peut utiliser cette automatisation :

1. Création de copies de tous les objets dans un autre stockage cloud. Les copies doivent être créées « à la volée », lors de l'ajout ou de la modification de fichiers.
2. Création automatique de séries de miniatures de fichiers graphiques, ajout de filigranes sur des photos, autres modifications d'images.
3. Notification de l'arrivée de nouveaux documents (par exemple, un service de comptabilité distribué met des rapports dans le cloud, et un service de surveillance financière reçoit des notifications sur les nouveaux rapports, les vérifie et les analyse).
4. Des cas légèrement plus complexes impliquent, par exemple, la génération d'une requête à Kubernetes, qui crée un pod avec les conteneurs requis, y transfère les paramètres de tâche et, après traitement, ferme le conteneur.

Comme exemple, nous allons réaliser la tâche 1, où les modifications dans le compartiment de stockage d'objets Mail.ru Cloud Solutions (MCS) sont synchronisées via des webhooks dans le stockage d'objets AWS. Dans un cas chargé réel, il faudrait prévoir un fonctionnement asynchrone en enregistrant les webhooks dans une file d'attente, mais pour cette tâche d'apprentissage, nous allons implémenter sans cela.

Schéma de fonctionnement

Le protocole d'interaction est décrit en détail dans le guide des webhooks S3 sur MCS. Dans le schéma de fonctionnement, les éléments suivants sont présents :

• Service de publication, qui se trouve du côté du stockage S3 et publie des requêtes HTTP lorsque le webhook se déclenche.
• Serveur de réception des webhooks, qui écoute les requêtes du service de publication via HTTP et exécute les actions correspondantes. Le serveur peut être écrit dans n'importe quel langage, dans notre exemple, nous allons écrire le serveur en Go.

Une particularité de la mise en œuvre des webhooks dans l'API S3 est l'enregistrement du serveur de réception des webhooks sur le service de publication. En particulier, le serveur de réception des webhooks doit confirmer son abonnement aux messages du service de publication (dans d'autres mises en œuvre de webhooks, la confirmation de l'abonnement n'est généralement pas requise).

Par conséquent, le serveur de réception des webhooks doit prendre en charge deux opérations principales :

• répondre à la requête du service de publication confirmant l'enregistrement,
• traiter les événements entrants.

Installation du serveur de réception des webhooks

Pour démarrer le serveur de réception des webhooks, un serveur Linux est nécessaire. Dans cet article, nous allons utiliser un exemple d'instance virtuelle déployée sur MCS.

Installons le logiciel nécessaire et lançons le serveur de réception des webhooks.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Lecture des listes de paquets... Fait
Construction de l'arbre des dépendances
Lecture des états d'information... Fait
Les paquets suivants ont été installés automatiquement et ne sont plus requis :
 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
Utilisez 'sudo apt autoremove' pour les supprimer.
Paquets suggérés :
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
Les nouveaux paquets suivants seront installés :
 git
0 mis à jour, 1 nouvellement installé, 0 à supprimer et 46 non mis à jour.
Besoin de récupérer 3915 kB d'archives.
Après cette opération, 32.3 Mo d'espace disque supplémentaires seront utilisés.
Get:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Récupéré 3915 kB en 1s (5639 kB/s)
Sélection du paquet git précédemment non sélectionné.
(Lecture de la base de données ... 53932 fichiers et répertoires actuellement installés.)
Préparation du dépaquetage .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Dépaquetage de git (1:2.17.1-1ubuntu0.7) ...
Configuration de git (1:2.17.1-1ubuntu0.7) ...

Clonons le dossier avec le serveur de réception des webhooks :

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Clonage dans 's3-webhook'...
remote: Énumération des objets : 48, fait.
remote: Comptage des objets : 100 % (48/48), fait.
remote: Compression des objets : 100 % (27/27), fait.
remote: Total 114 (delta 20), réutilisé 45 (delta 18), pack-réutilisé 66
Réception des objets : 100 % (114/114), 23.77 MiB | 20.25 MiB/s, fait.
Résolution des deltas : 100 % (49/49), fait.

Lançons le serveur :

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

Abonnement au service de publication

Vous pouvez enregistrer votre serveur de réception de webhooks via l'API ou l'interface web. Pour simplifier, nous allons enregistrer via l'interface web :

1. Allons dans la section des buckets dans le tableau de bord de gestion.
2. Accédez au bucket pour lequel nous allons configurer les webhooks et cliquez sur l'icône en forme de roue dentée :

Allez à l'onglet Webhooks et cliquez sur Ajouter :

Remplissez les champs :

ID — nom du webhook.

Événement — quels événements transmettre. Nous avons configuré la transmission de tous les événements liés aux fichiers (ajouts et suppressions).

URL — adresse du serveur de réception de webhooks.

Filtre préfixe/suffixe — filtre permettant de générer des webhooks uniquement pour les objets dont les noms répondent à des critères spécifiques. Par exemple, pour que le webhook ne s'active que pour les fichiers avec l'extension .png, dans filtre suffixe il faut écrire «png».

Pour le moment, seuls les ports 80 et 443 sont pris en charge pour accéder au serveur de réception de webhooks.

Cliquons sur Ajouter hook et nous verrons ce qui suit :

Hook ajouté.

Le serveur de réception de webhooks montre dans les journaux le processus d'enregistrement du hook :

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] requête HTTP entrante de 
95.163.216.92:42530
2020/06/15 12:01:14 Timestamp reçu : 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 Signature de réponse générée : 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Enregistrement terminé. Dans la section suivante, nous examinerons plus en détail l'algorithme de fonctionnement du serveur de réception de webhooks.

Description du serveur de réception de webhooks

Dans notre exemple, le serveur est écrit en Go. Examinons les principes de base de son fonctionnement.

package main

// Générer hmac_sha256_hex
func HmacSha256hex(message string, secret string) string {
}

// Générer hmac_sha256
func HmacSha256(message string, secret string) string {
}

// Envoyer confirmation d'abonnement
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Envoyer confirmation d'abonnement
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Liveness probe
func Ping(w http.ResponseWriter, req *http.Request) {
    // log demande
    log.Printf("[%s] requête HTTP Ping entrante de %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

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

func main() {

    // obtenir les arguments de ligne de commande
    bindPort := flag.Int("port", 80, "nombre entre 1-65535")
    bindAddr := flag.String("address", "", "adresse IP au format point")
    flag.StringVar(&actionScript, "script", "", "script externe à exécuter")
    flag.Parse()

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

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

Examinons les principales fonctionnalités :

• Ping() — route qui répond à l'URL /ping, une implémentation simple de l'outil de vérification de disponibilité.
• Webhook() — route principale, gestionnaire de l'URL /webhook :
  • confirme l'enregistrement sur le service de publication (passage à la fonction SubscriptionConfirmation),
  • traiter les webhooks entrants (fonction Gotrecords).
• Les fonctions HmacSha256 et HmacSha256hex — implémentations des algorithmes de cryptage HMAC-SHA256 et HMAC-SHA256 avec sortie sous forme de chaîne de chiffres hexadécimaux pour le calcul de la signature.
• main — fonction principale, traite les paramètres de ligne de commande et enregistre les gestionnaires d'URL.

Paramètres de ligne de commande acceptés par le serveur :

• -port — le port sur lequel le serveur écoutera.
• -address — l'adresse IP que le serveur écoutera.
• -script — programme externe qui est appelé pour chaque webhook reçu.

Examinons plus en détail certaines fonctions :

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

}

Cette fonction détermine ce qui est arrivé — une demande de confirmation d'enregistrement ou un webhook. Comme le montre documentation, en cas de confirmation d'enregistrement, la structure Json suivante arrive dans la requête 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":"Vous avez choisi de vous abonner au sujet $topic. Pour confirmer l'abonnement, vous devez répondre avec la signature calculée",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":«RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA»
}

À cette requête, il faut répondre :

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

Où la signature est calculée comme suit :

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

S'il s'agit d'un webhook, alors la structure de la requête Post ressemble à ceci :

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

Ainsi, en fonction de la requête, il faut comprendre comment traiter les données. J'ai choisi comme indicateur l'enregistrement "Type":"SubscriptionConfirmation", car elle est présente dans la demande de confirmation d'abonnement et n'est pas présente dans le webhook. Selon la présence/absence de cet enregistrement dans la requête POST, l'exécution ultérieure du programme passe soit à la fonction SubscriptionConfirmation, soit à la fonction GotRecords.

Nous n'examinerons pas en détail la fonction SubscriptionConfirmation, elle est implémentée selon les principes énoncés dans documentation. Le code source de cette fonction peut être consulté dans le référentiel git du projet.

La fonction GotRecords analyse la requête entrante et pour chaque objet Record appelle un script externe (dont le nom a été fourni dans le paramètre -script) avec les paramètres :

• nom du bucket
• clé de l'objet
• action :
  • copy — si dans la requête d'origine, EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete — si dans la requête d'origine, EventName = ObjectRemoved | DeleteObject

Ainsi, si un webhook arrive avec une requête POST, comme décrit supérieur, et que le paramètre -script=script.sh alors le script sera appelé de la manière suivante :

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

Il faut comprendre que ce serveur de réception de webhooks n'est pas une solution de production complète, mais un exemple simplifié d'une implémentation possible.

Exemple de fonctionnement

Nous allons synchroniser les fichiers du bucket principal dans MCS avec le bucket de sauvegarde dans AWS. Le bucket principal s'appelle myfiles-ash, le bucket de sauvegarde — myfiles-backup (la configuration du bucket dans AWS dépasse le cadre de cet article). Ainsi, lorsque un fichier est placé dans le bucket principal, sa copie doit apparaître dans le bucket de sauvegarde, et lorsqu'il est supprimé du principal, il doit être supprimé du sauvegarde.

Nous utiliserons l'outil awscli pour travailler avec les buckets, qui est compatible à la fois avec le stockage cloud MCS et le stockage cloud AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Lecture des listes de paquets... Fait
Construction de l'arbre des dépendances
Lecture des informations d'état... Fait
Après cette opération, 34,4 Mo d'espace disque supplémentaire seront utilisés.
Dépaquetage de awscli (1.14.44-1ubuntu1) ...
Configuration de awscli (1.14.44-1ubuntu1) ...

Configurons l'accès à l'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
Nom de la région par défaut [None]:
Format de sortie par défaut [None]:

Configurons l'accès à l'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
Nom de la région par défaut [None]:
Format de sortie par défaut [None]:

Vérifions l'accès :

À AWS :

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

Pour MCS, lors de l'exécution de la commande, il faut ajouter —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

Accès obtenu.

Nous allons maintenant écrire un script pour traiter le webhook entrant, que nous appellerons 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

Démarrons le serveur :

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

Vérifions comment cela fonctionnera. Via l'interface web MCS nous allons ajouter le fichier test.txt dans le bucket myfiles-ash. Dans les logs de la console, on peut voir qu'une requête a été faite au serveur webhook :

2020\/07\/06 09:43:08 [POST] requête HTTP entrante de 
95.163.216.92:56612
téléchargement : s3:\/\/myfiles-ash\/test.txt vers ..\/..\/..\/tmp\/myfiles-ash\/test.txt
upload : ..\/..\/..\/tmp\/myfiles-ash\/test.txt vers 
s3:\/\/myfiles-backup\/test.txt

Vérifions le contenu du bucket myfiles-backup dans 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

Nous allons maintenant supprimer le fichier du bucket myfiles-ash via l'interface web.

Logs du serveur :

2020\/07\/06 09:44:46 [POST] requête HTTP entrante de 
95.163.216.92:58224
délétion : s3:\/\/myfiles-backup\/test.txt

Contenu du bucket :

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

Le fichier a été supprimé, la tâche est accomplie.

Conclusion et ToDo

Tout le code utilisé dans cet article se trouve dans mon dépôt. Vous y trouverez également des exemples de scripts et des exemples de calculs de signatures pour l'enregistrement des webhooks.

Ce code n'est qu'un exemple de la façon dont vous pouvez utiliser les webhooks S3 dans votre activité. Comme je l'ai mentionné au début, si vous prévoyez d'utiliser un tel serveur en production, il est impératif de réécrire le serveur pour un fonctionnement asynchrone : enregistrez les webhooks entrants dans une file d'attente (RabbitMQ ou NATS), puis traitez-les avec des applications worker. Sinon, lors d'une arrivée massive de webhooks, vous pourriez être confronté à un manque de ressources serveur pour exécuter les tâches. Disposer de files d'attente permet de répartir le serveur et les workers, et de gérer les questions de réexécution des tâches en cas d'échecs. Il est également souhaitable d'améliorer la journalisation en la rendant plus détaillée et plus standardisée.

Bonne chance !

À lire également sur le sujet :

• Travailler avec le stockage objet S3 de Mail.ru Cloud Solutions comme avec un système de fichiers.
• Comment utiliser le stockage objet pour devancer la concurrence.

Source : habr.com