Grpc NameResolver + Eureka Service Discovery

Grpc 자바 클라이언트는 uri 를 ip:port 로 변환하기 위해 NameResolver 인터페이스를 사용합니다.

그 후, NameResolver 에게서 반환받은 ip:port 들을 이용하여 클라이언트 사이드 로드밸런싱을 수행합니다.

(Communtiy 에서는 RoundRobinLoadBalancer 를 사용하였습니다)

Community 에서는 msa 간 내부 호출을 위해 grpc 와 eurekea service discovery 를 이용하기 때문에 다음과 같은 기능이 필요하였습니다.

 

 

1. Eureka Service Discovery 에 등록된 서비스 이름을 이용해 grpc 연결을 맺을 수 있어야 합니다.

 

2. Eureka Discovery Client 는 일정 주기로 Eureka Server 로부터 등록된 애플리케이션들의 상태를 가져와 로컬에 캐싱해놓고 사용하는데, 이 캐시가 업데이트될 때 grpc 연결도 업데이트되어야 합니다.

즉, 유레카 서버에 grpc 연결을 맺어야할 서비스 인스턴스가 추가되거나, 제거되었을때 이를 반영해야 합니다.

 

 

이를 위해 먼저 유레카 클라이언트의 내부를 조금 들여다 보도록 하겠습니다.

service registry 에서 서비스 이름으로 등록된 인스턴스를 가져오기 위해서는 EurekaClient 인터페이스의 아래 메서드를 이용하면 됩니다.

 

 

 

 

EurekaClient 는 스프링의 EurekaClientAutoConfiguration 이 ApplicationContext 에 등록해 주기 때문에, 단순히 주입만 받으면 됩니다.

 

 

 

 

EurekaClient 의 구현체인 DiscoveryClient 를 보면 getInstanceByVipAddress 에서 위와 같은 메서드를 호출합니다. 예상처럼 메서드 호출시에 Eureka Server 를 찌르는 것이 아니라, 로컬 캐시에서 인스턴스 목록을 반환합니다.

 

따라서 (1) 의 요구사항은 EurekaClient.getInstanceByVipAddress(serviceName, false) 를 이용하여 인스턴스 목록을 가져온 뒤 로드밸런서에 등록해주면 됩니다.

 

이제 (2) 요구사항은 어떻게 해결해야할까요?

Eureka Discovery Client 에서 Eureka Server 에서 인스턴스 목록을 받아 캐시를 업데이트 한 뒤에

등록한 이벤트를 실행시켜 주는 등의 확장 포인트를 제공한다면 단순히 리스너를 만들어 끼워 넣어주면 됩니다.

DiscoveryClient 의 구현을 살펴보겠습니다.

 

 

 

 

refreshRegistry() 를 이용해 일정 주기로 레지스트리를 업데이트해주고 있습니다.

 

 

 

 

호출을 타고 들어가면 위와 같은 메서드가 호출되는 것을 볼 수 있습니다.

처음 요청에는 getAndStoreFullResgistry() 를 이용해 전체 레지스트리를 가져오고,

그 다음부터는 delta, 즉 변경된 사항만 getAndUpdateDelta() 를 호출해 반영한다고 합니다.

이렇게 레스트리를 업데이트 해준 이후 onCacheRefreshed() 를 호출해 주는 것을 볼 수 있습니다.

 

 

 

 

다행히 onCacheRefreshed 에서 CacheRefreshedEvent 를 호출해주고 있습니다.

 

따라서 (2) 요구사항을 위해서는,

EurekaDiscoveryClient 에 registerEventListener(EurekaEventListener) 를 이용해 이벤트 리스너를 등록하고, 리스너 내부에서 이벤트가 CacheRefreshedEvent 인지 확인한 후 grpc 로드밸런서에 업데이트해주면 됩니다.

 

 

 

 

NameResolver 인터페이스에서는 파라미터로 Listener 를 제공해주는데,

Listener::onAddresses() 를 이용해 로드 밸런서를 업데이트해줄 수 있습니다.

따라서 위와 같이 DiscoveryClientNameResolver 클래스를 구성하였습니다.

 

마지막으로 로드밸런서의 구현을 확인해보아야 하는데요. Listener::onAddresses 를 이용해 연결할 ip:port 를 새로고침해주는 경우, 변경된 연결만을 반영해야 하기 때문입니다. 즉, 변경 전후에 동일하게 UP 상태인 인스턴스의 커넥션을 닫고 다시 여는 경우 가 없어야 합니다.

 

 

 

 

Listner::onAddresses 호출을 따라가면 나오는 RoundRobinLoadBalancer 의 메서드입니다.

latestAddrs 는 업데이트할 주소들, currentAddrs 는 로드밸런서에 현재 등록된 주소들입니다.

setDifference(currentAddrs, lastestAddrs.keySet()) 을 이용해 currentAddrs 에는 있지만, lastestAddrs 에는 없는 주소들, 즉 연결을 닫아야할 주소들을 removeAddrs 에 저장하고, 메서드의 아래 부분에서 이 채널들을 닫아줍니다.

 

또, latestAddrs 를 순회하는 부분에서 if (existingSubchannel != null) 일 경우 채널을 새로 열지 않고 continue 해주는데, 바로 이미 해당 주소로 채널 연결이 있을 경우 새로 연결을 맺지 않기 위함입니다.

 

위 코드를 통해 RoundRobinLoadBalancer 는 걱정과 달리 변경된 주소만을 반영한다는 것을 확인할 수 있습니다.