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

Кафе машина Rube Goldberg

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

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

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

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

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

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

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

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

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

Съответно сървърът за получаване на webhook трябва да поддържа две основни операции:

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

Инсталиране на сървър за получаване на webhook

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

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

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

Клонирайте папката със сървъра за получаване на webhook:

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.

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

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

Абонирайте се за издателска услуга

Можете да регистрирате своя сървър за получаване на webhook чрез API или уеб интерфейс. За по-лесно ще се регистрираме чрез уеб интерфейса:

1. Да отидем в секцията с кофи в контролната зала.
2. Отидете до кофата, за която ще конфигурираме webhooks и щракнете върху зъбното колело:

Отидете в раздела Webhooks и щракнете върху Добавяне:

Попълнете полетата:

ID — името на webhook.

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

URL — адрес на сървър за получаване на webhook.

Префикс/суфикс на филтъра е филтър, който ви позволява да генерирате уебкукички само за обекти, чиито имена отговарят на определени правила. Например, за да може webhook да задейства само файлове с разширение .png, в Суфикс на филтъра трябва да напишете „png“.

Понастоящем само портове 80 и 443 се поддържат за достъп до сървъра за получаване на webhook.

Да щракнем Добавете кука и ще видим следното:

Добавена кука.

Сървърът за получаване на webhook показва в своите регистрационни файлове напредъка на процеса на регистрация на кука:

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

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

Описание на сървъра за получаване на webhook

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

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

Помислете за основните функции:

• Ping() - маршрут, който отговаря чрез URL/ping, най-простата реализация на сонда за живост.
• Webhook() - основен маршрут, URL/уеб манипулатор:
  • потвърждава регистрацията в услугата за публикуване (отидете на функцията SubscriptionConfirmation),
  • обработва входящи уебкукички (функция Gorecords).
• Функциите 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)
    }

}

Тази функция определя дали е пристигнала заявка за потвърждение на регистрация или webhook. Както следва от документация, ако регистрацията е потвърдена, в заявката за публикация се получава следната Json структура:

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

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

content-type: application/json

{"signature":«ea3fce4bb15c6de4fec365d36bcebbc34ccddf54616d5ca12e1972f82b6d37af»}

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

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

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

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

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

Няма да разглеждаме подробно функцията SubscriptionConfirmation; тя се изпълнява съгласно принципите, изложени в документация. Можете да видите изходния код за тази функция на проект git хранилища.

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

• име на кофата
• обектен ключ
• действие:
  • копие - ако в оригиналната заявка EventName = ObjectCreated | Поставете обект | PutObjectCopy
  • изтриване - ако в оригиналната заявка EventName = ObjectRemoved | Изтриване на обект

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

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

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

Работен пример

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

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

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

Нека конфигурираме достъпа до S3 MCS API:

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

Нека конфигурираме достъпа до API на AWS S3:

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

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

Към 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=""
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

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

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

Нека да видим как работи. През MCS уеб интерфейс добавете файла test.txt към кофата myfiles-ash. Регистрационните файлове на конзолата показват, че е направена заявка към сървъра на webhook:

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

Нека проверим съдържанието на контейнера 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] incoming HTTP request from 
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

Целият код, използван в тази статия, е в моето хранилище. Има също примери за скриптове и примери за преброяване на подписи за регистриране на webhooks.

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

Good Luck!

Още четива по темата:

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

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