S3 オブジェクト ストレージの Webhook に基づくイベント駆動型アプリケーションの例 Mail.ru クラウド ソリューション

ルーブ ゴールドバーグ コーヒーマシン

イベント駆動型アーキテクチャでは、リソースは必要なときにのみ使用されるため、使用されるリソースのコスト効率が向上します。 これを実装し、ワーカー アプリケーションとして追加のクラウド エンティティを作成しない方法については、多くのオプションがあります。 そして今日は FaaS についてではなく、Webhook について話します。 オブジェクト ストレージ Webhook を使用してイベントを処理するチュートリアルの例を示します。

オブジェクト ストレージと Webhook について少し説明します。 オブジェクト ストレージを使用すると、あらゆるデータをオブジェクトの形式でクラウドに保存でき、S3 または別の API (実装に応じて) HTTP/HTTPS 経由でアクセスできます。 Webhook は通常、カスタム HTTP コールバックです。 これらは通常、リポジトリにプッシュされたコードやブログに投稿されたコメントなどのイベントによってトリガーされます。 イベントが発生すると、発信元サイトは Webhook に指定された URL に HTTP リクエストを送信します。 その結果、あるサイトでのイベントが別のサイトでのアクションをトリガーすることができます (ウィキ）。 ソース サイトがオブジェクト ストレージである場合、イベントはそのコンテンツの変更として機能します。

このような自動化を使用できる単純なケースの例:

1. 別のクラウド ストレージにすべてのオブジェクトのコピーを作成します。 ファイルが追加または変更されるたびに、その場でコピーを作成する必要があります。
2. グラフィック ファイルの一連のサムネイルの自動作成、写真への透かしの追加、その他の画像の変更。
3. 新しいドキュメントの到着に関する通知 (たとえば、分散会計サービスはレポートをクラウドにアップロードし、財務監視は新しいレポートに関する通知を受信し、それらを確認して分析します)。
4. もう少し複雑なケースには、たとえば、Kubernetes へのリクエストの生成が含まれます。これにより、必要なコンテナーを含むポッドが作成され、そこにタスク パラメーターが渡され、処理後にコンテナーが折りたたまれます。

例として、Mail.ru クラウド ソリューション (MCS) オブジェクト ストレージ バケットの変更が Webhook を使用して AWS オブジェクト ストレージで同期されるタスク 1 の変形例を作成します。 実際にロードするケースでは、Webhookをキューに登録することで非同期作業を提供する必要がありますが、トレーニングタスクではこれを行わずに実装します。

作業スキーム

インタラクションプロトコルについては、以下で詳しく説明されています。 MCS での S3 Webhook のガイド。 作業スキームには次の要素が含まれます。

• 出版サービスこれは S3 ストレージ側にあり、webnhook がトリガーされたときに HTTP リクエストを発行します。
• Webhook受信サーバー、HTTP 公開サービスからのリクエストをリッスンし、適切なアクションを実行します。 サーバーは任意の言語で作成できます。この例では、Go でサーバーを作成します。

S3 API での Webhook 実装の特別な機能は、公開サービスへの Webhook 受信サーバーの登録です。 特に、Webhook 受信サーバーは、公開サービスからのメッセージへのサブスクリプションを確認する必要があります (他の Webhook 実装では、通常、サブスクリプションの確認は必要ありません)。

したがって、Webhook 受信サーバーは XNUMX つの主要な操作をサポートする必要があります。

• 出版サービスの登録確認リクエストに応答し、
• 受信イベントを処理します。

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

出版サービスに登録する

API または Web インターフェイスを介して Webhook 受信サーバーを登録できます。 簡単にするために、Web インターフェイス経由で登録します。

1. バケットセクションに行きましょう 制御室で。
2. Webhook を設定するバケットに移動し、歯車をクリックします。

「Webhooks」タブに移動し、「追加」をクリックします。

次のフィールドに入力します。

ID — Webhook の名前。

イベント - どのイベントを送信するか。 ファイルの操作 (追加および削除) 時に発生するすべてのイベントの送信を設定しました。

URL — Webhook 受信サーバーのアドレス。

フィルター プレフィックス/サフィックスは、名前が特定のルールに一致するオブジェクトに対してのみ Webhook を生成できるフィルターです。 たとえば、Webhook が拡張子 .png を持つファイルのみをトリガーするには、次のようにします。 フィルタのサフィックス 「png」と書く必要があります。

現在、Webhook 受信サーバーへのアクセスにはポート 80 と 443 のみがサポートされています。

クリックしましょう フックを追加する そして、次のことがわかります。

フックを追加しました。

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 を介して応答するルート、liveness プローブの最も単純な実装。
• Webhook() - メインルート、URL/Webhook ハンドラー:
  • 公開サービスへの登録を確認します (Subscriptionconfirmation 関数に移動します)。
  • 受信した Webhook を処理します (Gorecords 機能)。
• 関数 HmacSha256 および HmacSha256hex は、HMAC-SHA256 および HMAC-SHA256 暗号化アルゴリズムの実装であり、署名を計算するための 16 進数の文字列として出力されます。
• main は 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 リクエストで受信されます。

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

Webhook が到着すると、Post リクエストの構造は次のようになります。

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 パラメーターで渡された名前) を呼び出します。

• バケット名
• オブジェクトキー
• アクション：
  • copy - 元のリクエストの場合 EventName = ObjectCreated | オブジェクトを置く | PutObjectCopy
  • delete - 元のリクエストの場合 EventName = ObjectRemoved | オブジェクトの削除

したがって、説明したように、フックが Post リクエストとともに到着すると、 上記、パラメータ -script=script.sh を指定すると、スクリプトは次のように呼び出されます。

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

この Webhook 受信サーバーは完全な運用ソリューションではなく、可能な実装の簡略化された例であることを理解してください。

作品例

MCS のメイン バケットから AWS のバックアップ バケットにファイルを同期してみましょう。 メイン バケットは myfiles-ash と呼ばれ、バックアップ バケットは myfiles-backup と呼ばれます (AWS でのバケット構成については、この記事の範囲外です)。 したがって、ファイルがメインバケットに配置されると、そのコピーがバックアップバケットに表示され、メインバケットから削除されると、バックアップバケットから削除される必要があります。

MCS クラウド ストレージと AWS クラウド ストレージの両方と互換性のある awscli ユーティリティを使用してバケットを操作します。

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

AWS S3 API へのアクセスを設定しましょう。

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

AWS の myfiles-backup バケットの内容を確認してみましょう。

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

ここで、Web インターフェイスを介して、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

この記事で使用されているすべてのコードは、 私のリポジトリで。 Webhook を登録するためのスクリプトの例と署名のカウントの例もあります。

このコードは、アクティビティで S3 Webhook を使用する方法の例にすぎません。 冒頭で述べたように、このようなサーバーを運用環境で使用する予定がある場合は、少なくとも非同期作業用にサーバーを書き直す必要があります。つまり、受信 Webhook をキュー (RabbitMQ または NATS) に登録し、そこからそれらを解析して処理します。ワーカーアプリケーションを使用します。 そうしないと、Webhook が大量に到着したときに、タスクを完了するためのサーバー リソースが不足する可能性があります。 キューの存在により、サーバーとワーカーを分散できるだけでなく、障害が発生した場合の繰り返しタスクの問題を解決することができます。 また、ログをより詳細で標準化されたものに変更することをお勧めします。

グッドラック！

このトピックについてさらに読む:

• Mail.ru クラウド ソリューション S3 オブジェクト ストレージをファイル システムとして使用します。
• 競合他社に勝つためのオブジェクト ストレージの使用方法。

出所： habr.com