Příklad aplikace řízené událostmi založené na webhooks v objektovém S3-úložišti Mail.ru Cloud Solutions

Rube Goldbergova kávovar

Architektura řízená událostmi zvyšuje cenovou efektivitu využívaných zdrojů, protože se aktivují pouze v okamžiku, kdy jsou potřeba. Existuje mnoho způsobů, jak toho dosáhnout, aniž by bylo nutné vytvářet další cloudové entity jako worker-aplikace. Dnes nebudu mluvit o FaaS, ale o webhooks. Ukážu Vám vzdělávací příklad zpracování událostí pomocí webhooků objektového úložiště.

Pár slov o objektovém úložišti a o webhooks. Objektová úložiště umožňují ukládat jakákoli data v cloudu ve formě objektů, které jsou přístupné přes S3 nebo jiná API (v závislosti na implementaci) prostřednictvím HTTP/HTTPS. Webhooks (webhooky) jsou obecně uživatelské zpětné volání přes HTTP. Obvykle se spouštějí událostí, například odesláním kódu do repozitáře nebo komentářem zveřejněným v blogu. Když událost nastane, zdrojový web odešle HTTP požadavek na URL adresu určenou pro webhook. V důsledku toho můžete udělat tak, aby události na jednom webu spouštěly akce na jiném. V případě, kdy zdrojovým webem je objektové úložiště, jsou událostmi změny jeho obsahu.wiki). V případě, že zdrojovým webem je objektové úložiště, představují události změny jeho obsahu.

Příklady jednoduchých případů, kdy lze tuto automatizaci použít:

1. Vytváření kopií všech objektů v jiném cloudovém úložišti. Kopie by měly být vytvářeny "za běhu", při jakémkoli přidání nebo změně souborů.
2. Automatické vytváření sérií miniatur grafických souborů, přidávání vodoznaků k fotografiím, další úpravy obrázků.
3. Oznámení o příchodu nových dokumentů (například distribuovaná účetní služba zveřejňuje v cloudu zprávy, zatímco finanční monitoring dostává oznámení o nových zprávách, kontroluje je a analyzuje).
4. O něco složitější případy zahrnují například vytvoření dotazu na Kubernetes, který vytvoří pod se správnými kontejnery, předává mu parametry úkolu a po zpracování kontejner uzavře.

Jako příklad provedeme variantu úkolu 1, kdy se změny v bucketu objektového úložiště Mail.ru Cloud Solutions (MCS) synchronizují pomocí webhooků do objektového úložiště AWS. V reálném zatíženém případě by mělo být navrženo asynchronní zpracování registrací webhooků do fronty, ale pro vzdělávací úkol provedeme implementaci bez toho.

Průběh práce

Protokol interakce je podrobně popsán v příručce pro S3 webhooks na MCS. V průběhu práce jsou následující prvky:

• Služba publikace, která se nachází na straně S3 úložiště a publikuje HTTP požadavky při aktivaci webhooku.
• Server pro příjem webhooků, který naslouchá požadavkům služby publikace přes HTTP a vykonává odpovídající akce. Server může být napsán v jakémkoliv jazyce, v našem případě napíšeme server v Go.

Specifikou implementace webhooků v S3 API je registrace serveru pro příjem webhooků u služby publikace. Konkrétně musí server pro příjem webhooků potvrdit odběr zpráv od služby publikace (v jiných implementacích webhooků obvykle potvrzení odběru není vyžadováno).

Server pro příjem webhooků tedy musí podporovat dvě základní operace:

• odpovídat na požadavek služby publikace o potvrzení registrace,
• zpracovávat přicházející události.

Nastavení serveru pro příjem webhooků

Pro spuštění serveru pro příjem webhooků je potřeba server s Linuxem. V tomto článku použijeme virtuální instanci, kterou rozbalujeme na MCS.

Nainstalujeme potřebný software a spustíme server pro příjem webhooků.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Čtení seznamu balíčků... Hotovo
Budování stromu závislostí
Čtení informací o stavu... Hotovo
Následující balíčky byly automaticky nainstalovány a již nejsou potřebné:
 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
Použijte 'sudo apt autoremove' k jejich odstranění.
Navrhované balíčky:
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
Následující NOVÉ balíčky budou nainstalovány:
 git
0 aktualizováno, 1 nově nainstalováno, 0 k odstranění a 46 neaktualizováno.
Je potřeba stáhnout 3915 kB archivů.
Po této operaci bude použito 32,3 MB dodatečného místa na disku.
Stahování:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
 amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Stáhnuto 3915 kB za 1s (5639 kB/s)
Výběr dříve nenainstalovaného balíčku git.
(Čtení databáze ... 53932 souborů a adresářů je aktuálně nainstalováno.)
Připravuji k rozbalení .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Rozbalování git (1:2.17.1-1ubuntu0.7) ...
Nastavování git (1:2.17.1-1ubuntu0.7) ...

Klónujeme složku se serverem pro příjem webhooků:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Klonování do 's3-webhook'...
remote: Sčítání objektů: 48, hotovo.
remote: Počet objektů: 100% (48/48), hotovo.
remote: Komprese objektů: 100% (27/27), hotovo.
remote: Celkem 114 (delta 20), znovu použito 45 (delta 18), balení znovu použito 66
Přijato objekty: 100% (114/114), 23.77 MiB | 20.25 MiB/s, hotovo.
Zpracování delt: 100% (49/49), hotovo.

Spustíme server:

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

Předplatné služby publikování

Své server pro příjem webhooků můžete zaregistrovat přes API nebo webové rozhraní. Pro jednoduchost se zaregistrujeme přes webové rozhraní:

1. Přejdeme do sekce Buckety v administraci.
2. Otevřeme bucket, pro který budeme nastavovat webhooks, a klikneme na ozubené kolečko:

Přejdeme na kartu Webhooks a klikneme na Přidat:

Vyplníme pole:

ID — název webhooku.

Událost — jaké události předávat. Nastavili jsme předávání všech událostí, které se dějí při práci se soubory (přidání a odstranění).

URL — adresa serveru pro příjem webhooků.

Filtr prefix/suffix — filtr, který umožňuje generovat webhooky pouze pro objekty, jejichž názvy odpovídají určitým pravidlům. Například, aby webhook fungoval pouze pro soubory s příponou .png, do Filtr přípona musíme napsat "png".

V současné době jsou pro přístup k serveru příjmu webhooků podporovány pouze porty 80 a 443.

Klikneme na Přidat hook a uvidíme následující:

Hook byl přidán.

Server příjmu webhooků v logu ukazuje průběh registrace hooku:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80
2020/06/15 12:01:14 [POST] příchozí HTTP požadavek z
95.163.216.92:42530
2020/06/15 12:01:14 Získán časový podpis: 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 Generovat podpis odpovědi: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Registrace byla dokončena. V následující sekci podrobněji prozkoumáme algoritmus práce serveru příjmu webhooků.

Popis serveru příjmu webhooků

V našem příkladu je server napsán v Go. Rozebere se základní principy jeho fungování.

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] příchozí HTTP Ping požadavek od %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, "číslo mezi 1-65535")
    bindAddr := flag.String("address", "", "ip adresa ve formátu tečka")
    flag.StringVar(&actionScript, "script", "", "externí skript k provedení")
    flag.Parse()

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

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

Podívejme se na hlavní funkce:

• Ping() — routa, která odpovídá na URL/ping, nejjednodušší implementace liveness probe.
• Webhook() — hlavní routa, zpracovatel URL/webhook:
  • potvrzuje registraci na službě publikace (přechod do funkce SubscriptionConfirmation),
  • zpracovává příchozí webhooky (funkce Gotrecords).
• Funkce HmacSha256 a HmacSha256hex — implementace algoritmů šifrování HMAC-SHA256 a HMAC-SHA256 s výstupem jako řetězec 16-ciferných čísel pro výpočet podpisu.
• main — hlavní funkce, zpracovává parametry příkazové řádky a registruje zpracovatele URL.

Parametry příkazové řádky, které server přijímá:

• -port — port, na kterém bude server naslouchat.
• -address — IP adresa, na které bude server naslouchat.
• -script — externí program, který se volá při každém příchozím webhooku.

Podívejme se podrobněji na některé funkce:

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

}

Tato funkce určuje, co přišlo — požadavek na potvrzení registrace nebo webhook. Jak vyplývá z dokumentaci, v případě potvrzení registrace přichází následující struktura JSON v požadavku 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":"Zvolil(a) jste odběr k tématu $topic. Pro potvrzení odběru je nutné odpovědět vypočítaným podpisem",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":"RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA"
}

Na tuto žádost je potřeba odpovědět:

content-type: application/json

{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}

Kde se podpis vypočítá jako:

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

Pokud přijde webhook, struktura Post požadavku vypadá takto:

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

V závislosti na žádosti je tedy třeba pochopit, jak zpracovat data. Jako indikátor jsem vybral záznam "Type":"SubscriptionConfirmation", protože je obsažena v dotazu na potvrzení předplatného a není obsažena v webhooku. V závislosti na přítomnosti/nepřítomnosti tohoto záznamu v POST dotazu, další provádění programu přechází buď do funkce SubscriptionConfirmation, nebo do funkce GotRecords.

Funkci SubscriptionConfirmation podrobně nezvažujeme, je implementována podle zásad uvedených v dokumentaci. Původní kód této funkce si můžete prohlédnout v gitovém repozitáři projektu.

Funkce GotRecords zpracovává příchozí dotaz a pro každý objekt Record volá externí skript (jehož jméno bylo předáno v parametru -script) s parametry:

• název bucketu
• klíč objektu
• akce:
  • copy — pokud je v původním dotazu EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete — pokud je v původním dotazu EventName = ObjectRemoved | DeleteObject

Takže pokud dorazil hook s Post-dotazem, jak je popsáno přístupu., a parametr -script=script.sh, bude skript zavolán následujícím způsobem:

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

Je třeba chápat, že tento server pro příjem webhooků není dokončeným produkčním řešením, ale zjednodušeným příkladem možné implementace.

Příklad práce

Synchronizujeme soubory z hlavního bucketu v MCS do záložního bucketu v AWS. Hlavní bucket se nazývá myfiles-ash, záložní — myfiles-backup (konfigurace bucketu v AWS přesahuje rámec tohoto článku). Takže když je soubor uložen do hlavního bucketu, jeho kopie by měla být vytvořena v záložním, a když je z hlavního odstraněn — měla by být odstraněna v záložním.

Budeme pracovat s bucketem pomocí nástroje awscli, se kterým je kompatibilní jak cloudové úložiště MCS, tak cloudové úložiště AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Čtení seznamu balíčků... Hotovo
Sestavování stromu závislostí
Čtení stavu informací... Hotovo
Po této operaci bude použito 34,4 MB dodatečného místa na disku.
Rozbalování awscli (1.14.44-1ubuntu1) ...
Nastavení awscli (1.14.44-1ubuntu1) ...

Nakonfigurujeme přístup k 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
Výchozí název regionu [None]:
Výchozí formát výstupu [None]:

Nakonfigurujeme přístup k 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
Výchozí název regionu [None]:
Výchozí formát výstupu [None]:

Otestujeme přístupy:

K AWS:

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

Pro MCS, při práci s příkazem je třeba přidávat —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

Přístup byl získán.

Nyní napíšeme skript pro zpracování příchozího webhooku, nazveme ho 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

Spouštíme server:

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

Zkontrolujeme, jak to funguje. Přes webové rozhraní MCS přidáme soubor test.txt do bucketu myfiles-ash. V logu v konzoli je vidět, že byl proveden požadavek na server webhooků:

2020/07/06 09:43:08 [POST] příchozí HTTP požadavek z 
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

Zkontrolujeme obsah bucketu myfiles-backup v 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

Nyní přes webové rozhraní odstraníme soubor z bucketu myfiles-ash.

Logy serveru:

2020/07/06 09:44:46 [POST] příchozí HTTP požadavek z 
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Obsah bucketu:

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

Soubor byl odstraněn, úkol je splněn.

Závěr a ToDo

Veškerý kód použitý v tomto článku se nachází v mém repozitáři. Tam jsou také příklady skriptů a příklady výpočtu podpisů pro registraci webhooků.

Tento kód je pouze příkladem toho, jak lze využít S3 webhooky ve své činnosti. Jak jsem řekl na začátku, pokud plánujete použití takového serveru v produkci, je nutné přepsat server pro asynchronní práci: příchozí webhooky registrovat do fronty (RabbitMQ nebo NATS), a odtud je zpracovávat pracovními aplikacemi. Jinak se při masivním příchodu webhooků můžete setkat s nedostatkem prostředků serveru pro provádění úkolů. Mít fronty umožňuje oddělit server a pracovníky a také řešit problémy s opakováním úkolů v případě selhání. Je také žádoucí přejít k podrobnějšímu a standardizovanějšímu logování.

Hodně štěstí!

Další čtení na téma:

• Práce s objektovým S3 úložištěm Mail.ru Cloud Solutions jako s filesystémem.
• Jak využívají objektové úložiště k překonání konkurentů.

Zdroj: habr.com