Ereignisgesteuerte Architektur erhöht die wirtschaftliche Effizienz der verwendeten Ressourcen, da diese erst zum Einsatz kommen, wenn sie benötigt werden. Es gibt zahlreiche Möglichkeiten, dies zu realisieren, ohne zusätzliche Cloud-Entitäten als Worker-Anwendungen zu schaffen. Heute werde ich nicht über FaaS sprechen, sondern über Webhooks. Ich werde ein Lernbeispiel für die Ereignisverarbeitung mit Webhooks in einem Objektspeicher zeigen.
Ein paar Worte zu Objektspeichern und 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 Rückrufe über HTTP. Sie werden normalerweise durch ein Ereignis ausgelöst, wie z.B. das Hochladen von Code in ein Repository oder einen Kommentar, der in einem Blog veröffentlicht wird. Wenn ein Ereignis eintritt, sendet die ursprüngliche Website eine HTTP-Anfrage an die für den Webhook angegebene URL. Dadurch können Ereignisse auf einer Website Aktionen auf einer anderen auslösen. Im Fall, dass die ursprüngliche Website ein Objektspeicher ist, bestehen die Ereignisse aus Änderungen seines Inhalts..
Beispiele für einfache Anwendungsfälle, in denen eine solche Automatisierung genutzt werden kann:
- Erstellung von Kopien aller Objekte in einem anderen Cloud-Speicher. Die Kopien sollten "on-the-fly" erstellt werden, bei jeder Hinzufügung oder Änderung von Dateien.
- Automatische Erstellung von Miniaturansichten grafischer Dateien, Hinzufügen von Wasserzeichen zu Fotos, andere Modifikationen von Bildern.
- Benachrichtigung über das Eintreffen neuer Dokumente (zum Beispiel veröffentlicht ein verteiltes Buchhaltungsbüro Berichte in der Cloud, während die Finanzaufsicht Benachrichtigungen über neue Berichte erhält, diese überprüft und analysiert).
- Etwas komplexere Anwendungsfälle umfassen beispielsweise das Erstellen einer Anfrage an Kubernetes, das einen Pod mit den benötigten Containern erstellt, ihm die Aufgabenparameter übergibt und nach der Verarbeitung den Container wieder schließt.
Als Beispiel werden wir die Aufgabe 1 realisieren, bei der Änderungen im Bucket des Objektspeichers von Mail.ru Cloud Solutions (MCS) mithilfe von Webhooks mit dem Objektspeicher von AWS synchronisiert werden. In einem realen, stark belasteten Anwendungsfall sollte die asynchrone Arbeit durch das Registrieren von Webhooks in einer Warteschlange berücksichtigt werden, aber für die Lernaufgabe werden wir diese Implementierung ohne dies machen.
Arbeitsdiagramm
Das Interaktionsprotokoll ist detailliert beschrieben in . Im Arbeitsdiagramm sind folgende Elemente enthalten:
- Veröffentlichungsdienst, der sich auf der Seite des S3-Speichers befindet und HTTP-Anfragen veröffentlicht, wenn das Webhook ausgelöst wird.
- Webhook-Empfangsserver, der die Anfragen des Veröffentlichungsdienstes über HTTP abhört und entsprechende Maßnahmen ergreift. Der Server kann in jeder Programmiersprache geschrieben sein, in unserem Beispiel werden wir einen Server in Go schreiben.
Eine Besonderheit der Implementierung von Webhooks in der S3-API ist die Registrierung des Webhook-Empfangsservers im Veröffentlichungsdienst. Insbesondere muss der Webhook-Empfangsserver das Abonnement für Nachrichten des Veröffentlichungsdienstes bestätigen (in anderen Implementierungen von Webhooks ist eine Bestätigung des Abonnements normalerweise nicht erforderlich).
Daher muss der Webhook-Empfangsserver zwei Hauptoperationen unterstützen:
- auf die Anfrage des Veröffentlichungsdienstes zur Bestätigung der Registrierung zu antworten,
- eingehende Ereignisse zu verarbeiten.
Einrichtung des Webhook-Empfangsservers
Um den Webhook-Empfangsserver zu starten, ist ein Linux-Server erforderlich. In diesem Artikel verwenden wir als Beispiel eine virtuelle Instanz, die wir auf MCS bereitstellen.
Installieren wir die erforderliche Software und starten den Webhook-Empfangsserver.
ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Paketlisten werden gelesen... Fertig
Abhängigkeitsbaum wird erstellt
Statusinformationen werden gelesen... Fertig
Die folgenden Pakete wurden automatisch installiert und werden nicht mehr benötigt:
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
Verwenden Sie 'sudo apt autoremove', um sie zu entfernen.
Vorgeschlagene Pakete:
git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui
gitk gitweb git-cvs git-mediawiki git-svn
Die folgenden NEUEN Pakete werden installiert:
git
0 aktualisiert, 1 neu installiert, 0 zu entfernen und 46 nicht aktualisiert.
Muss 3915 kB an Archiven herunterladen.
Nach dieser Operation werden 32,3 MB zusätzlichen Speicherplatz benötigt.
Lade:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Hat 3915 kB in 1s (5639 kB/s) heruntergeladen.
Wähle zuvor nicht ausgewähltes Paket git.
(Datenbank wird gelesen ... 53932 Dateien und Verzeichnisse sind derzeit installiert.)
Vorbereitung zum Entpacken .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Entpacken von git (1:2.17.1-1ubuntu0.7) ...
Konfiguration von git (1:2.17.1-1ubuntu0.7) ...Klone den Ordner mit dem Webhook-Empfangsserver:
ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Klone in 's3-webhook'...
remote: Objekte werden aufgezählt: 48, fertig.
remote: Objekte werden gezählt: 100% (48/48), fertig.
remote: Objekte werden komprimiert: 100% (27/27), fertig.
remote: Insgesamt 114 (delta 20), wiederverwendet 45 (delta 18), pack-wiederverwendet 66
Empfangene Objekte: 100% (114/114), 23.77 MiB | 20.25 MiB/s, fertig.
Deltas werden aufgelöst: 100% (49/49), fertig.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 80Abonnement für den Veröffentlichungsservice
Sie können Ihren Webhook-Server entweder über die API oder die Weboberfläche registrieren. Zur Vereinfachung werden wir die Registrierung über die Weboberfläche durchführen:
- im Verwaltungsbereich.
- Wir öffnen den Bucket, für den wir die Webhooks konfigurieren möchten, und klicken auf das Zahnrad:
Wir wechseln zur Registerkarte Webhooks und klicken auf Hinzufügen:
Wir füllen die Felder aus:
ID — Name des Webhooks.
Ereignis — welche Ereignisse übertragen werden sollen. Wir haben es so eingestellt, dass alle Ereignisse, die beim Arbeiten mit Dateien (Hinzufügen und Entfernen) auftreten, übertragen werden.
URL — Adresse des Webhook-Servers.
Filter-Präfix/Suffix — ein Filter, der es ermöglicht, Webhooks nur für Objekte zu generieren, deren Namen bestimmten Regeln entsprechen. Zum Beispiel, um den Webhook nur für Dateien mit der Erweiterung .png auszulösen, müssen wir bei Filter-Suffix „png“ eingeben.
Aktuell werden nur die Ports 80 und 443 für den Zugriff auf den Webhook-Server unterstützt.
Klicken wir auf Webhook hinzufügen und sehen Folgendes:
Webhook hinzugefügt.
Der Webhook-Server zeigt in den Protokollen den Verlauf des Registrierungsprozesses des Hooks:
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 generieren:
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66dDie Registrierung ist abgeschlossen. Im nächsten Abschnitt werden wir den Arbeitsablauf des Webhook-Servers detailliert 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ätigung des Abonnements senden
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}
// Bestätigung des Abonnements senden
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}
// Live-Prü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() {
// Befehlszeilenargumente abrufen
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 wichtigsten Funktionen:
- Ping() – Route, die auf URL/ping antwortet, eine einfache Implementierung des Liveness Probes.
- Webhook() – Haupt-Route, die URL/webhook verarbeitet:
- bestätigt die Registrierung beim Publikationsservice (Übergang zur Funktion SubscriptionConfirmation),
- verarbeitet eingehende Webhooks (Funktion Gotrecords).
- Die Funktionen HmacSha256 und HmacSha256hex sind Implementierungen der HMAC-SHA256-Verschlüsselungsalgorithmen und HMAC-SHA256 mit Ausgabe als Zeichenfolge von 16-stelligen Zahlen zur Berechnung der Signatur.
- main – Hauptfunktion, die die Parameter der Befehlszeile verarbeitet und URL-Handler registriert.
Die von dem Server akzeptierten Befehlszeilenparameter:
- -port – der Port, auf dem der Server lauscht.
- -address – die IP-Adresse, auf der der Server lauscht.
- -script – externe Programme, die für jeden eingehenden Hook aufgerufen werden.
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 Anfrage zur Bestätigung der Registrierung oder ein Webhook. Wie aus hervorgeht, kommt bei der Bestätigung der Registrierung 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, dem 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 man antworten:
content-type: application/json
{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}Wo die Signatur berechnet wird als:
signature = hmac_sha256(url, hmac_sha256(TopicArn,
hmac_sha256(Timestamp, Token)))Wenn jedoch ein Webhook ankommt, sieht die Struktur der POST-Anfrage so 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"
}
}
]
} Dementsprechend muss man je nach Anfrage verstehen, wie mit den Daten umzugehen ist. Ich habe als Indikator die Aufnahme gewählt "Type":"SubscriptionConfirmation", da es in der Bestätigungsanforderung für das Abonnement vorhanden ist und nicht im Webhook. Je nach Vorhandensein/Nichtvorhandensein dieses Eintrags in der POST-Anforderung wechselt die weitere Ausführung des Programms entweder in die Funktion SubscriptionConfirmation, oder in die Funktion GotRecords.
Die Funktion SubscriptionConfirmation werden wir nicht detailliert betrachten, sie ist nach den Prinzipien umgesetzt, die in dargelegt sind. Den Quellcode dieser Funktion kann man in .
Die Funktion GotRecords analysiert die eingehende Anfrage und ruft für jedes Objekt Record ein externes Skript (dessen Name im Parameter -script übergeben wurde) mit den Parametern auf:
- Name des Buckets
- Objektschlüssel
- Aktion:
- copy — wenn im ursprünglichen Anfrage EventName = ObjectCreated | PutObject | PutObjectCopy
- delete — wenn im ursprünglichen Anfrage EventName = ObjectRemoved | DeleteObject
Wenn also ein Hook mit einer Post-Anforderung eingegangen ist, wie es beschrieben wurde , und der Parameter -script=script.sh, wird das Skript wie folgt aufgerufen:
script.sh bucketA some-file-to-bucket copyEs ist zu verstehen, dass dieser Webhook-Empfangsserver keine abgeschlossene Produktionslösung, sondern ein vereinfachtes Beispiel einer möglichen Implementierung ist.
Beispielarbeit
Wir werden eine Synchronisierung der Dateien des Hauptbuckets im MCS mit einem Backup-Bucket in AWS vornehmen. Der Hauptbucket heißt myfiles-ash, das Backup ist myfiles-backup (die Konfiguration des Buckets in AWS geht über diesen Artikel hinaus). Dementsprechend sollte, wenn eine Datei in den Hauptbucket gelegt wird, eine Kopie im Backup erscheinen, und wenn sie aus dem Hauptbucket gelöscht wird, muss sie auch im Backup gelöscht werden.
Wir arbeiten mit den Buckets mit dem Tool awscli, 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
Paketlisten werden gelesen... Fertig
Abhängigkeitsbaum wird gebaut
Statusinformationen werden gelesen... Fertig
Nach dieser Operation werden 34.4 MB zusätzlicher Speicherplatz benötigt.
Entpacken von awscli (1.14.44-1ubuntu1) ...
Einrichten von awscli (1.14.44-1ubuntu1) ...Lassen Sie uns den Zugang zur S3-API von MCS konfigurieren:
ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile mcs
AWS Access Key ID [None]: hdywEPtuuJTExxxxxxxxxxxxxx
AWS Secret Access Key [None]: hDz3SgxKwXoxxxxxxxxxxxxxxxxxx
Standard-Regionsname [None]:
Standard-Ausgabeformat [None]:Lassen Sie uns den Zugang zur S3-API von AWS konfigurieren:
ubuntu@ubuntu-basic-1-2-10gb:~$ aws configure --profile aws
AWS Access Key ID [None]: AKIAJXXXXXXXXXXXX
AWS Secret Access Key [None]: dfuerphOLQwu0CreP5Z8l5fuXXXXXXXXXXXXXXXX
Standard-Regionsname [None]:
Standard-Ausgabeformat [None]:Überprüfen wir die Zugriffsrechte:
Auf AWS:
ubuntu@ubuntu-basic-1-2-10gb:~$ aws s3 ls --profile aws
2020-07-06 08:44:11 myfiles-backupFür MCS muss beim Ausführen des Befehls —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-ashZugriff erhalten.
Jetzt schreiben wir das 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
;;
esacServer starten:
ubuntu@ubuntu-basic-1-2-10gb:~\/s3-webhook$ sudo .\/s3-webhook -port 80 -
script scripts\/s3_backup_mcs_aws.shÜberprüfen wir, wie das funktioniert. Über fügen wir die Datei test.txt zum Bucket myfiles-ash hinzu. In den Logs in der Konsole sehen wir, dass eine Anfrage an den Webhook-Server gesendet wurde:
2020\/07\/06 09:43:08 [POST] eingehende 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.txtJetzt entfernen wir über das Web-Interface die Datei aus dem Bucket myfiles-ash.
Server-Logs:
2020\/07\/06 09:44:46 [POST] eingehende HTTP-Anfrage von
95.163.216.92:58224
delete: s3:\/\/myfiles-backup\/test.txtInhalt 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 . Dort finden Sie auch Skriptbeispiele und Beispiele für die Berechnung von Signaturen zur Registrierung von Webhooks.
Dieser Code ist nicht mehr als ein Beispiel dafür, wie S3-Webhooks in Ihrer Arbeit verwendet werden können. Wie ich zu Beginn sagte, ist es notwendig, den Server für den produktiven Einsatz mindestens auf asynchrone Arbeit umzustellen: 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 massivem Eingang von Webhooks zu Ressourcenengpässen auf dem Server kommen. Die Einrichtung von Warteschlangen ermöglicht es, Server und Worker zu entkoppeln und Probleme mit der Wiederholung von Aufgaben im Falle von Fehlern zu lösen. Außerdem ist es wünschenswert, das Logging auf detailliertere und standardisierte Weise umzuschalten.
Viel Erfolg!
Weiterführende Literatur zu diesem Thema:
Quelle: habr.com
