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

Rube Goldberg Kaffeemaschine

Die ereignisgesteuerte Architektur erhöht die Kosteneffizienz der eingesetzten Ressourcen, da sie nur dann aktiviert werden, wenn sie benötigt werden. Es gibt viele Möglichkeiten, dies zu realisieren, ohne zusätzliche Cloud-Entitäten in Form von Worker-Anwendungen zu schaffen. Heute werde ich nicht über FaaS sprechen, sondern über Webhooks. Ich zeige ein Beispiel zur Ereignisverarbeitung mithilfe von Webhooks im Objektspeicher.

Ein paar Worte zum Objektspeicher und zu Webhooks. Objektspeicher ermöglichen es, beliebige Daten in der Cloud in Form von Objekten zu speichern, die über S3 oder eine andere API (je nach Implementierung) über HTTP/HTTPS zugänglich sind. Webhooks sind im Allgemeinen benutzerdefinierte HTTP-Rückrufe. Sie werden normalerweise durch ein Ereignis ausgelöst, beispielsweise durch das Hochladen von Code in ein Repository oder durch einen Kommentar, der in einem Blog veröffentlicht wird. Wenn ein Ereignis eintritt, sendet die Ursprungswebsite eine HTTP-Anfrage an die für den Webhook angegebene URL. Dadurch können Ereignisse auf einer Website Aktionen auf einer anderen Website auslösen.Wiki). Im Falle, dass das Ausgangs-Website ein Objektspeicher ist, bestehen die Ereignisse aus Änderungen seines Inhalts.

Beispiele einfacher Anwendungsfälle, in denen eine solche Automatisierung eingesetzt werden kann:

1. Erstellung von Kopien aller Objekte in einem anderen Cloud-Speicher. Die Kopien sollten „in Echtzeit“ erstellt werden, bei jeder Hinzufügung oder Änderung der Dateien.
2. Automatische Erstellung von Serien von Miniaturansichten von Grafikdateien, Hinzufügen von Wasserzeichen zu Fotos, andere Bildmodifikationen.
3. Benachrichtigung über das Eintreffen neuer Dokumente (zum Beispiel veröffentlicht ein dezentrales Buchhaltungsbüro Berichte in der Cloud, und die Finanzüberwachung erhält Benachrichtigungen über neue Berichte, prüft und analysiert diese).
4. Etwas komplexere Anwendungsfälle umfassen beispielsweise die Erstellung einer Anfrage an Kubernetes, die ein Pod mit den benötigten Containern erstellt, ihm Parameter für die Aufgabe übergibt und nach der Verarbeitung den Container herunterfährt.

Als Beispiel nehmen wir die Aufgabe 1, bei der Änderungen im Objekt-Speicher von Mail.ru Cloud Solutions (MCS) über Webhooks mit dem Objekt-Speicher von AWS synchronisiert werden. In einem realen Lastfall sollte eine asynchrone Verarbeitung durch die Registrierung von Webhooks in einer Warteschlange vorgesehen werden, jedoch setzen wir in dieser Übung die Implementierung ohne dies um.

Funktionsschema

Das Protokoll der Interaktion ist detailliert beschrieben in dem Handbuch zu S3-Webhooks auf MCS. Im Funktionsschema gibt es folgende Elemente:

• Veröffentlichungsdienst, der sich auf der Seite des S3-Speichers befindet und HTTP-Anfragen veröffentlicht, wenn der Webhook ausgelöst wird.
• Webhook-Server, der HTTP-Anfragen vom Veröffentlichungsdienst empfängt und entsprechende Aktionen ausführt. Der Server kann in jeder Sprache geschrieben werden; in unserem Beispiel werden wir den Server in Go schreiben.

Eine Besonderheit der Implementierung von Webhooks im S3-API ist die Registrierung des Webhook-Servers beim Veröffentlichungsdienst. Insbesondere muss der Webhook-Server das Abonnement für Nachrichten des Veröffentlichungsdienstes bestätigen (in anderen Implementierungen ist die Bestätigung des Abonnements normalerweise nicht erforderlich).

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

• auf die Bestätigungsanfrage des Publikationsdienstes zu antworten,
• eingehende Ereignisse zu verarbeiten.

Installation des Webhook-Empfangsservers

Für den Betrieb des Webhook-Empfangsservers wird ein Linux-Server benötigt. 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) ...

Wir klonen den Ordner mit dem Webhook-Server:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Klonen in 's3-webhook'...
remote: Objekte werden aufgezählt: 48, fertig.
remote: Objekte zählen: 100% (48/48), fertig.
remote: Objekte komprimieren: 100% (27/27), fertig.
remote: Insgesamt 114 (delta 20), 45 wiederverwendet (delta 18), pack-wiederverwendet 66
Objekte empfangen: 100% (114/114), 23.77 MiB | 20.25 MiB/s, fertig.
Deltas auflösen: 100% (49/49), fertig.

Lassen Sie uns den Server starten:

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

Abonnement für den Veröffentlichungsdienst

Registrieren Sie Ihren Webhook-Server entweder über die API oder die Web-Oberfläche. Zur Vereinfachung registrieren wir über die Web-Oberfläche:

1. Gehen Sie zu den Buckets im Verwaltungspanel.
2. Öffnen Sie den Bucket, für den Sie Webhooks einrichten möchten, und klicken Sie auf das Zahnrad:

Wechseln Sie zum Tab Webhooks und klicken Sie auf Hinzufügen:

Füllen Sie die Felder aus:

ID - Name des Webhooks.

Ereignis - welche Ereignisse übermittelt werden sollen. Wir haben die Übertragung aller Ereignisse, die beim Arbeiten mit Dateien stattfinden (Hinzufügen und Löschen), festgelegt.

URL - Adresse des Webhook-Servers.

Filterpräfix/-suffix - ein Filter, der es ermöglicht, Webhooks nur für Objekte zu generieren, deren Namen bestimmten Regeln entsprechen. Zum Beispiel, wenn der Webhook nur für Dateien mit der Erweiterung .png ausgelöst werden soll, geben Sie in Filter suffix „png“ ein.

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

Klicken Sie auf Webhook hinzufügen und sehen Sie Folgendes:

Webhook hinzugefügt.

Der Webhook-Server zeigt in den Protokollen den Registrierungsvorgang des Webhooks an:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] eingehende HTTP-Anfrage von 
95.163.216.92:42530
2020/06/15 12:01:14 Zeitstempel erhalten: 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 Antwortsignatur generiert: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Die Registrierung ist abgeschlossen. Im nächsten Abschnitt werden wir den Arbeitsablauf des Webhook-Server detaillierter betrachten.

Beschreibung des Webhook-Servers

In unserem Beispiel ist der Server in Go geschrieben. Lassen Sie uns die grundlegenden Prinzipien seiner Funktionsweise untersuchen.

package main

// HmacSha256hex generieren
func HmacSha256hex(message string, secret string) string {
}

// HmacSha256 generieren
func HmacSha256(message string, secret string) string {
}

// Bestätigungsnachricht senden
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Bestätigungsnachricht senden
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Lebenssignalisierungsprüfung
func Ping(w http.ResponseWriter, req *http.Request) {
    // Anfrage protokollieren
    log.Printf("[%s] eingehende HTTP Ping-Anfrage von %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

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

func main() {

    // Kommandozeilenargumente auslesen
    bindPort := flag.Int("port", 80, "Zahl zwischen 1-65535")
    bindAddr := flag.String("address", "", "IP-Adresse im Punktformat")
    flag.StringVar(&actionScript, "script", "", "auszuführendes externes Skript")
    flag.Parse()

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

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

Betrachten wir die Hauptfunktionen:

• Ping() — Route, die auf die URL/ping antwortet, einfachste Implementierung einer Lebenssignalisierungsprüfung.
• Webhook() — Hauptroute, Handler für die URL/Webhook:
  • bestätigt die Registrierung im Publikationsdienst (Übergang zur Funktion SubscriptionConfirmation),
  • verarbeitet eingehende Webhooks (Funktion GotRecords).
• Die Funktionen HmacSha256 und HmacSha256hex sind Implementierungen der HMAC-SHA256-Verschlüsselungsalgorithmen und der HMAC-SHA256-Ausgabe in Form einer hexadezimalen Zeichenfolgen für die Berechnung der Signatur.
• main — die Hauptfunktion, die die Befehlszeilenparameter verarbeitet und URL-Handler registriert.

Befehlszeilenparameter, die vom Server akzeptiert werden:

• -port — der Port, auf dem der Server lauscht.
• -address — die IP-Adresse, auf der der Server lauscht.
• -script — das externe Programm, das für jeden eingehenden Hook aufgerufen wird.

Betrachten wir einige Funktionen im Detail:

//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 bestimmt, was eingegangen ist — eine Bestätigungsanfrage oder ein Webhook. Wie folgt aus Dokumentation., im Falle einer Bestätigung des Abonnements kommt die folgende JSON-Struktur in der POST-Anfrage:

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":"Sie haben sich entschieden, das Thema $topic zu abonnieren. Um das Abonnement zu bestätigen, müssen Sie mit der berechneten Signatur antworten",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":"RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA"
}

Auf diese Anfrage muss geantwortet werden:

content-type: application/json

{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}

Hierbei wird die Signatur berechnet als:

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

Wenn ein Webhook kommt, sieht die POST-Anfrage folgendermaßen aus:

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

Daher ist es wichtig, je nach Anfrage zu verstehen, wie die Daten verarbeitet werden sollen. Ich habe als Indikator den Eintrag gewählt "Type":"SubscriptionConfirmation", da dieser in der Anfrage zur Bestätigung des Abonnements vorhanden ist, jedoch nicht im Webhook. Je nach Vorhandensein oder Abwesenheit dieses Eintrags im POST-Request wechselt die weitere Ausführung des Programms entweder in die Funktion SubscriptionConfirmation, oder in die Funktion GotRecords.

Die Funktion SubscriptionConfirmation werden wir nicht weiter betrachten, sie ist nach den beschriebenen Prinzipien implementiert. Dokumentation.. Den Quellcode dieser Funktion können Sie in dem Git-Repository des Projekts.

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

• der Bucket-Name
• der Objekt-Schlüssel
• Aktion:
  • copy — wenn im ursprünglichen Ereignisname EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete — wenn im ursprünglichen Ereignisname EventName = ObjectRemoved | DeleteObject

Daher wird, wenn ein Hook mit einer POST-Anfrage wie beschrieben ankommt oben, und der Parameter -script=script.sh gesetzt ist, das Skript wie folgt aufgerufen:

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

Es ist wichtig zu verstehen, dass dieser Webhook-Server keine vollständige Produktionslösung ist, sondern ein vereinfachtes Beispiel einer möglichen Implementierung.

Ein Arbeitsbeispiel

Wir werden die Dateien des Haupt-Buckets in MCS mit dem Backup-Bucket in AWS synchronisieren. Der Haupt-Bucket heißt myfiles-ash, der Backup-Bucket heißt myfiles-backup (die Konfiguration des Buckets in AWS überschreitet den Rahmen dieses Artikels). Folglich sollte, wenn eine Datei in den Haupt-Bucket gelegt wird, eine Kopie im Backup-Bucket erscheinen, und wenn sie aus dem Haupt-Bucket gelöscht wird, sollte sie auch im Backup-Bucket gelöscht werden.

Wir werden mit dem Tool awscli an Buckets arbeiten, das sowohl mit dem Cloud-Speicher MCS als auch mit dem Cloud-Speicher AWS kompatibel ist.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Lesen der Paketlisten... Fertig
Abhängigkeitsbaum wird aufgebaut
Lesen der Statusinformationen... Fertig
Nach dieser Aktion werden 34,4 MB zusätzlicher Speicherplatz belegt.
Entpacken von awscli (1.14.44-1ubuntu1)...
Einrichten von awscli (1.14.44-1ubuntu1)...

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

ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile mcs
AWS Access Key ID [None]: hdywEPtuuJTExxxxxxxxxxxxxx
AWS Secret Access Key [None]: hDz3SgxKwXoxxxxxxxxxxxxxxxxxx
Standardregion [None]:
Standardausgabformat [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
Standardregion [None]:
Standardausgabformat [None]:

Überprüfen wir die Zugriffe:

Zu AWS:

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

Für MCS muss bei der Arbeit mit dem Befehl —endpoint-url hinzugefügt werden:

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

Jetzt schreiben wir ein Skript zur Verarbeitung des eingehenden Hooks, das wir s3_backup_mcs_aws.sh nennen.

#!/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

Server starten:

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

Lassen Sie uns überprüfen, wie das funktionieren wird. Über das Web-Interface von MCS Wir fügen die Datei test.txt in den Bucket myfiles-ash hinzu. In den Konsolenprotokollen ist zu sehen, dass eine Anfrage an den Webhook-Server gesendet wurde:

2020/07/06 09:43:08 [POST] Eingang HTTP-Anfrage von 
95.163.216.92:56612
download: s3://myfiles-ash/test.txt nach ..//..//..//tmp/myfiles-ash/test.txt
upload: ..//..//..//tmp/myfiles-ash/test.txt nach 
s3://myfiles-backup/test.txt

Überprüfen wir den Inhalt des Buckets 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

Jetzt entfernen wir die Datei über die Webschnittstelle aus dem Bucket myfiles-ash.

Serverprotokolle:

2020/07/06 09:44:46 [POST] Eingang HTTP-Anfrage von 
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Inhalt des Buckets:

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

Die Datei wurde gelöscht, die Aufgabe ist erledigt.

Fazit und ToDo

Der gesamte Code, der in diesem Artikel verwendet wird, befindet sich in meinem Repository. Dort finden sich auch Beispiele für Skripte und Beispiele zur Berechnung von Signaturen zur Registrierung von Webhooks.

Dieser Code ist lediglich ein Beispiel dafür, wie S3-Webhooks in der eigenen Arbeit genutzt werden können. Wie ich zu Beginn sagte, sollte man, wenn man plant, einen solchen Server produktiv zu nutzen, mindestens den Server für asynchrone Arbeitsabläufe anpassen: Eingehende Webhooks sollten in einer Warteschlange (RabbitMQ oder NATS) registriert werden, von wo aus sie dann von Worker-Anwendungen verarbeitet werden. Andernfalls kann es bei einem massiven Eingang von Webhooks zu Ressourcenengpässen des Servers kommen. Die Verwendung von Warteschlangen ermöglicht es, den Server und die Worker zu trennen und auch die Wiederholung von Aufgaben im Falle von Fehlern zu handhaben. Es wäre auch ratsam, das Logging auf detailliertere und standardisierte Protokollierung umzustellen.

Viel Erfolg!

Weitere Lektüre zum Thema:

• Arbeiten mit dem objektbasierten S3-Speicher von Mail.ru Cloud Solutions wie mit einem Dateisystem.
• Wie objektbasierte Speicherlösungen genutzt werden, um Konkurrenten auszustechen.

Quelle: habr.com