서버리스, 람다, SQS를 이용한 프로젝트

<도넛-스테이츠>는 온라인으로 도너츠를 판매합니다.

웹사이트 통해서 주문 버튼을 누르는 것으로 구매(Sales API)가 가능합니다.

창고에 재고가 있다면 재고가 감소하고 구매가 완료됩니다.

한 유튜버가 도넛-스테이츠의 도너츠가 맛있다고 영상을 올려 주문이 급등하였습니다.

창고에 재고가 없어 구매가 불가능한 경우 제조 공장에 알려 다시 창고를 채우는 시스템을 구축해야합니다.

제조 공장인 <팩토리>에 주문을 요청 (Lagacy Factory API)할 수 있습니다.

주문이 요청되면 일정 시간이 지난 후 창고에 재고가 증가합니다.

구성요소

Sales API

2. Factory API

3. 프론트엔드 (웹사이트) : cURL / Postman / k6 등을 통한 API 호출로만 구현

- Sales API를 통해 백엔드에 요청

4. 백엔드 (서버) : 구매 시 창고에서 재고 확인 후 재고 감소 로직 구현

- 재고가 부족할 경우 Factory API를 통해 재고 확보 요청

5. 데이터베이스 (창고) : RDS에 mysql db 구성

- 요청에 따른 재고 상태 변경

Day 1 (Tutorial)

목표

Serverless 를 이용한 AWS 리소스 생성

메시지 Queue가 사용되는 구조 이해

Step 1 : Serverless 를 이용한 Lambda 생성

(Serverless framework 를 이용하여 Lambda 함수 생성 및 배포)

#1. Serverless 튜토리얼을 통한 프로젝트 생성

서버리스 프레임워크 설치 및 새 서비스 생성

% npm install -g serverless
// node와 npm이 설치된 상태에서 Serverless Framework 를 글로벌 모듈로 설치하는 것을 권장함

% serverless
// AWS - Node.js - Starter       // 생성 템플릿 선택
// Project name 은 원하는대로 작성
// Do you want to deploy now?  N   // 답변 no로 하기 - yes로 하면 기본 리전으로 바로 배포됨

AWS 자격증명 원하는 방식대로 진행 (필자는 로컬 AWS 자격 증명 사용함)
가끔 aws configure list 로 aws 설정 확인

#2. Serverless.yaml 레퍼런스를 참고하여 리전 변경

serverless.yaml 파일 수정

service: aws-node-project
frameworkVersion: '3'

provider:
  name: aws
  runtime: nodejs18.x
  region: ap-northeast-2    // 리전 서울로 변경

functions:
  function1:
    handler: index.hello
# The `events` block defines how to trigger the handler.helloWorld code
    events:
      - http:
          path: hello       // index.js 함수 연결
          method: post      // post 메소드를 확인 예정
          cors: true

#3. handler.js 편집 (서버구현)

module.exports.hello = async (event) => {
  let inputValue, outputValue
  console.log(event.body)

  if (event.body) {

    let body = JSON.parse(event.body)

// 추가 도전과제: body가 { input: 숫자 } 가 맞는지 검증하고, 검증에 실패하면 응답코드 400 및 에러 메시지를 반환하는 코드를 넣어봅시다.

    inputValue = parseInt(body.input)
    outputValue = inputValue + 1
  }

  const message = `메시지를 받았습니다. 입력값: ${inputValue}, 결과: ${outputValue}`

  return {
    statusCode: 200,
    body: JSON.stringify(
      {
        message
      },
      null,
      2
    ),
  };
};

#4. serverless deploy 를 통한 배포

serverless deploy

%cd <serverless 프로젝트 템플릿 폴더>
// serverless 프로젝트 진입 후 deploy 진행

% serverless deploy
// deploy 완료 후 AWS console 내 lambda, cloudWatch, s3 등 생성 확인

!https://blog.kakaocdn.net/dn/bhMqXc/btrZNutNhPo/gIqfWD7a4tSEqg2nxKxv2K/img.png

#5. cURL 을 통한 테스트 : 입력값의 +1 반환 확인

% curl -X POST https://API_GATEWAY_ID.execute-api.ap-northeast-2.amazonaws.com/dev/hello --header 'Content-type: application/json' --data-raw '{ "input": 1 }'

응답

{ "message": "메시지를 받았습니다. 입력값: 1, 결과: 2" }

#6. 튜토리얼 완료 후 서버리스 삭제

% serverless remove

// deploy로 배포된 것들 삭제됨

오류

{"message":"Missing Authentication Token"}

오류 출력 시 오타 또는 URL 뒷부분 /dev/hello 경로 확인

Serverless 튜토리얼 참고 링크

https://www.serverless.com/framework/docs/tutorial

Tutorial: Your First Serverless Framework Project
The Serverless Framework documentation for AWS Lambda, API Gateway, EventBridge, DynamoDB and much more.
www.serverless.com

Step 2 : Serverless를 이용한 Lambda -SQS - Lambda 구조 생성

#1. serverless 프레임워크 설치 확인

% sls --version

#2. serverless 프로젝트 생성

% serverless // sls => 프로젝트 생성// AWS - Node.js - SQS Worker// Project name 은 적당하게 설정// Do you want to deploy now? N // no 로 선택하기 - yes로 하면 기본값으로 deploy % cd <생성 폴더 이름>// sls 프로젝트 진입% code . // visual code 로 작업할 경우

#3. region 추가 (serverless.yaml)

provider:
  name: aws
  runtime: nodejs18.x
  region: ap-northeast-2

#4. 배포 진행 및 클라우드 포메이션 확인

% serverless deploy // jobsWorker (컨슈머) 와 producer 두 함수가 생성된다

#5. 생성된 함수 별 트리거 확인

Producer 트리거 -> API Gateway 생성 확인

JobsWorker 트리거 -> SQS 생성 확인

#6. Producer Test

소스코드를 확인해보니 body 라는 키-값을 필요로함

  if (!event.body) {
    return {

Producer Lambda 내 Test Event 생성 -> 테스트 실행

실행결과 확인

Jobs Worker Lambda의 View CloudWatch logs (최근 생성된 로그 스트림 내 hello-world)

#7. handler.js (index.js)편집 - 컨슈머 구현

이후 serverless deploy

const { SQSClient, SendMessageCommand } = require("@aws-sdk/client-sqs");
const sqs = new SQSClient();

const producer = async (event) => {
  let statusCode = 200;
  let message;

  if (!event.body) {
    return {
      statusCode: 400,
      body: JSON.stringify({
        message: "No body was found",
      }),
    };
  }

  try {
    await sqs.send(new SendMessageCommand({
      QueueUrl: process.env.QUEUE_URL,
      MessageBody: event.body,
      MessageAttributes: {
        AttributeName: {
          StringValue: "Attribute Value",
          DataType: "String",
        },
      },
    }));

    message = "Message accepted!";
  } catch (error) {
    console.log(error);
    message = error;
    statusCode = 500;
  }

  return {
    statusCode,
    body: JSON.stringify({
      message,
    }),
  };
};

const consumer = async (event) => {
  /*
  for (const record of event.Records) {
    const messageAttributes = record.messageAttributes;
    console.log(
      "Message Attribute: ",
      messageAttributes.AttributeName.stringValue
    );
    console.log("Message Body: ", record.body);
  }
  */
  for (const record of event.Records) {
    console.log("Message Body: ", record.body);

    let inputValue, outputValue
    // TODO: Step 1을 참고하여, +1 를 하는 코드를 넣으세요
    if (record.body) {

      let body = JSON.parse(record.body)

      // 추가 도전과제: body가 { input: 숫자 } 가 맞는지 검증하고, 검증에 실패하면 응답코드 400 및 에러 메시지를 반환하는 코드를 넣어봅시다.

      inputValue = parseInt(body.input)
      outputValue = inputValue + 1
    }

    const message = `메시지를 받았습니다. 입력값: ${inputValue}, 결과: ${outputValue}`
    console.log(message)

  }
};

module.exports = {
  producer,
  consumer,
};

#8. cURL을 통한 테스트

% curl -X POST https://API_GATEWAY_ID.execute-api.ap-northeast-2.amazonaws.com/produce --header 'Content-type: application/json' --data-raw '{ "input": 1 }'

응답

{"message":"Message accepted!"}

오류

{"message":"Not Found"}

오류 출력 시 URL 뒷부분 serverless.yaml 의 fucntions path인 /produce 경로 확인

#9. 쉘 스크립트의 반복문을 이용한 반복 실행 확인

#!/bin/bash

for i in {1..5}

do

echo curl -X POST https://API_GATEWAY_ID.execute-api.ap-northeast-2.amazonaws.com/ --header 'Content-type: application/json' --data-raw '{ "input": 1 }'

done

쉘스크립트 반복문 참고 링크

https://www.cyberciti.biz/faq/bash-for-loop/

#10. CloudWatch를 통해 컨슈머가 메시지를 소비하는 것 확인

Step 3 : DLQ 연결 및 K6 성능 테스트

DLQ 란?

Dead Letter Queue
메시징 시스템에서 메시지를 처리하지 못한 경우, 메시지를 보관하는 대기열을 의미함
DLQ는 메시지 처리 실패 시 수동으로 재처리하거나, 다른 서비스나 시스템으로 전달하여 처리함
메시지 처리 실패에 대한 안정성과 신뢰성을 향상시킬 수 있음

K6 란?

AWS에서 제공하는 로드 테스트 도구 중 하나인 K6를 실행하는 데 사용되는 AWS Lambda Layer

#1. deplay 함수 추가 및 소스코드 수정

기존 index.js 내 소스코드 수정
await delay(5000) 부분을 수정하며 테스트 진행 예정

function delay(time) {
  return new Promise(resolve => setTimeout(resolve, time));
}

const consumer = async (event) => {
  await delay(5000)
  for (const record of event.Records) {
    console.log("Message Body: ", record.body);

// 생략

#2. 재배포 후 람다 및 SQS, SQS DLQ 생성 확인 (queue 2개 생성 확인 - 프로듀서, 컨슈머)

추가로 CloudWatch 도 생성확인

% sls deploy// step 2 단계를 다시 deploy 한다. // 프로듀서와 컨슈머 생성 확인 (AWS Lambda)

#3. Lambda -> Application -> 리소스 에서 생성된 리소스 확인

#4. Producer 테스트

JSON 형식으로 -> body에 메시지를 작성하여 테스트 이벤트 실행

컨슈머 (jobs worker) 에서 body 정상 수신 확인

#5. 함수 실행 지연함수 delay 수정 후 재 테스트

index.js 내 await delay(5000)을 15000으로 수정 (15초) ==> 재배포 sls deploy
Producer 테스트 이벤트 재실행

CloudWatch 컨슈머 로그 확인

=> 타임아웃에 걸려 람다가 제대로 실행되지 않았단 뜻

=> 람다함수는 최소 15초 이상의 시간이 필요함 (delay를 15초를 걸었으므로)

=> 지금 이 람다 함수의 실행 제한 시간이 6초 이므로 타임아웃이 걸림

SQS 에 연결된 소비자는 주기적으로 폴링을 하기 때문에 같은오류가 반복적으로 출력됨

Producer 람다 -> 구성 -> 일반구성 -> 제한시간

=> 현재 6초로 설정되어 있음 (최대 15분)

#6. SQS 확인

메시지 소비 실패 시 jobs의 '이동중인 메시지'에 숫자가 카운팅 되며
반복적 메시지 소비 최종 실패 시 jobs-dlq 의 사용 가능한 메시지 숫자로 카운팅 된다

sqs내 jobs -> 클릭 -> 자세히 -> 내용 확인

=> 기본 표시 제한시간 Visibility Timeout (36s)

sqs 내 jobs -> 배달 못한 편지 대기열 -> 최대 수신 수 (3)

=> 이전 테스트에서 3회 시도 후 dlq 로 넘어갔던것 확인 했었음

#7. SQS DLQ 확인

SQS 내 jobs-dlq -> 클릭 -> 메시지 전송 및 수신

사용 가능한 메시지 확인 후 우측 메시지 폴링 클릭 (콘솔에서도 메시지 폴링이 가능하다)

하단부에서 미시지 확인

=> 이전에 작성했던 메시지 확인 가능

#8. SQS DLQ 원리

일반 큐에서 메시지를 수신 했으나 최종적으로 소비되지 못하여 (재시도 회수 까지도) 지워지지 않은 메시지가 DLQ로 넘어간것

#9. K6 성능 테스트 도구 활용 (프로듀서 반복 실행 및 테스트)

k6 설치

% brew install k6

제공받은 run.sh , single-request.k6.js

// run.sh
#!/bin/bash
k6 run -u 1 -i 100 ./single-request.k6.js

// single-request.k6.js
import http from 'k6/http';
import { sleep, check } from 'k6';

// you can specify stages of your test (ramp up/down patterns) through the options object
// target is the number of VUs you are aiming for

export const options = {
  stages: [
    { target: 20, duration: '20s' },
    { target: 15, duration: '20s' },
    { target: 0, duration: '20s' },
  ],
  thresholds: {
    http_reqs: ['count <= 100'],
  },
};

export let input = 1

export default function () {
  // our HTTP request, note that we are saving the response to res, which can be accessed later
  const payload = { input: input++ };
  const headers = {
    'Content-Type': 'application/json',
    'dataType': 'json'
  };
  const res = http.request('POST', 'https://API_GATEWAY_ID.execute-api.ap-northeast-2.amazonaws.com/produce',
  JSON.stringify(payload),  {
    headers: headers,
  });
  console.log(JSON.stringify(payload))
  sleep(0.1);

  const checkRes = check(res, {
    'status is 200': (r) => r.status === 200, // 기대한 HTTP 응답코드인지 확인합니다.
    'response body': (r) => r.body.indexOf('{"message":"Message accepted!"}') !== -1,  // 기대한 응답인지 확인합니다.
  });
}