GDmarket-Premium

GDmarket : 근대마켓 - 근거리 대여 마켓 (Premium)

Table of contents

• 근대마켓
  • 체크포인트
  • 서비스 시나리오
    • 기능적 요구사항
    • 비기능적 요구사항
  • 설계
    • 기본 모델
    • Premium 모델
  • 구현
    • Saga
    • CQRS
    • Correlation
    • Req/Res
    • Gateway
    • 구현 검증
  • 운영
    • Deploy
    • CirCuit Breaker
    • Autoscale
    • 무정지 재배포
    • Config Map
    • Polyglot
    • Self-healing

체크포인트

1. Saga
2. CQRS
3. Correlation
4. Req/Res
5. Gateway
6. Deploy / Pipeline
7. Circuit Breaker
8. Autoscale (HPA)
9. Zero-downtime deploy (Readiness Probe)
10. Config Map/ Persistence Volume
11. Polyglot
12. Self-healing (Liveness Probe)

서비스 시나리오

기능적 요구사항

1. 물건관리자는 물건을 등록할 수 있다
2. 물건관리자는 물건을 삭제할 수 있다.
3. 대여자는 물건을 선택하여 예약한다.
4. 대여자는 예약을 취소할 수 있다.
5. 예약이 완료되면 해당 물건은 대여불가 상태로 변경된다.
6. 대여자가 결제한다.
7. 대여자는 결제를 취소할 수 있다.
8. 물건관리자는 물건을 대여해준다.
9. 대여자가 대여요청을 취소할 수 있다.
10. 물건이 반납되면 물건은 대여가능 상태로 변경된다.
11. 물건관리자는 물건 통합상태를 중간중간 조회할 수 있다.
12. (Premium) 물건이 등록되면 등록 알람이 발생한다.
13. (Premium) 물건이 삭제되면 삭제 알람이 발생한다.

비기능적 요구사항

1. 트랜잭션
   1. 결제승인이 되지 안은 건은 결제요청이 완료되지 않아야한다. Sync 호출
   2. 등록 알람이 발생되지 않은 건은 물건등록이 완료되지 않아야 한다. Sync 호출 (Premium)
2. 장애격리
   1. 물건관리시스템이 수행되지 않더라도 대여 요청은 365일 24시간 받을 수 있어야 한다. > Async
   2. 결제시스템이 과중되면 결제요청을 잠시동안 받지 않고 결제를 잠시 후에 하도록 유도한다. > Circuit breaker
   3. (Premium) 알람시스템이 수행되지 않더라도 물건 삭제 요청은 365일 24시간 받을 수 있다 > Async
   4. (Premium) 알람시스템이 과중되면 물건등록을 잠시동안 받지 않고 알람을 잠시 후에 하도록 유도한다. > Circuit breaker
3. 성능
   1. 물건관리자가 등록한 물건의 통합상태를 별도로 확인할 수 있어야 한다. > CQRS

설계

기본 모델

• item, reservation, payment 서비스 구현

Premium 모델

• alarm 서비스가 추가됨

구현

Saga

• Pub/Sub을 구현한다.
• (Pub) 호출 서비스 및 Event : item / item이 삭제됨

   // item > Item.java
   
    @PreRemove
    public void onPreRemove() {
        ItemDeleted itemDeleted = new ItemDeleted();
        itemDeleted.setItemNo(this.getItemNo());
        itemDeleted.setAlarmStatus("DeleteAlerted");

        ObjectMapper objectMapper = new ObjectMapper();
        String json = null;
        try {
            json = objectMapper.writeValueAsString(itemDeleted);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("JSON format exception", e);
        }
        KafkaProcessor processor = ItemApplication.applicationContext.getBean(KafkaProcessor.class);
        MessageChannel outputChannel = processor.outboundTopic();
        outputChannel.send(org.springframework.integration.support.MessageBuilder
                .withPayload(json)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());
        System.out.println("@@@@@@@ itemDeleted to Json @@@@@@@");
        System.out.println(itemDeleted.toJson());
    }

• (Sub) 피호출 서비스 및 Policy : alarm / alarm 상태 변경 (Alerted -> DeleteAlerted)

   // alarm > PolicyHandler.java
   
   @StreamListener(KafkaProcessor.INPUT)
    public void wheneverItemDeleted_(@Payload ItemDeleted itemDeleted){
        if(itemDeleted.isMe()){
            System.out.println("##### listener  : " + itemDeleted.toJson());
            if("DeleteAlerted".equals(itemDeleted.getAlarmStatus())){
                Alarm alarm = (Alarm) alarmManagementRepository.findByItemNo(itemDeleted.getItemNo()).get(0);
                alarm.setAlarmStatus("DeleteAlerted");
                alarmManagementRepository.save(alarm);
            }
        }
    }

CQRS

• command와 query의 역할을 분리한다. (view 구현)
• item 이 등록될 때 알람상태(AlarmStatus)를 View를 통해 확인할 수 있도록 구현
• item 코드 구현

// item > Item.java

    @PostPersist
    public void onPostPersist(){
        ItemRegistered itemRegistered = new ItemRegistered();
        itemRegistered.setItemNo(this.getItemNo());
        itemRegistered.setItemName(this.getItemName());
        itemRegistered.setItemPrice(this.getItemPrice());
        itemRegistered.setItemStatus("Rentable");
        itemRegistered.setRentalStatus("NotRenting");
        itemRegistered.setAlarmStatus("Alerted");
        
        // view를 위해 Kafka Send
        ObjectMapper objectMapper = new ObjectMapper();
        String json = null;
        try {
            json = objectMapper.writeValueAsString(itemRegistered);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("JSON format exception", e);
        }
        KafkaProcessor processor = ItemApplication.applicationContext.getBean(KafkaProcessor.class);
        MessageChannel outputChannel = processor.outboundTopic();
        outputChannel.send(org.springframework.integration.support.MessageBuilder
                .withPayload(json)
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
                .build());

        System.out.println("@@@@@@@ ItemRegistered to Json @@@@@@@");
        System.out.println(itemRegistered.toJson());
   }

• view 코드 구현

// item > ItemInfoViewHandler.java

   @StreamListener(KafkaProcessor.INPUT)
    public void whenItemRegistered_then_CREATE_1 (@Payload ItemRegistered itemRegistered) {
        try {
            if (itemRegistered.isMe()) {
                // view 객체 생성
                ItemInfo itemInfo= new ItemInfo();
                // view 객체에 이벤트의 Value 를 set 함
                itemInfo.setItemNo(itemRegistered.getItemNo());
                itemInfo.setItemName(itemRegistered.getItemName());
                itemInfo.setItemStatus(itemRegistered.getItemStatus());
                itemInfo.setItemPrice(itemRegistered.getItemPrice());
                itemInfo.setAlarmStatus(itemRegistered.getAlarmStatus());
                // view 레파지 토리에 save
                itemInfoRepository.save(itemInfo);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

Correlation

• 각 마이크로 서비스는 상호 관련 키를 갖는다.
• CQRS 구현을 위해, Alarm과 ItemInfo는 상호 관련 키 'itemNo', 'alarmStatus'를 갖는다.
• Alarm.java
  
• ItemInfo.java
  

Req/Res

• Sync 호출을 구현한다.
• (Req) 호출 서비스 구현

// item.java > onPostPersist()

// alarm REQ/RES
System.out.println("@@@ Alarm @@@");
System.out.println("@@@ ItemNo : " + getItemNo());

gdmarketpremium.external.Alarm alarm = new gdmarketpremium.external.Alarm();
alarm.setAlarmStatus("Alerted");
alarm.setAlarmNo(getItemNo());
alarm.setItemNo(getItemNo());

• (Res) 피호출 서비스 구현

// AlarmService.java

@FeignClient(name="alarm", url="${api.alarm.url}")
public interface AlarmService {
 @RequestMapping(method= RequestMethod.POST, path="/alarms")
    public void alert(@RequestBody Alarm alarm);
}

Gateway

• 각 마이크로서비스는 gateway를 통해서 호출할 수 있다
• gateway 서비스 > application.yml 파일에 구현한다.
• local 세팅
  
• docker 세팅
  

구현 검증

• item 등록함

http POST item:8080/items/ itemName=Camera itemPrice=100 itemStatus=Rentable rentalStatus=NotRenting

• Sync 호출로 alarm 생성됨 (Req/Res)

http alarm:8080/alarms

• CQRS : alarmStatus를 확인할 수 있음

http item:8080/itemInfoes

• item 삭제함

http DELETE item:8080/items/1

• Async 호출로 alarm 상태 변경됨 (Alerted -> DeleteAlerted) (Pub/Sub)

http alarm:8080/alarms

• gateway로 reservation 서비스 GET 호출

http gateway:8080/reservations

운영

Deploy

• (1-1) 컨테이너라이징 : 디플로이 생성 확인 (gateway 서비스는 api 사용)

kubectl create deploy gateway --image=gdpremiumacr.azurecr.io/gateway:latest

• (1-2) 컨테이너라이징 : 디플로이 생성 확인 (그 외 서비스는 yml 사용)

kubectl apply -f kubernetes/deployment.yml

• (2) 컨테이너라이징: 서비스 생성 확인 (gateway 포함모든 서비스 동일)

kubectl expose deploy gateway --type="ClusterIP" --port=8080

• 모든 서비스가 잘 Running 됨을 확인

kubectl get all

CirCuit Breaker

• CirCuit Breaker Framework : Spring FeignClient + Hystrix 사용
• (호출) item 서비스 > application.yml : timeoutInMilliseconds 610 으로 설정

hystrix:
  command:
    default:
      execution.isolation.thread.timeoutInMilliseconds: 610

• (피호출) alarm 서비스 > Alarm.java : sleep 440 + 랜덤 220 으로 설정

Thread.currentThread().sleep((long) (400 + Math.random() * 220));

• siege 툴을 통한 서킷 브레이커 동작 확인
• 동시사용자 10명 , 30초 동안 siege 부하 테스트 실시

siege -c10 -t30S -r10 -v --content-type "application/json" 'http://item:8080/items POST {"itemName":"Camera"}'

Autoscale

• item 시스템에 대한 replica 를 동적으로 늘려주도록 HPA 를 설정한다.
• item 시스템의 resource 에 제한을 걸어둔다.

// item > deployment.yml

 resources:
   limits:
     cpu: 500m
   requests:
     cpu: 200m

• HPA 설정은 CPU 사용량이 15프로를 넘어서면 replica 를 10개까지 늘려준다.
• CirCuit Breaker와 동일한 방법으로 워크로드를 50초 걸어준다.

 kubectl autoscale deploy reservation --min=1 --max=10 --cpu-percent=15
 kubectl exec -it pod/siege-5459b87f86-wpwkt -c siege -- /bin/bash
 siege -c250 -t50S -r1000 -v --content-type "application/json" 'http://item:8080/items POST {"itemName":"Camera"}'

• 오토스케일 모니터링을 한다.

kubectl get deploy reservation -w

무정지 재배포

Config Map

• alarm 서비스를 REQ로 호출하는 item 서비스에 config map을 구현한다.
• item > application.yml : local
  
• item > application.yml : docker
  
• item > deployment.yml : env 세팅
  
• item > external > AlarmService.java : env 사용
  
• config map 생성 및 확인

# 생성
kubectl create configmap newurl --from-literal=url=http://alarm:8080

# 확인
kubectl get configmap newurl -o yaml

Polyglot

• 다형성을 만족하도록 구현한다.

• item, reservation 서비스는 H2 DB로 구현하고, 그와 달리 payment, alarm 서비스의 경우 Hsql DB로 구현한다.

• item, reservation 서비스의 pom.xml 설정
  

• payment, alarm 서비스의 pom.xml 설정
  

Self-healing