Ein Beispiel für eine ereignisgesteuerte Anwendung basierend auf Webhooks im S3-Objektspeicher Mail.ru Cloud Solutions

Rube Goldberg Kaffeemaschine

Eine ereignisgesteuerte Architektur erhöht die Kosteneffizienz der eingesetzten Ressourcen, da diese nur dann genutzt werden, wenn sie benötigt werden. Es gibt viele Möglichkeiten, dies zu implementieren und keine zusätzlichen Cloud-Entitäten als Worker-Anwendungen zu erstellen. Und heute werde ich nicht über FaaS sprechen, sondern über Webhooks. Ich zeige ein Tutorial-Beispiel für die Verarbeitung von Ereignissen mithilfe von Objektspeicher-Webhooks.

Ein paar Worte zu Objektspeicher und Webhooks. Mit Objektspeicher können Sie beliebige Daten in der Cloud in Form von Objekten speichern, auf die über S3 oder eine andere API (je nach Implementierung) über HTTP/HTTPS zugegriffen werden kann. Webhooks sind im Allgemeinen benutzerdefinierte HTTP-Rückrufe. Sie werden in der Regel durch ein Ereignis ausgelöst, beispielsweise durch die Übertragung von Code in ein Repository oder durch die Veröffentlichung eines Kommentars in einem Blog. Wenn ein Ereignis auftritt, sendet die Ursprungssite eine HTTP-Anfrage an die für den Webhook angegebene URL. Dadurch können Sie Ereignisse auf einer Site dazu bringen, Aktionen auf einer anderen auszulösen (Wiki). Wenn es sich bei der Quellsite um einen Objektspeicher handelt, fungieren Ereignisse als Änderungen an dessen Inhalten.

Beispiele für einfache Fälle, in denen eine solche Automatisierung verwendet werden kann:

1. Erstellen von Kopien aller Objekte in einem anderen Cloud-Speicher. Kopien müssen im Handumdrehen erstellt werden, wenn Dateien hinzugefügt oder geändert werden.
2. Automatische Erstellung einer Reihe von Miniaturansichten von Grafikdateien, Hinzufügen von Wasserzeichen zu Fotos und andere Bildänderungen.
3. Benachrichtigung über den Eingang neuer Dokumente (z. B. lädt ein verteilter Buchhaltungsdienst Berichte in die Cloud hoch und die Finanzüberwachung empfängt Benachrichtigungen über neue Berichte, prüft und analysiert diese).
4. Etwas komplexere Fälle umfassen beispielsweise das Generieren einer Anfrage an Kubernetes, das einen Pod mit den erforderlichen Containern erstellt, Aufgabenparameter an ihn übergibt und nach der Verarbeitung den Container zusammenfaltet.

Als Beispiel erstellen wir eine Variante von Aufgabe 1, bei der Änderungen im Objektspeicher-Bucket von Mail.ru Cloud Solutions (MCS) mithilfe von Webhooks im AWS-Objektspeicher synchronisiert werden. In einem real geladenen Fall sollte asynchrone Arbeit durch die Registrierung von Webhooks in einer Warteschlange bereitgestellt werden, aber für die Trainingsaufgabe werden wir die Implementierung ohne dies durchführen.

Arbeitsweise

Das Interaktionsprotokoll ist ausführlich beschrieben in Leitfaden für S3-Webhooks auf MCS. Das Arbeitsschema enthält folgende Elemente:

• Verlagsservice, das sich auf der S3-Speicherseite befindet und HTTP-Anfragen veröffentlicht, wenn der Webnhook ausgelöst wird.
• Webhook-Empfangsserver, das auf Anfragen des HTTP-Veröffentlichungsdienstes lauscht und entsprechende Aktionen ausführt. Der Server kann in jeder Sprache geschrieben werden; in unserem Beispiel schreiben wir den Server in Go.

Eine Besonderheit der Implementierung von Webhooks in der S3 API ist die Registrierung des Webhook-Empfangsservers beim Publishing-Dienst. Insbesondere muss der Webhook-Empfangsserver das Abonnement von Nachrichten vom Veröffentlichungsdienst bestätigen (in anderen Webhook-Implementierungen ist eine Bestätigung des Abonnements normalerweise nicht erforderlich).

Dementsprechend muss der Webhook-Empfangsserver zwei Hauptoperationen unterstützen:

• auf die Anfrage des Verlagsdienstes zur Bestätigung der Registrierung antworten,
• eingehende Ereignisse verarbeiten.

Installieren eines Webhook-Empfangsservers

Um den Webhook-Empfangsserver auszuführen, benötigen Sie einen Linux-Server. In diesem Artikel verwenden wir als Beispiel eine virtuelle Instanz, die wir auf MCS bereitstellen.

Lassen Sie uns die erforderliche Software installieren und den Webhook-Empfangsserver starten.

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) ...

Klonen Sie den Ordner mit dem Webhook-Empfangsserver:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Cloning into 's3-webhook'...
remote: Enumerating objects: 48, done.
remote: Counting objects: 100% (48/48), done.
remote: Compressing objects: 100% (27/27), done.
remote: Total 114 (delta 20), reused 45 (delta 18), pack-reused 66
Receiving objects: 100% (114/114), 23.77 MiB | 20.25 MiB/s, done.
Resolving deltas: 100% (49/49), done.

Starten wir den Server:

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

Abonnieren Sie den Veröffentlichungsdienst

Sie können Ihren Webhook-Empfangsserver über die API oder das Webinterface registrieren. Der Einfachheit halber registrieren wir uns über die Weboberfläche:

1. Gehen wir zum Abschnitt „Eimer“. im Kontrollraum.
2. Gehen Sie zu dem Bucket, für den wir Webhooks konfigurieren, und klicken Sie auf das Zahnrad:

Gehen Sie zur Registerkarte Webhooks und klicken Sie auf Hinzufügen:

Füllen Sie die Felder aus:

ID – der Name des Webhooks.

Ereignis – welche Ereignisse übertragen werden sollen. Wir haben die Übertragung aller Ereignisse eingestellt, die beim Arbeiten mit Dateien (Hinzufügen und Löschen) auftreten.

URL – Adresse des Webhook-Empfangsservers.

Filterpräfix/Suffix ist ein Filter, mit dem Sie Webhooks nur für Objekte generieren können, deren Namen bestimmten Regeln entsprechen. Damit der Webhook beispielsweise nur Dateien mit der Erweiterung .png auslöst, in Filtersuffix Sie müssen „png“ schreiben.

Derzeit werden nur die Ports 80 und 443 für den Zugriff auf den Webhook-Empfangsserver unterstützt.

Lasst uns klicken Haken hinzufügen und wir werden Folgendes sehen:

Haken hinzugefügt.

Der Webhook-Empfangsserver zeigt in seinen Protokollen den Fortschritt des Hook-Registrierungsprozesses an:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] incoming HTTP request from 
95.163.216.92:42530
2020/06/15 12:01:14 Got 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 Generate responce signature: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Die Registrierung ist abgeschlossen. Im nächsten Abschnitt werden wir uns den Betriebsalgorithmus des Webhook-Empfangsservers genauer ansehen.

Beschreibung des Webhook-Empfangsservers

In unserem Beispiel ist der Server in Go geschrieben. Schauen wir uns die Grundprinzipien seiner Funktionsweise an.

package main

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

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

// Send subscription confirmation
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Send subscription confirmation
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Liveness probe
func Ping(w http.ResponseWriter, req *http.Request) {
    // log request
    log.Printf("[%s] incoming HTTP Ping request from %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

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

func main() {

    // get command line args
    bindPort := flag.Int("port", 80, "number between 1-65535")
    bindAddr := flag.String("address", "", "ip address in dot format")
    flag.StringVar(&actionScript, "script", "", "external script to execute")
    flag.Parse()

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

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

Betrachten Sie die Hauptfunktionen:

• Ping() – eine Route, die über URL/Ping antwortet, die einfachste Implementierung einer Liveness-Prüfung.
• Webhook() – Hauptroute, URL/Webhook-Handler:
  • bestätigt die Registrierung beim Veröffentlichungsdienst (gehen Sie zur Funktion „Abonnementbestätigung“),
  • verarbeitet eingehende Webhooks (Gorecords-Funktion).
• Die Funktionen HmacSha256 und HmacSha256hex sind Implementierungen der Verschlüsselungsalgorithmen HMAC-SHA256 und HMAC-SHA256 mit Ausgabe als Folge von Hexadezimalzahlen zur Berechnung der Signatur.
• main ist die Hauptfunktion, verarbeitet Befehlszeilenparameter und registriert URL-Handler.

Vom Server akzeptierte Befehlszeilenparameter:

• -port ist der Port, den der Server abhört.
• -address – IP-Adresse, die der Server abhört.
• -script ist ein externes Programm, das für jeden eingehenden Hook aufgerufen wird.

Schauen wir uns einige der Funktionen genauer an:

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

}

Diese Funktion ermittelt, ob eine Anfrage zur Bestätigung der Registrierung oder ein Webhook eingetroffen ist. Wie folgt aus Dokumentation, wenn die Registrierung bestätigt wird, wird die folgende Json-Struktur in der Post-Anfrage empfangen:

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":"You have chosen to subscribe to the topic $topic. To confirm the subscription you need to response with calculated signature",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":«RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA»
}

Diese Frage muss beantwortet werden:

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

Wobei die Signatur wie folgt berechnet wird:

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

Wenn ein Webhook eintrifft, sieht die Struktur der Post-Anfrage wie folgt aus:

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

Dementsprechend müssen Sie je nach Anfrage verstehen, wie die Daten verarbeitet werden. Ich habe den Eintrag als Indikator gewählt "Type":"SubscriptionConfirmation", da es in der Abonnementbestätigungsanforderung vorhanden ist und nicht im Webhook. Basierend auf dem Vorhandensein/Fehlen dieses Eintrags in der POST-Anfrage erfolgt die weitere Ausführung des Programms entweder an die Funktion SubscriptionConfirmation, oder in die Funktion GotRecords.

Wir werden die SubscriptionConfirmation-Funktion nicht im Detail betrachten; sie wird gemäß den in dargelegten Grundsätzen implementiert Dokumentation. Sie können den Quellcode für diese Funktion unter einsehen Projekt-Git-Repositorys.

Die GotRecords-Funktion analysiert eine eingehende Anfrage und ruft für jedes Record-Objekt ein externes Skript (dessen Name im Parameter -script übergeben wurde) mit den folgenden Parametern auf:

• Bucket-Name
• Objektschlüssel
• Aktion:
  • kopieren – wenn in der ursprünglichen Anfrage EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete – wenn in der ursprünglichen Anfrage EventName = ObjectRemoved | Objekt löschen

Wenn also wie beschrieben ein Hook mit einer Post-Anfrage eintrifft obenund den Parameter -script=script.sh, dann wird das Skript wie folgt aufgerufen:

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

Es versteht sich, dass es sich bei diesem Webhook-Empfangsserver nicht um eine vollständige Produktionslösung, sondern um ein vereinfachtes Beispiel einer möglichen Implementierung handelt.

Beispiel der Arbeit

Lassen Sie uns die Dateien vom Haupt-Bucket in MCS mit dem Backup-Bucket in AWS synchronisieren. Der Haupt-Bucket heißt myfiles-ash, der Backup-Bucket heißt myfiles-backup (die Bucket-Konfiguration in AWS geht über den Rahmen dieses Artikels hinaus). Wenn eine Datei im Haupt-Bucket abgelegt wird, sollte ihre Kopie dementsprechend im Backup-Bucket erscheinen, und wenn sie aus dem Haupt-Bucket gelöscht wird, sollte sie im Backup-Bucket gelöscht werden.

Wir werden mit Buckets mithilfe des Dienstprogramms awscli arbeiten, das sowohl mit MCS-Cloud-Speicher als auch mit AWS-Cloud-Speicher kompatibel ist.

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) ...

Lassen Sie uns den Zugriff auf die S3 MCS-API konfigurieren:

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]:

Lassen Sie uns den Zugriff auf die AWS S3-API konfigurieren:

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]:

Schauen wir uns die Zugriffe an:

An AWS:

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

Für MCS müssen Sie beim Ausführen des Befehls —endpoint-url hinzufügen:

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

Zugriff.

Schreiben wir nun ein Skript zur Verarbeitung des eingehenden Hooks, nennen wir es 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=""
SOURCE_FILE=""
ACTION=""

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: 
#!/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
 sourcebucket sourcefile copy/delete"
    exit 1
    ;;
esac

Starten wir den Server:

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

Mal sehen, wie es funktioniert. Durch MCS-Weboberfläche Fügen Sie die Datei test.txt zum myfiles-ash-Bucket hinzu. Die Konsolenprotokolle zeigen, dass eine Anfrage an den Webhook-Server gestellt wurde:

2020/07/06 09:43:08 [POST] incoming HTTP request from 
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

Lassen Sie uns den Inhalt des myfiles-backup-Buckets in AWS überprüfen:

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

Jetzt löschen wir über die Weboberfläche die Datei aus dem myfiles-ash-Bucket.

Serverprotokolle:

2020/07/06 09:44:46 [POST] incoming HTTP request from 
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Eimerinhalt:

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

Die Datei wird gelöscht, das Problem ist behoben.

Fazit und ToDo

Der gesamte in diesem Artikel verwendete Code ist in meinem Repository. Es gibt auch Beispiele für Skripte und Beispiele für das Zählen von Signaturen für die Registrierung von Webhooks.

Dieser Code ist lediglich ein Beispiel dafür, wie Sie S3-Webhooks in Ihren Aktivitäten verwenden können. Wie ich zu Beginn sagte: Wenn Sie planen, einen solchen Server in der Produktion zu verwenden, müssen Sie den Server zumindest für asynchrone Arbeit umschreiben: Registrieren Sie eingehende Webhooks in einer Warteschlange (RabbitMQ oder NATS) und analysieren Sie sie von dort aus und verarbeiten Sie sie mit Arbeitnehmeranwendungen. Andernfalls kann es bei einem massiven Eintreffen von Webhooks zu einem Mangel an Serverressourcen zur Erledigung von Aufgaben kommen. Durch das Vorhandensein von Warteschlangen können Sie den Server und die Mitarbeiter verteilen und bei Fehlern Probleme mit sich wiederholenden Aufgaben lösen. Es empfiehlt sich außerdem, die Protokollierung auf eine detailliertere und standardisiertere umzustellen.

Good Luck!

Mehr Lektüre zum Thema:

• Arbeiten mit Mail.ru Cloud Solutions S3-Objektspeicher als Dateisystem.
• Wie Objektspeicher genutzt wird, um die Konkurrenz zu schlagen.

Source: habr.com