Пример на приложение, основано на събития, с уебхуков в обектно S3 хранилище на Mail.ru Cloud Solutions

Машина за кафе Руб Голдберг

Архитектура, базирана на събития, увеличава цена ефективността на използваните ресурси, защото те се задействат само в момента, когато са необходими. Има множество начини да се реализира това, без да се създават допълнителни облачни единици като worker приложения. Днес няма да говоря за FaaS, а за уебхукове. Ще покажа учебен пример за обработка на събития с помощта на уебхукове от обектно хранилище.

Несколко думи за обектното хранилище и за уебхуковете. Обектните хранилища позволяват да се съхраняват всякакви данни в облака под формата на обекти, достъпни през S3 или друг API (в зависимост от реализацията) чрез HTTP/HTTPS. Уебхуковете (webhooks) обикновено са потребителски обратни повиквания по HTTP. Обикновено те се активират от събитие, например, при изпращане на код в репозитория или публикуване на коментар в блог. Когато се случи събитие, изходният сайт изпраща HTTP заявка на URL адреса, указан за уебхука. В резултат на това е възможно събития на един сайт да предизвикват действия на друг сайт, когато изходният сайт е обектно хранилище, промените в съдържанието му действат като събития.wiki). В ситуация, при която изходният сайт е обектно хранилище, промените в него се явяват събития.

Примери за прости случаи, при които може да се ползва такава автоматизация:

1. Създаване на копия на всички обекти в друго облачно хранилище. Копията трябва да бъдат създавани "на живо", при всяко добавяне или изменение на файлове.
2. Автоматично създаване на серии миниатюри на графични файлове, добавяне на водни знаци към снимки, други модификации на изображения.
3. Уведомление за новопристигнали документи (например, разпределено счетоводство публикува отчети в облака, а финансовият мониторинг получава известия за нови отчети, проверява и анализира тези).
4. Трудните случаи предвиждат, например, генерирането на заявка към Kubernetes, което създава pod с необходимите контейнери, предава на него параметрите на задачата и след обработка затваря контейнера.

Като пример ще направим вариант на задача 1, при който промените в бакета на обектното хранилище Mail.ru Cloud Solutions (MCS) се синхронизират в обектното хранилище AWS чрез уебхукове. В реален натоварен случай трябва да предвидим асинхронна работа посредством регистрация на уебхукове в опашка, но за учебната задача ще направим реализация без това.

Схема на работа

Протоколът на взаимодействие е подробно описан в ръководството за S3 уебхукове на MCS. В схемата на работа има следните елементи:

• Услуга за публикуване, която се намира от страна на S3 хранилището и публикува HTTP заявки при задействане на уебхука.
• Сървър за приемане на уебхукове, който слуша повикванията на услугата за публикуване по HTTP и изпълнява съответните действия. Сървърът може да бъде написан на какъвто и да е език, в нашия пример ще напишем сървъра на Go.

Характеристика на реализирането на уебхукове в S3 API — регистрация на сървъра за приемане на уебхукове в услугата за публикуване. По-специално, сървърът за приемане на уебхукове трябва да потвърди абонамента за съобщения от услугата за публикуване (в други реализации на уебхукове обикновено не се изисква потвърждаване на абонамента).

Съответно, сървърът за приемане на уебхукове трябва да поддържа две основни операции:

• да отговаря на заявката на услугата за публикуване относно потвърждението на регистрацията,
• да обработва идващите събития.

Настройка на сървъра за приемане на уебхукове

За стартиране на сървъра за приемане на уебхукове е нужен Linux сървър. В тази статия за пример ще използваме виртуален инстанс, който разгръщаме на MCS.

Ще инсталираме необходимия софтуер и ще стартираме сървъра за приемане на уебхукове.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install git
Четене на списъци с пакети... Готово
Създаване на дърво на зависимости
Четене на информация за състоянието... Готово
Следните пакети бяха инсталирани автоматично и вече не са необходими:
 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
Използвайте 'sudo apt autoremove' за да ги премахнете.
Предложени пакети:
 git-daemon-run | git-daemon-sysvinit git-doc git-el git-email git-gui 
gitk gitweb git-cvs git-mediawiki git-svn
Следните НОВИ пакети ще бъдат инсталирани:
 git
0 актуализирани, 1 ново инсталиран, 0 за премахване и 46 не актуализирани.
Необходимо е да се изтеглят 3915 kB архиви.
След тази операция ще се използват допълнително 32.3 MB дисково пространство.
Get:1 http://MS1.clouds.archive.ubuntu.com/ubuntu bionic-updates/main 
amd64 git amd64 1:2.17.1-1ubuntu0.7 [3915 kB]
Изтеглени 3915 kB за 1s (5639 kB/s)
Избиране на преди не избрания пакет git.
(Четене на базата данни ... 53932 файлове и директории в момента са инсталирани.)
Подготовка за разопаковане .../git_12.17.1-1ubuntu0.7_amd64.deb ...
Разопаковане на git (1:2.17.1-1ubuntu0.7) ...
Настройване на git (1:2.17.1-1ubuntu0.7) ...

Клониране на папка с сървъра за прием на уебхукове:

ubuntu@ubuntu-basic-1-2-10gb:~$ git clone
https://github.com/RomanenkoDenys/s3-webhook.git
Клонираме в 's3-webhook'...
отдалечено: Изброяване на обекти: 48, готово.
отдалечено: Броене на обекти: 100% (48/48), готово.
отдалечено: Компресиране на обекти: 100% (27/27), готово.
отдалечено: Общо 114 (delta 20), повторно използвани 45 (delta 18), пакети повторно използвани 66
Получаване на обекти: 100% (114/114), 23.77 MiB | 20.25 MiB/s, готово.
Разрешаване на делти: 100% (49/49), готово.

Стартираме сървъра:

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

Абониране за услугата за публикуване

Регистрирането на сървъра за прием на уебхукове може да се извърши чрез API или уеб интерфейс. За простота ще регистрираме чрез уеб интерфейс:

1. Отиваме в раздела за бакети в контролния панел.
2. Влизаме в бакета, за който ще настройваме уебхуковете, и натискаме иконката с зъбно колело:

Преминаваме в таба Webhooks и натискаме Добавяне:

Попълваме полетата:

ID — името на уебхука.

Събитие — какви събития да се предават. Ние зададохме предаване на всички събития, които се случват при работа с файлове (добавяне и изтриване).

URL — адреса на сървъра за прием на уебхукове.

Филтър префикс/суфикс — филтър, който позволява генерирането на уебхукове само за обекти, чиито имена отговарят на определени правила. Например, за да уебхукът сработи само при файлове с разширение .png, в Филтър суфикс трябва да напишете «png».

В момента се поддържат само портове 80 и 443 за свързване със сървера за прием на уебхукове.

Натискаме Добави hook и виждаме следното:

Hook добавен.

Сървърът за прием на уебхукове в логовете показва преминаването на процеса на регистрация на хука:

ubuntu@ubuntu-basic-1-2-10gb:~\/s3-webhook$ sudo .\/s3-webhook -port 80
2020\/06\/15 12:01:14 [POST] входящая HTTP заявка от 
95.163.216.92:42530
2020\/06\/15 12:01:14 Получено време: 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 Генериран подпис за отговор: 
3754ce36636f80dfd606c5254d64ecb2fd8d555c27962b70b4f759f32c76b66d

Регистрация завърши. В следващия раздел по-подробно ще разгледаме алгоритъма на работа на сървъра за приемане на уебхукове.

Описание на сървъра за приемане на уебхукове

В нашия пример сървърът е написан на Go. Нека разгледаме основните принципи на неговата работа.

package main

// Генериране на hmac_sha256_hex
func HmacSha256hex(message string, secret string) string {
}

// Генериране на hmac_sha256
func HmacSha256(message string, secret string) string {
}

// Изпращане на потвърждение за абонамент
func SubscriptionConfirmation(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Обработка на получени записи
func GotRecords(w http.ResponseWriter, req *http.Request, body []byte) {
}

// Liveness probe
func Ping(w http.ResponseWriter, req *http.Request) {
    // лог на заявката
    log.Printf("[%s] входящ HTTP Ping заявка от %sn", req.Method, req.RemoteAddr)
    fmt.Fprintf(w, "Pongn")
}

// Уебхук
func Webhook(w http.ResponseWriter, req *http.Request) {
}

func main() {

    // получаване на аргументи от командния ред
    bindPort := flag.Int("port", 80, "число между 1-65535")
    bindAddr := flag.String("address", "", "IP адрес във вида точка.")
    flag.StringVar(&actionScript, "script", "", "външен скрипт за изпълнение")
    flag.Parse()

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

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

Нека разгледаме основните функции:

• Ping() — рут, отговарящ на URL\/ping, най-простата реализация на liveness probe.
• Webhook() — основен рут, обработчик на URL\/уебхука:
  • потвърдява регистрацията в услуга за публикуване (преминаване към функцията SubscriptionConfirmation),
  • обработва получените уебхукове (функция GotRecords).
• Функции HmacSha256 и HmacSha256hex — реализации на алгоритми за шифроване HMAC-SHA256 и HMAC-SHA256 с изходен формат от шестнадесетични числа за изчисляване на подпис.
• main — главната функция, обработва параметрите на командния ред и регистрира обработчиците на URL.

Параметри на командния ред, които сървърът приема:

• -port — порт, на който сървърът ще слуша.
• -address — IP адрес, на който сървърът ще слуша.
• -script — външна програма, която се извиква при всеки получен хук.

Нека разгледаме по-подробно някои функции:

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

}

Тази функция определя какво е получено — заявка за потвърждение на регистрация или уебхук. Както е видно от документацията, в случай на потвърждение на регистрацията, идва следната структура Json в 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":"Вие избрахте да се абонирате за темата $topic. За да потвърдите абонамента, трябва да отговорите с изчислената сигнатура",
    "TopicArn":"mcs2883541269|bucketA|s3:ObjectCreated:Put",
    "SignatureVersion":1,
    "Token":"RPE5UuG94rGgBH6kHXN9FUPugFxj1hs2aUQc99btJp3E49tA"
}

На този запитване трябва да се отговори:

content-type: application/json

{"signature":"ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af"}

Където сигнатурата се изчислява като:

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

Ако получим уебхук, структурата на Post-запитването изглежда така:

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

Съответно, в зависимост от запитването, трябва да разберем как да обработим данните. Избрах като индикатор записа "Type":"SubscriptionConfirmation", тъй като той присъства в запитването за потвърждение на абонамент и не присъства в уебхука. Въз основа на наличието/отсъствието на този запис в POST-запитването, последващото изпълнение на програмата преминава или в функцията SubscriptionConfirmation, или в функцията GotRecords.

Функцията SubscriptionConfirmation няма да бъде разглеждана подробно, тя е реализирана на принципите, изложени в документацията. Изучаване на изходния код на тази функция може да се направи в git-репозитория на проекта.

Функцията GotRecords анализира полученото запитване и за всеки обект Record извиква външния скрипт (името на който е предадено в параметъра -script) с параметрите:

• име на бакета
• ключ на обекта
• действие:
  • copy — ако в изходното запитване EventName = ObjectCreated | PutObject | PutObjectCopy
  • delete — ако в изходното запитване EventName = ObjectRemoved | DeleteObject

По този начин, ако е получен хук с Post-запитване, както е описано по-горе, и параметърът -script=script.sh, тогава скриптът ще бъде извикан по следния начин:

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

Трябва да се разбере, че този уебхук сървър не представлява завършено production решение, а опростен пример за възможна реализация.

Пример за работа

Ще извършим синхронизация на файловете от основния бакет в MCS към резервния бакет в AWS. Основният бакет се нарича myfiles-ash, а резервният — myfiles-backup (конфигурацията на бакета в AWS надхвърля рамките на тази статия). Съответно, когато файл бъде добавен в основния бакет, неговото копие трябва да се появи в резервния, а когато бъде изтрит от основния — да бъде изтрито и от резервния.

Ще работим с бакетите с помощта на утилитата awscli, която е съвместима както с облачното хранилище MCS, така и с облачното хранилище AWS.

ubuntu@ubuntu-basic-1-2-10gb:~$ sudo apt-get install awscli
Четене на списъци с пакети... Готово
Изграждане на зависимостно дърво
Четене на информация за състоянието... Готово
След операцията ще се използват 34.4 MB допълнително дисково пространство.
Разпаковане на awscli (1.14.44-1ubuntu1) ...
Настройване на awscli (1.14.44-1ubuntu1) ...

Нека конфигурираме достъп до 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
Име на основния регион [None]:
Формат на изхода по подразбиране [None]:

Нека конфигурираме достъп до 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
Име на основния регион [None]:
Формат на изхода по подразбиране [None]:

Нека проверим достъпите:

Към AWS:

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

За MCS, при работа на командата трябва да добавим —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

Достъпът е получен.

Сега ще напишем скрипт за обработка на входящия хук, наречен 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

Стартираме сървъра:

ubuntu@ubuntu-basic-1-2-10gb:~/s3-webhook$ sudo ./s3-webhook -port 80 -
скрипт scripts/s3_backup_mcs_aws.sh

Проверяваме как ще сработи. Чрез веб-интерфейса на MCS добавяме файла test.txt в бакета myfiles-ash. В логовете в конзолата се вижда, че е направен запитване на сървъра на уебхуковете:

2020/07/06 09:43:08 [POST] входящ HTTP запит от 
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

Нека проверим съдържанието на бакета myfiles-backup в 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

Сега чрез веб-интерфейса премахваме файла от бакета myfiles-ash.

Логовете на сървъра:

2020/07/06 09:44:46 [POST] входящ HTTP запит от 
95.163.216.92:58224
delete: s3://myfiles-backup/test.txt

Съдържанието на бакета:

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

Файлът е изтрит, задачата е решена.

Заключение и ToDo

Целият код, използван в тази статия, се намира в моя репозиторий. Тук можете да намерите примери на скриптове и примери за изчисляване на сигнатури за регистрация на уебхукове.

Този код е само пример за това как можете да използвате S3 уебхукове в своята дейност. Както споменах в началото, ако планирате да използвате такъв сървър в продуктивна среда, е необходимо най-малко да препишете сървъра за асинхронна работа: входящите уебхукове да се регистрират в опашка (RabbitMQ или NATS), а оттам да се обработват от worker приложения. В противен случай, при масовото пристигане на уебхукове, може да се сблъскате с недостиг на ресурси на сървъра за изпълнение на задачите. Наличието на опашки позволява да се разпределят сървърът и worker-ите, а също така да се решават проблемите с повторението на задачи в случай на срив. Също така е желателно да се променят логовете на по-подробни и по-стандартизирани.

Успех!

Допълнителна литература по темата:

• Работа с обектно S3 хранилище Mail.ru Cloud Solutions като с файлова система.
• Как се използва обектното хранилище, за да се изпреварят конкурентите.

Източник: habr.com