장애 처리와 Microservice 분산 추적

마이크로 서비스는 하나의 어플리케이션이 아니라 여러개의 서비스로 나눠져서 개발되다보니 각각의 서비스에서 문제가 생길경우 어떻게 처리하고 시작점이 어디고 끝났을때 반환을 어디에 해야할지 흐름이 중요하다.

 

CircuitBreaker와 Resilience4j 사용

Microservice 통신 시 연쇄 오류

현재 order-service를 실행시키지 않고 요청했을 경우 getOrders()를 호출하는 과정에서 에러가 발생하여 user-service로부터 500에러를 반환받게 된다. getOrders() 가 에러가 발생하면서 500 반환되는게 맞는거 같지만 하나의 서비스가 멈춘다고해서 모든 서비스가 정지 되면 안된다. 그래서 CircuitBreaker를 활용해서 microservice 사이의 호출에서 에러가 발생했을 경우 처리해줘야 한다.

 

CircuitBreaker 

• 장애가 발생하는 서비스에 반복적인 호출이 되지 못하게 차단
• 특정 서비스가 정상적으로 동작하지 않을 경우 다른 기능으로 대체 수행 -> 장애 회피

• 초기 Service3 호출 정상 동작 상태시에는 Closed
• Service3 호출 시 이상 발생 한 경우 Open 상태로 변경되고 접속 차단(fail fast)
            -> Service3의 호출 결과는 fallback으로 정의된 내용이 return 된다.
• 주기적으로 서비스 상태 확인하여 정상 확인될 경우 Close 상태로 변경 

https://luvstudy.tistory.com/150

 

Resilience4j

Resilience4J는 Java 전용으로 개발된 경량화된 Fault Tolerance(장애감내) 제품입니다. 
Resilience4J는 아래 6가지 핵심모듈로 구성되어 있습니다. 
- Circuit Breaker: Count(요청건수 기준) 또는 Time(집계시간 기준)으로 Circuit Breaker제공
- Bulkhead: 각 요청을 격리함으로써, 장애가 다른 서비스에 영향을 미치지 않게 함(bulkhead-격벽이라는 뜻)
- RateLimiter: 요청의 양을 조절하여 안정적인 서비스를 제공. 즉, 유량제어 기능임. 
- Retry: 요청이 실패하였을 때, 재시도하는 기능 제공
- TimeLimiter: 응답시간이 지정된 시간을 초과하면 Timeout을 발생시켜줌
- Cache: 응답 결과를 캐싱하는 기능 제공

 

 

Dependencies (User service)

Circuit Breaker만 사용하기 때문에 Circuit Breaker 만 추가

 - spring-cloud-starter-circuitbreaker-resilience4j

<!-- Resilience4j-->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-circuitbreaker-resilience4j</artifactId>
</dependency>

 

UserServiceImple.java

 

Resilience4JCircuitBreakerFactory는 dependencies 추가해서 Bean으로 사용가능하고 따로 Custom 해서 사용할수 도 있다.

문제가 생기면 사이즈 0인 리스트를 반환해서 내역이 없는 형태로 조회하도록 반환

@Service
@Slf4j
public class UserServiceImpl implements UserService {

    UserRepository userRepository;
    BCryptPasswordEncoder passwordEncoder;
    Environment env;
    OrderServiceClient orderServiceClient;
    CircuitBreakerFactory circuitBreakerFactory;

    @Autowired
    public UserServiceImpl(UserRepository userRepository, BCryptPasswordEncoder passwordEncoder, Environment env,
                           OrderServiceClient orderServiceClient, CircuitBreakerFactory circuitBreakerFactory) {
        this.userRepository = userRepository;
        this.passwordEncoder = passwordEncoder;
        this.env = env;
        this.orderServiceClient = orderServiceClient;
        this.circuitBreakerFactory = circuitBreakerFactory;
    }

    @Override
    public UserDto getUserByUserId(String userId) {
        UserEntity userEntity = userRepository.findByUserId(userId);

        if (userEntity == null) {
            throw new UsernameNotFoundException("User not found");
        }

        UserDto userDto = new ModelMapper().map(userEntity, UserDto.class);

//        List<ResponseOrder> ordersList = orderServiceClient.getOrders(userId);
        CircuitBreaker circuitbreaker = circuitBreakerFactory.create("circuitbreaker");
        List<ResponseOrder> ordersList = circuitbreaker.run(() -> orderServiceClient.getOrders(userId),
                throwable -> new ArrayList<>());

        userDto.setOrders(ordersList);

        return userDto;
    }
}

 

에러는 발생하지만 정상적으로 orders 목록은 조회가 되지 않고 반환된다.

CustomFactory를 생성

Customizer를 import할때 Resilience4J package에서 임포트해야한다!

 

package com.example.userservice.config;

import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.timelimiter.TimeLimiterConfig;
import org.springframework.cloud.circuitbreaker.resilience4j.Resilience4JCircuitBreakerFactory;
import org.springframework.cloud.circuitbreaker.resilience4j.Resilience4JConfigBuilder;
import org.springframework.cloud.client.circuitbreaker.Customizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
public class Resilience4JConfig {

    @Bean
    public Customizer<Resilience4JCircuitBreakerFactory> globalCustomConfiguration(){

        CircuitBreakerConfig circuitBreakerConfig = CircuitBreakerConfig.custom()
                .failureRateThreshold(4)
                .waitDurationInOpenState(Duration.ofMillis(1000))
                .slidingWindowType(CircuitBreakerConfig.SlidingWindowType.TIME_BASED)
                .slidingWindowSize(2)
                .build();

        TimeLimiterConfig timeLimiterConfig = TimeLimiterConfig.custom()
                .timeoutDuration(Duration.ofSeconds(4))
                .build();

        return factory -> factory.configureDefault(id -> new Resilience4JConfigBuilder(id)
                .timeLimiterConfig(timeLimiterConfig)
                .circuitBreakerConfig(circuitBreakerConfig)
                .build()

        );
    }
}

CircuitBreakerConfig

• failureRateThreshold : CircuitBreaker를 열지 결정하는 failure ratethreshold percentage (default : 50)
• waitDurationInputOpenState : CicuitBreaker를 open 한 상태를 유지하는 지속 기간을 의미, 이 기간 이후에 half-open 상태 (default : 60 seconds)
• sildingWindowType : CircuitBreaker 가 닫힐 때 통화 결과를 기록하는데 사용되는 슬라이딩 창의 유형을 구성 (카운트 기반 or 시간 기반)
• sildingWindowSize : CircuitBreaker 가 닫힐 때 호출 결과를 기록하는데 사용되는 슬라이딩 창의 크기를 구성 (default : 100)

 

TimeLimiterConfig

• TimeLimiter : future supplier의 time limit을 정하는 API (default : 1 seconds)

 

 

 

 

Microservice 분산 추적 - Zipkin, Sleuth

해당 요청 정보가 어떻게 실행되고 어떤 마이크로서비스를 거치는지 추적하도록 한다.

 

Zipkin

• Twitter에서 사용하는 분산 환경의 Timing 데이터 수집, 추적 시스템 (오픈소스)
• Google Drapper에서 발전하였으며, 분산환경에서의 시스템 병목 현상 파악
• Collector, Query Service, Database, WebUI로 구성
• Span - 하나의 요청에 사용되는 작업의 단위 (64bit unique ID)
• Trace - 트리 구조로 이뤄진 Span 셋, 하나의 요청에 대한 같은 Trace ID 발급

 

 

Spring Cloud Sleuth

Zipkin 과 연동해서 가지고 있는 로그 데이터나 스트리밍 데이터를 Zipkin 에 전달하는 기능

• 스프링 부트 어플리케이션을 Zipkin과 연동
• 요청 값에 따른 Trace ID, Span ID 부여
• Trace와 Span IDs를 로그에 추가 가능
  • servlet filter
  • rest template
  • scheduled actions
  • message channels
  • feign client

• trace id란 zipkin에서 추적할 때 사용자의 한번의 요청 단위를 trace id라고 한다.
• span id란 사용자의 한번의 요청 단위에서 내부적으로 추가로 요청되는 microservice 단위를 span id라고 한다.

 

# 설치
curl -sSL http://zipkin.io/quickstart.sh | bash -s
 
# 실행 
java -jar zipkin.jar

 

Users Microservice 수정

Dependencies 

 - spring-cloud-starter-sleuth

 - spring-cloud-starter-zipkin

<!-- Zipkin -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
    <version>2.2.3.RELEASE</version>
</dependency>

 

application.yml

spring:
  application:
    name: user-service
  zipkin:
    base-url: http://127.0.0.1:9411
    enabled: true
  sleuth:
    sampler:
      probability: 1.0

 

 

UsersServiceImpl.java 로그 추가

log.info("Before call orders microservice");
CircuitBreaker circuitbreaker = circuitBreakerFactory.create("circuitbreaker");
List<ResponseOrder> ordersList = circuitbreaker.run(() -> orderServiceClient.getOrders(userId),
        throwable -> new ArrayList<>());
log.info("After call orders microservice");

 

Orders Microservice 수정

Dependencies 

 - spring-cloud-starter-sleuth

 - spring-cloud-starter-zipkin

<!-- Zipkin -->
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-sleuth</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-starter-zipkin</artifactId>
    <version>2.2.3.RELEASE</version>
</dependency>

 

application.yml

spring:
  application:
    name: order-service
  zipkin:
    base-url: http://127.0.0.1:9411
    enabled: true
  sleuth:
    sampler:
      probability: 1.0

 

 

OrderController.java 로그 추가

    @PostMapping("/{userId}/orders")
    public ResponseEntity<ResponseOrder> createOrder(@PathVariable("userId") String userId, @RequestBody RequestOrder orderDetails) {
        log.info("Before add orders data");

        ModelMapper mapper = new ModelMapper();
        mapper.getConfiguration().setMatchingStrategy(MatchingStrategies.STRICT);

        OrderDto orderDto = mapper.map(orderDetails, OrderDto.class);


        orderDto.setUserId(userId);

        // JPA
        OrderDto createOrder = orderService.createOrder(orderDto);
        ResponseOrder responseOrder = mapper.map(createOrder, ResponseOrder.class);

        // Kafka
//        orderDto.setOrderId(UUID.randomUUID().toString());
//        orderDto.setTotalPrice(orderDetails.getQty() * orderDetails.getUnitPrice());
//
//        // kafka 에 주문 전달
//        kafkaProducer.send("example-catalog-topic", orderDto);
//        orderProducer.send("orders", orderDto);
//
//
//        ResponseOrder responseOrder = mapper.map(orderDto, ResponseOrder.class);
        log.info("Before add orders data");

        return ResponseEntity.status(HttpStatus.CREATED).body(responseOrder);
    }

    @GetMapping("/{userId}/orders")
    public ResponseEntity<List<ResponseOrder>> createOrder(@PathVariable("userId") String userId) {
        log.info("Before retrieve orders data");
        Iterable<OrderEntity> orderList = orderService.getOrdersByUserId(userId);

        List<ResponseOrder> result = new ArrayList<>();
        orderList.forEach(v -> {
            result.add(new ModelMapper().map(v, ResponseOrder.class));
        });

        log.info("after retrieve orders data");

        return ResponseEntity.ok().body(result);
    }

 

주문 등록시 로그 확인

TraceID와 SpanID 확인 가능

 

Zipkin 확인 (localhost:9411)

 

사용자 정보확인 호출 확인

orderService

 

userService

하나의 요청에서 실행되었기 때문에 TraceID는 같고 microService 단위의 요청인 SpanID는 다른게 출력된 걸 확인할 수 있다.

 

 

 

순차적으로 실행된 내역 확인 가능

 

서비스도로 조회가능

Dependencies는 호출한 관계를 확인할 수 있다.

 

 

오류가 발생했을 경우

오류 강제로 발생에서 테스트

@GetMapping("/{userId}/orders")
public ResponseEntity<List<ResponseOrder>> createOrder(@PathVariable("userId") String userId) throws Exception {
    log.info("Before retrieve orders data");
    Iterable<OrderEntity> orderList = orderService.getOrdersByUserId(userId);

    List<ResponseOrder> result = new ArrayList<>();
    orderList.forEach(v -> {
        result.add(new ModelMapper().map(v, ResponseOrder.class));
    });

    try {
        Thread.sleep(1000);
        throw new Exception("장애 발생");
    } catch (InterruptedException exception) {
        log.warn(exception.getMessage());
    }

    log.info("after retrieve orders data");

    return ResponseEntity.ok().body(result);
}