S3 객체 스토리지 Mail.ru 클라우드 솔루션의 웹후크를 기반으로 하는 이벤트 기반 애플리케이션의 예

루브 골드버그 커피 머신

이벤트 기반 아키텍처는 필요한 순간에만 사용되기 때문에 사용되는 리소스의 비용 효율성을 높입니다. 이를 구현하고 추가 클라우드 엔터티를 작업자 애플리케이션으로 생성하지 않는 방법에 대한 다양한 옵션이 있습니다. 오늘은 FaaS가 아닌 웹훅에 대해 이야기하겠습니다. 객체 스토리지 웹후크를 사용하여 이벤트를 처리하는 튜토리얼 예제를 보여드리겠습니다.

객체 스토리지 및 웹후크에 대한 몇 마디입니다. 객체 스토리지를 사용하면 HTTP/HTTPS를 통해 S3 또는 다른 API(구현에 따라 다름)를 통해 액세스할 수 있는 객체 형태로 클라우드에 모든 데이터를 저장할 수 있습니다. 웹후크는 일반적으로 사용자 정의 HTTP 콜백입니다. 일반적으로 코드가 저장소로 푸시되거나 블로그에 댓글이 게시되는 등의 이벤트에 의해 트리거됩니다. 이벤트가 발생하면 원본 사이트는 Webhook에 지정된 URL로 HTTP 요청을 보냅니다. 결과적으로 한 사이트의 이벤트가 다른 사이트의 작업을 트리거하도록 할 수 있습니다(위키). 소스 사이트가 객체 스토리지인 경우 이벤트는 해당 콘텐츠에 대한 변경 사항으로 작용합니다.

이러한 자동화를 사용할 수 있는 간단한 사례의 예:

1. 다른 클라우드 스토리지에 있는 모든 객체의 복사본을 생성합니다. 파일이 추가되거나 변경될 때마다 즉시 복사본을 생성해야 합니다.
2. 그래픽 파일의 일련의 축소판 자동 생성, 사진에 워터마크 추가 및 기타 이미지 수정.
3. 새 문서 도착에 대한 알림(예: 분산 회계 서비스가 보고서를 클라우드에 업로드하고 재무 모니터링에서 새 보고서에 대한 알림을 받아 확인하고 분석함)
4. 약간 더 복잡한 경우에는 필요한 컨테이너가 포함된 포드를 생성하고 작업 매개변수를 전달하며 처리 후 컨테이너를 축소하는 Kubernetes에 대한 요청 생성이 포함됩니다.

예를 들어, MCS(Mail.ru Cloud Solutions) 객체 스토리지 버킷의 변경 사항이 웹후크를 사용하여 AWS 객체 스토리지에서 동기화되는 경우 작업 1의 변형을 만들어 보겠습니다. 실제 로드된 경우에는 큐에 웹후크를 등록하여 비동기 작업을 제공해야 하지만 훈련 작업의 경우 이를 구현하지 않고 수행합니다.

업무 계획

상호작용 프로토콜은 다음에 자세히 설명되어 있습니다. MCS의 S3 웹훅 가이드. 작업 계획에는 다음 요소가 포함됩니다.

• 출판 서비스는 S3 스토리지 측에 있으며 webnhook이 트리거될 때 HTTP 요청을 게시합니다.
• 웹훅 수신 서버, HTTP 게시 서비스의 요청을 수신하고 적절한 작업을 수행합니다. 서버는 어떤 언어로든 작성할 수 있습니다. 이 예에서는 Go로 서버를 작성하겠습니다.

S3 API에서 웹훅 구현의 특별한 기능은 게시 서비스에 웹훅 수신 서버를 등록하는 것입니다. 특히 웹훅 수신 서버는 게시 서비스의 메시지 구독을 확인해야 합니다(다른 웹훅 구현에서는 일반적으로 구독 확인이 필요하지 않음).

따라서 웹훅 수신 서버는 두 가지 주요 작업을 지원해야 합니다.

• 게시 서비스의 등록 확인 요청에 응답하고,
• 들어오는 이벤트를 처리합니다.

Webhook 수신 서버 설치

Webhook 수신 서버를 실행하려면 Linux 서버가 필요합니다. 이 문서에서는 예를 들어 MCS에 배포하는 가상 인스턴스를 사용합니다.

필요한 소프트웨어를 설치하고 웹훅 수신 서버를 실행해 보겠습니다.

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

웹훅 수신 서버로 폴더를 복제합니다.

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 또는 웹 인터페이스를 통해 웹훅 수신 서버를 등록할 수 있습니다. 단순화를 위해 웹 인터페이스를 통해 등록하겠습니다.

1. 버킷 섹션으로 이동해 보겠습니다. 통제실에서.
2. 웹훅을 구성할 버킷으로 이동하여 기어를 클릭합니다.

Webhooks 탭으로 이동하여 추가를 클릭합니다.

다음 필드를 입력하세요.

ID — 웹훅의 이름입니다.

이벤트 - 전송할 이벤트입니다. 파일 작업(추가 및 삭제) 시 발생하는 모든 이벤트의 전송을 설정했습니다.

URL - 웹훅 수신 서버 주소입니다.

필터 접두사/접미사는 이름이 특정 규칙과 일치하는 개체에 대해서만 웹훅을 생성할 수 있는 필터입니다. 예를 들어 웹후크가 확장자가 .png인 파일만 트리거하려면 필터 접미사 "png"를 써야 합니다.

현재 웹훅 수신 서버 접속은 80번과 443번 포트만 지원됩니다.

클릭해보자 후크 추가 그리고 우리는 다음을 보게 될 것입니다:

후크가 추가되었습니다.

웹훅 수신 서버는 후크 등록 프로세스의 진행 상황을 로그에 표시합니다.

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는 서명 계산을 위한 256진수 문자열로 출력되는 HMAC-SHA256 및 HMAC-SHA16 암호화 알고리즘을 구현한 것입니다.
• 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)
    }

}

등록 확인 요청이나 웹훅이 도착했는지 확인하는 기능입니다. 다음과 같이 선적 서류 비치, 등록이 확인되면 Post 요청에서 다음 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 요청의 구조는 다음과 같습니다.

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", 구독 확인 요청에는 존재하고 웹훅에는 존재하지 않기 때문입니다. POST 요청에 이 항목이 있는지 여부에 따라 프로그램의 추가 실행은 다음 중 하나의 함수로 이동합니다. SubscriptionConfirmation, 또는 함수로 GotRecords.

SubscriptionConfirmation 기능은 자세히 고려하지 않으며, 다음에 명시된 원칙에 따라 구현됩니다. 선적 서류 비치. 이 함수의 소스 코드는 다음에서 볼 수 있습니다. 프로젝트 Git 저장소.

GotRecords 함수는 들어오는 요청을 구문 분석하고 각 Record 객체에 대해 매개변수를 사용하여 외부 스크립트(이름이 -script 매개변수로 전달됨)를 호출합니다.

• 버킷 이름
• 객체 키
• 동작:
  • 복사 - 원래 요청에 있는 경우 EventName = ObjectCreated | PutObject | PutObject복사
  • 삭제 - 원래 요청에 있는 경우 EventName = ObjectRemoved | 객체 삭제

따라서 설명된 대로 Post 요청과 함께 후크가 도착하면 이상, -script=script.sh 매개변수를 사용하면 스크립트는 다음과 같이 호출됩니다.

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

이 웹훅 수신 서버는 완전한 프로덕션 솔루션이 아니라 가능한 구현의 단순화된 예라는 점을 이해해야 합니다.

작품의 예

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 버킷에 추가합니다. 콘솔 로그는 웹훅 서버에 대한 요청이 이루어졌음을 보여줍니다.

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

이제 웹 인터페이스를 통해 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

이 글에 사용된 모든 코드는 내 저장소에. 웹후크 등록을 위한 스크립트의 예와 서명 계산의 예도 있습니다.

이 코드는 활동에서 S3 웹후크를 사용하는 방법의 예일 뿐입니다. 처음에 말했듯이 이러한 서버를 프로덕션에서 사용할 계획이라면 적어도 비동기 작업을 위해 서버를 다시 작성해야 합니다. 즉, 들어오는 웹후크를 대기열(RabbitMQ 또는 NATS)에 등록하고 거기서 구문 분석하여 처리해야 합니다. 작업자 애플리케이션으로. 그렇지 않으면 웹훅이 대량으로 도착할 때 작업을 완료하는 데 서버 리소스가 부족할 수 있습니다. 대기열이 있으면 서버와 작업자를 분산시킬 수 있을 뿐만 아니라 오류 발생 시 반복 작업으로 인한 문제를 해결할 수도 있습니다. 또한 로깅을 보다 상세하고 표준화된 로깅으로 변경하는 것이 좋습니다.

행운을 빕니다!

주제에 대한 추가 자료:

• Mail.ru Cloud Solutions S3 객체 스토리지를 파일 시스템으로 사용합니다.
• 경쟁 우위를 확보하기 위해 객체 스토리지를 사용하는 방법

출처 : habr.com