Példa egy eseményvezérelt alkalmazásra, amely webhookon alapul az S3 objektumtárolóban, Mail.ru Cloud Solutions

Rube Goldberg kávéfőző

Az eseményvezérelt architektúra növeli a felhasznált erőforrások költséghatékonyságát, mivel azokat csak akkor használják fel, amikor szükség van rájuk. Számos lehetőség kínálkozik arra, hogyan valósítsa meg ezt, és ne hozzon létre további felhő-entitásokat dolgozói alkalmazásként. És ma nem a FaaS-ről, hanem a webhookról fogok beszélni. Bemutatok egy bemutató példát az események objektumtároló webhook segítségével történő kezelésére.

Néhány szó az objektumtárolásról és a webhoookról. Az objektumtárolás lehetővé teszi bármilyen adat tárolását a felhőben objektumok formájában, amelyek S3-on vagy más API-n keresztül érhetők el (a megvalósítástól függően) HTTP/HTTPS-en keresztül. A webhookok általában egyéni HTTP-visszahívások. Általában egy esemény váltja ki őket, például a kód tárolóba kerülése vagy egy blogon közzétett megjegyzés. Amikor egy esemény bekövetkezik, a származási hely HTTP-kérést küld a webhook számára megadott URL-címre. Ennek eredményeként az egyik webhely eseményei műveleteket indíthatnak el a másikon (wiki). Abban az esetben, ha a forráshely egy objektumtároló, az események a tartalom változásaként működnek.

Példák egyszerű esetekre, amikor ilyen automatizálás használható:

1. Másolatok létrehozása az összes objektumról egy másik felhőtárolóban. Fájlok hozzáadásakor vagy módosításakor menet közben kell másolatokat készíteni.
2. Grafikus fájlok miniatűreinek sorozatának automatikus létrehozása, vízjelek hozzáadása a fényképekhez és egyéb képmódosítások.
3. Értesítés új dokumentumok érkezéséről (például egy elosztott könyvelési szolgáltatás jelentéseket tölt fel a felhőbe, a pénzügyi monitoring pedig értesítést kap az új jelentésekről, ellenőrzi és elemzi azokat).
4. Valamivel bonyolultabb esetekben például egy kérést generálnak a Kubernetes felé, amely létrehoz egy pod-ot a szükséges tárolókkal, átadja neki a feladat paramétereit, majd a feldolgozás után összecsukja a tárolót.

Példaként elkészítjük az 1. feladat egy változatát, amikor a Mail.ru Cloud Solutions (MCS) objektumtárhelyének módosításait webhookok segítségével szinkronizálják az AWS objektumtárhelyen. Valós betöltött esetben az aszinkron munkavégzést webhookok sorban történő regisztrálásával kell biztosítani, de a betanítási feladatnál e nélkül is elvégezzük a megvalósítást.

Munkarendszer

Az interakciós protokollt részletesen a Útmutató az S3 webhookhoz MCS-en. A munkaséma a következő elemeket tartalmazza:

• Kiadói szolgáltatás, amely az S3 tárhely oldalán található, és HTTP kéréseket tesz közzé, amikor a webnhook aktiválódik.
• Webhook fogadó szerver, amely meghallgatja a HTTP közzétételi szolgáltatástól érkező kéréseket, és elvégzi a megfelelő műveleteket. A szerver bármilyen nyelven írható, példánkban a szervert Go nyelven írjuk.

A webhookok S3 API-ban való megvalósításának sajátossága a webhook fogadó szerver regisztrációja a közzétételi szolgáltatáson. A webhook-fogadó szervernek különösen meg kell erősítenie a közzétételi szolgáltatástól érkező üzenetekre való feliratkozást (más webhook-megvalósításoknál általában nincs szükség az előfizetés megerősítésére).

Ennek megfelelően a webhook fogadó szervernek két fő műveletet kell támogatnia:

• válaszoljon a kiadói szolgáltatás regisztrációjának megerősítésére irányuló kérésére,
• feldolgozza a bejövő eseményeket.

Webhook fogadó szerver telepítése

A webhook fogadószerver futtatásához Linux-kiszolgálóra van szükség. Ebben a cikkben példaként egy virtuális példányt használunk, amelyet az MCS-re telepítünk.

Telepítsük a szükséges szoftvert és indítsuk el a webhook fogadó szervert.

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

A mappa klónozása a webhook fogadó szerverrel:

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.

Indítsuk el a szervert:

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

Feliratkozás a kiadói szolgáltatásra

A webhook fogadószerverét az API-n vagy webes felületen keresztül regisztrálhatja. Az egyszerűség kedvéért a webes felületen keresztül regisztrálunk:

1. Menjünk a vödrök részhez a vezérlőteremben.
2. Lépjen a gyűjtőhelyre, amelyhez a webhookot konfiguráljuk, és kattintson a fogaskerékre:

Lépjen a Webhooks lapra, és kattintson a Hozzáadás gombra:

Töltse ki a mezőket:

ID – a webhook neve.

Esemény – mely eseményeket kell továbbítani. Beállítottuk minden olyan esemény átvitelét, amely a fájlokkal való munka során (hozzáadás és törlés) történik.

URL – webhook fogadó szerver címe.

A szűrőelőtag/utótag egy olyan szűrő, amely lehetővé teszi webhook létrehozását csak olyan objektumokhoz, amelyek neve megfelel bizonyos szabályoknak. Például annak érdekében, hogy a webhook csak a .png kiterjesztésű fájlokat indítsa el, a Szűrő utótag be kell írnia, hogy „png”.

Jelenleg csak a 80-as és 443-as port támogatja a webhook fogadószerver elérését.

Kattintsunk Adjon hozzá horgot és a következőket fogjuk látni:

– tette hozzá Hook.

A webhookot fogadó szerver naplóiban mutatja a hook regisztrációs folyamatának előrehaladását:

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

A regisztráció befejeződött. A következő részben részletesebben megvizsgáljuk a webhook fogadó szerver működési algoritmusát.

A webhook fogadó szerver leírása

Példánkban a szerver Go nyelven van írva. Nézzük meg működésének alapelveit.

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

Fontolja meg a fő funkciókat:

• Ping() - egy útvonal, amely URL/ping-en keresztül válaszol, az élénkség-próba legegyszerűbb megvalósítása.
• Webhook() - fő útvonal, URL/webhook kezelő:
  • megerősíti a regisztrációt a kiadói szolgáltatásban (lépjen az Előfizetés megerősítése funkcióra),
  • feldolgozza a bejövő webhookokat (Gorecords funkció).
• A HmacSha256 és HmacSha256hex függvények a HMAC-SHA256 és HMAC-SHA256 titkosítási algoritmusok megvalósításai, amelyek kimenete hexadecimális számokból áll az aláírás kiszámításához.
• A main a fő funkció, feldolgozza a parancssori paramétereket és regisztrálja az URL-kezelőket.

A szerver által elfogadott parancssori paraméterek:

• -port az a port, amelyen a szerver figyelni fog.
• -cím - IP-cím, amelyet a szerver figyelni fog.
• A -script egy külső program, amely minden bejövő hook számára meghívásra kerül.

Nézzünk meg közelebbről néhány funkciót:

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

}

Ez a funkció határozza meg, hogy regisztráció megerősítésére vonatkozó kérés vagy webhook érkezett-e. Amint az abból következik dokumentáció, ha a regisztrációt megerősítik, a következő Json-struktúra érkezik a bejegyzéskérésben:

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»
}

Erre a kérdésre választ kell adni:

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

Ahol az aláírást a következőképpen számítják ki:

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

Ha webhook érkezik, a közzétételi kérés szerkezete így néz ki:

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

Ennek megfelelően a kéréstől függően meg kell értenie az adatok feldolgozásának módját. A bejegyzést választottam mutatónak "Type":"SubscriptionConfirmation", mivel jelen van az előfizetés-visszaigazolási kérelemben, és nincs jelen a webhookban. Ennek a bejegyzésnek a POST kérésben való megléte/hiánya alapján a program további végrehajtása vagy a függvényre megy. SubscriptionConfirmation, vagy egy függvénybe GotRecords.

A Subscription Confirmation funkciót nem vesszük figyelembe részletesen, azt a pontban meghatározott elvek szerint valósítjuk meg dokumentáció. Ennek a függvénynek a forráskódját a következő címen tekintheti meg projekt git adattárak.

A GotRecords függvény elemzi a bejövő kéréseket, és minden rekord objektumhoz meghív egy külső szkriptet (amelynek nevét a -script paraméterben adták át) a következő paraméterekkel:

• vödör neve
• tárgykulcs
• akció:
  • másolat - ha az eredeti kérésben EventName = ObjectCreated | PutObject | PutObjectCopy
  • törlés - ha az eredeti kérésben EventName = ObjectRemoved | DeleteObject

Így ha egy horog érkezik Posta kéréssel, a leírtak szerint felett, és a -script=script.sh paramétert, akkor a szkript a következőképpen lesz meghívva:

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

Meg kell érteni, hogy ez a webhook fogadó szerver nem egy teljes éles megoldás, hanem egy lehetséges megvalósítás egyszerűsített példája.

Példa a munkára

Szinkronizáljuk a fájlokat az MCS fő tárolójából az AWS biztonsági tárolójába. A fő tároló neve myfiles-ash, a biztonsági tároló neve myfiles-backup (a tárolóhely-konfiguráció az AWS-ben kívül esik ennek a cikknek a hatókörén). Ennek megfelelően, amikor egy fájlt a fő vödörbe helyezünk, a másolatának meg kell jelennie a tartalékban, a fő tárolóból való törlésekor pedig a tartalékban.

Az awscli segédprogrammal fogunk dolgozni a gyűjtőfalakkal, amely kompatibilis mind az MCS felhőalapú tárolással, mind az AWS felhőalapú tárolással.

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

Konfiguráljuk az S3 MCS API-hoz való hozzáférést:

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

Konfiguráljuk az AWS S3 API-hoz való hozzáférést:

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

Ellenőrizzük a hozzáféréseket:

Az AWS-hez:

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

MCS esetén a parancs futtatásakor hozzá kell adni a —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

Elérve.

Most írjunk egy szkriptet a bejövő hook feldolgozásához, nevezzük 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

Indítsuk el a szervert:

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

Lássuk, hogyan működik. Keresztül MCS webes felület adja hozzá a test.txt fájlt a myfiles-ash tárolóhoz. A konzolnaplók azt mutatják, hogy egy kérés érkezett a webhook-kiszolgálóhoz:

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

Ellenőrizzük a myfiles-backup tároló tartalmát az AWS-ben:

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

Most a webes felületen keresztül töröljük a fájlt a myfiles-ash vödörből.

Szervernaplók:

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

Vödör tartalma:

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

A fájl törlődik, a probléma megoldódott.

Következtetés és teendő

A cikkben használt összes kód az én tárhelyemben. Példák is találhatók a szkriptekre és a webhookok regisztrálásához szükséges aláírások számlálására.

Ez a kód nem más, mint egy példa arra, hogyan használhatja az S3 webhookot tevékenységei során. Ahogy az elején mondtam, ha egy ilyen szervert éles üzemben szeretne használni, akkor legalább át kell írnia a szervert az aszinkron munkára: a bejövő webhook-okat regisztrálnia kell egy sorba (RabbitMQ vagy NATS), majd onnan elemezni és feldolgozni. dolgozói alkalmazásokkal. Ellenkező esetben, amikor tömegesen érkeznek webhookok, előfordulhat, hogy a feladatok elvégzéséhez nem állnak rendelkezésre szervererőforrások. A sorok jelenléte lehetővé teszi a szerver és a dolgozók szétosztását, valamint az ismétlődő feladatok megoldását hiba esetén. A naplózást is célszerű részletesebbre és szabványosabbra módosítani.

Sok szerencsét!

További olvasnivalók a témában:

• A Mail.ru Cloud Solutions S3 objektumtárolás fájlrendszerként való használata.
• Hogyan használják az objektumtárolást a verseny legyőzésére.

Forrás: will.com