Consistent Hash 기반의 대량 메시지 처리 인프라

Contents

IoT 플랫폼에서 사용 할 수 있는 대용량 메시지 처리 시스템 개발에 대한 아이디어를 정리한다. 크게 두 가지 아이디어가 있을 건데, 그 중 consistent hash를 이용한 방법을 살펴본다. 이 메시지 처리 인프라는 아래의 특징들을 가지고 있다.

1. 유저간 채팅 서비스가 목적이 아니다. 유저간 채팅 서비에도 사용 할 수 있긴 하지만, IoT를 구성하는 디바이스 들간의 메시지 교환을 목적으로 한다.
2. 대량의 디바이스들이 연결 할 수 있다. 억 단위의 연결을 감당 할 수 있어야 한다.
3. 유저 채팅 서비스보다 대량의 메시지를 감당 할 수 있어야 한다. 메시지 크기는 상대적으로 작을 것이다.

Consistent Hash는 노드가 변경될 때 K/N만큼만 재분되 되는 해시(hash) 알고리즘이다. K는 Key의 갯수이고 n은 노드(혹은 slots)의 갯수다. 클러스터에 포함된 노드(node)의 갯수를 알고 있다면, 자원이 배치된 노드를 한번의 해시 연산으로 찾을 수 있다. Object Storage나 Cache 시스템, 라우터(router)등 대량의 데이터를 다루는 시스템에서 널리 사용하고 있다.

Object Storage인 Swift는 Consistent Hash를 응용해서 파일들을 배치한다. 파일 대신 메시징 채널을 배치하면 어떨까 ? 아래 그림을 보자.

네모상자들은 메시지 채널이다. 노란색 유저의 key는 56, 파란색 유저의 키는 12다. 노란색 유저(56)에게 메시지를 보내면, 해시를 돌려서 노란색 유저의 채널이 있는 노드를 찾아서 메시지를 전송한다. 이 방식의 장점은 아래와 같다.

1. 라우팅 테이블(routing table)을 유지할 필요가 없다.
2. 노드를 추가하는 것으로 확장이 쉽다. 노드가 추가되면 자동으로 채널이 (K/N 만큼) 재 분배된다. 재 분배를 위해서 해시 함수를 다시 실행해야 하는 문제가 있지만, 이 문제는 충분히 해결 할 수 있다.

Consistent hash를 이용해서 메시징 시스템 아키텍처를 만들어 보자. Swift의 hash ring 모델을 사용한다고 가정한다.

Swift는 karger consistent hasing의 응용으로 미래 가상노드(virtual node)를 만들어서 노드의 확장과 실패에 대응한다. 각 key에 대한 virtual node 정보는 데이터베이스에 저장되 있으며, Proxy 서버는 이 데이터베이스를 이용해서 object의 위치를 찾는다. 이 방식의 단점은 아래와 같다.

1. 반드시 Proxy 서버가 있어야 한다.
2. Hash table을 유지하기 위해서 많은 메모리가 필요하다.
3. Hash table을 중앙에서 유지해야 한다. 분산 할 수가 없다.
4. 모든 요청에 대해서, node를 찾기 위한 연산을 해야 한다. 이 연산 역시 proxy에 집중 된다.

특히 1번 단점은 문제가 될 수 있다. Object Storage에서라면, 대부분의 작업이 file node에 집중 될 것이다. 하지만 메시징의 경우 채널에서 하는 일은 메시지의 전달로 상대적으로 단순하기 때문에 노드가 늘어나는 만큼 proxy도 늘어나게 된다. 간단하지만 효율적인 구성이 아니다.

나는 jump consistent hash를 메시징 인프라 아키텍처 개발에 응용하기로 했다. 이 방식의 가장 큰 장점은 해시 테이블을 분산 할 수 있다는 것이다. 디바이스가 자신이 연결할 메시지 플랫폼의 노드갯수(NodeNum)를 알고 있다면 Hash(key, NodeNum)를 이용해서 해시테이블을 만들 수 있다. NodeNum만 알고 있다면, 디바이스가 직접 해시테이블을 만들 수 있으므로 메모리와 CPU 연산을 각 디바이스(클라이언트)에 분산 할 수 있다. 아래 그림을 보자.

key가 1과 5인 모바일 디바이스 두 개가 있다. 이들 디바이스가 메시지 플랫폼에 연결을 할 경우 Hash(1, NodeNum), Hash(5, NodeNum) 연산을 통해서 자신이 연결해야 할 노드를 계산할 수 있다. 이렇게 연결된 모바일 디바이스에 메시지를 보낸다고 가정해 보자. 메시지를 주고 받는 디바이스의 목록은 데이터베이스에 저장되어 있을 테니, 동일한 Hash 연산을 이용해서 메시지를 수신할 디바이스가 연결된 노드 위치를 계산 할 수 있다. 이렇게 Hash Table을 만들어 두면, NodeNum이 변할 때까지는 계속 사용 할 수 있다. 만약 NodeNum이 변한다면, Hash Table에 대한 리빌드(rebuild)하면 된다. 리빌드 작업 역시 클라이언트에 분산 된다.

시스템 인프라는 대략 아래와 같이 구성 할 수 있을 것이다.

디바이스들은 Hash함수를 이용해서 자신이 연결할 디바이스의 노드상에서의 위치를 알 수 있다. 노드는 도메인 이름(Domain Name)으로 그리고 디바이스의 위치는 URI로 표현 할 수 있을 것이다. 이제 REST를 이용해서 자원(여기에서는 메시지 채널)을 찾고 여기에 메시지를 전송 할 수 있다.

메시지는 서비스로도 보낼 수 있어야 한다. 메시지 수집, 분석, NLP, ML과 같은 백앤드 서비스일 수도 있으며, IFTTT와 같은 외부의 다른 인터넷 서비스일 수도 있다. 메시지는 수집서버와 실시간 분석 서비스등 동시에 두 개이상의 서비스로 보낼 수도 있어야 한다. 메시지에 대한 브로드캐스팅, 멀티캐스팅 등이 가능해야 한다는 의미다. 아래 그림을 보자.

간단하게 서비스별 메시지 채널을 만들면 된다. 하지만 서비스가 추가될 때마다, 디바이스에 key를 알려줘야 하기 때문에 (가능은 하지만)좋은 방법은 아니다. 이런 문제는 브로드 캐스팅이나 멀티캐스팅등의 복잡한 기능을 추가하는 것 보다는 서비스 설계를 달리 하는 것으로 해결 해야 할 것이다. 아래 그림을 보자.

BigData Service를 만든 다음 하나의 인터페이스(그림의 message collect)를 통해서 메시지를 받고 내부에서 분배하는 식의 설계가 올바른 설계일 것이다. 이런 백앤드 서비스에는 수많은 디바이스가 연결될 수 있다. 따라서 채널의 분산이 필요 할 수 있는데, 이는 아래 장에서 자세히 살펴보겠다.

메시지 플랫폼에 연결되는 서비스는 대량의 메시지를 처리해야 할 수 있다. 따라서 메시지 채널에 대한 Scale-In/Out을 보장해야 한다.

서비스를 위해서 두 개 이상의 key를 할당하면 된다. 물론 Hash(key, NodeNum) 결과가 노드에 분산되도록 key를 만들어야 할 것이다. 이렇게 구성하면 디바이스가 두 개 이상의 key를 가져야 하는 문제가 발생한다. 그리고 이 key는 늘어나거나 줄어들 수 있기 때문에(scale-in/scale-out) 이를 관리하기 위한 방법을 고안해야 한다. 이 문제는 Message channel discovery 에서 자세히 살펴보겠다.

디바이스나 서비스는 가용성과 확장성을 위해서 두 개 이상의 분산된 채널을 가질 수 있다. 이렇게 채널이 늘어나면 key도 따라서 늘어나는데, 이걸 디바이스에게 알려주는 건 좋은 방법이 아니다. DNS를 이용해서 구성하기로 했다.

BigData Service는 2개의 메시지 채널을 가지고 있다. 이 채널을 DNS로 묶는다. 도메인 이름은 bigdata.example.com으로 하고 DNS SRV 레코드에 기록 했다. SRV의 레코드에 있는 priority와 weight를 이용해서 백업 서비스 및 요청 분산을 조절 할 수 있다. 위 그림에서 node2-ip.message.priv의 weight는 40, node3-ip.message.priv의 weight는 60으로 설정했다. 이를 이용해서 대략 4:6으로 요청을 분산 할 수 있다.

노드에 문제가 생기면, 채널은 다른 노드로 자동으로 이동해야 한다. 디바이스는 간단한 연산으로 노드의 위치를 찾을 수 있어야 한다.

Node-02에 문제가 생기면, 여기에 있는 채널들은 남아있는 다른 노드들로 골고루 분산해야 한다. 전제 조건은 디바이스가 NodeNum을 동기화 해야 한다는 점이다. 동기화가 끝났다고 가정하면 문제의 노드를 향하고 있는 디바이스들에 대해서 Hash(k, NodeNum) 연산을 수행하면 된다. 간단한 방법이지만, 노드들이 빈번하게 변화할 경우 디바이스들의 연산량이 늘어날 수 있다는 문제점이 생긴다. 큰 문제는 아닐 것 같지만 그래도 효율적인 방법이 있는지 고민해 볼만하다.

Object Storage swift 처럼 Replica 테이블을 구성하는 방법이 있다. Swift의 경우에는 Proxy에서 모든 테이블을 관리하기 때문에, 테이블이 중복되지만 않도록 알고리즘을 만들면 된다. 하지만 Jump consistent hash에서는 테이블이 클라이언트에게 분산되므로, 분산된 클라이언트가 동일한 결과를 가지도록 알고리즘을 만들어야 한다.

아래와 같은 테이블을 만들어야 할 것이다.

Key	Real Node	Replica-01	Replica-02
50	3	2	5
87	2	4	6
90	2	4	6
95	1	3	5

아래와 같은 방법을 생각했다.

Node03이 정상적인 상태에서 연결되는 노드다. 2개의 백업본을 가진다면, 정상 상태에서의 연결 노드(그림에서 Node03)를 제외하고 2개의 가상 클러스터 그룹으로 나눈다. 그 후 아래의 과정을 거쳐서 백업 노드를 설정한다.

1. mod 연산을 이용해서 클러스터를 랜덤하게 선택한다.
2. 각각의 가상 클러스터에서 hash(key, node_num) 연산을 한다.

필요한 노드의 갯수는 "(백업 클러스터 크기 * 2) + 1"이 될 것이다. 예를 들어 백업 클러스터의 크기가 2라면 3개, 3개라면 7개의 node가 필요하다. 대부분의 경우 백업 클러스터는 2개면 충분할 것이다.

백업 구성은 하나의 메시지 채널만을 가지는 "디바이스"에 대해서만 이루어지면 된다. 다수의 디바이스를 처리하는 서비스의 경우 대역폭을 위해서 두 개 이상의 메시지 채널을 유지해야 겠지만 개별 디바이스들은 그럴 필요가 없기 때문이다. 시스템 구성은 아래와 같을 것이다.

서비스와 노드, 디바이스와 노드들 사이에는 프락시가 위치한다. 프락시 서버는 노드들의 상태 정보를 가지고 있다. 특정 노드에 문제가 생기면, 프락시 서버는 서비스와 디바이스에 이 정보를 알려줘서, 백업 노드에 연결 할 수 있도록 한다. 프락시 서버는 노드의 상태 정보만 알려주고, 어떤 노드에 연결해야 할지는 서비스와 디바이스가 판단하도록 구성하는게 효율적일 것이다. 디바이스들 간에 그룹을 만들 수 있어야 한다. 사람이 사용 할 수도 있는 범용플랫폼이고 이 경우 채팅그룹과 같은 기능이 중요 할 수 있기 때문이다. 다른 디바이스들에도 그룹기능이 필요 할 수 있을 것이다. 기본적인 구성은 아래와 같다.

그룹 메시지 교환 타입의 메시지 채널을 만든다. 이 채널로 메시지를 보내면 브로드캐스팅 된다. 메시지 저장을 위한 메시지큐도가 있어야 한다. 메시지큐는 메시지 인프라의 중요 컴포넌트들 중 하나가 될 것이다. 메시지큐를 이용한 메시지박스의 구현은 따로 다룬다.

Hash(key, NodeNum)이 제대로 작동하려면 클라이언트들과 메시지 인프라의 NodeNum이 동기화 돼야 한다. 이를 위해서 클러스터를 구성하고 클러스터에 포함된 노드들의 상태를 관리하는 분산 자원관리 시스템을 만들어야 한다.

분산 커널 모델로 MesOS의 아키텍처를 참고했다. MasterOS가 커널 역할을 하며, 노드들의 상태를 관리한다. 노드들은 Active, Deactive, Fail 등 몇가지 상태를 가진다. 예를 들어 Deactive나 Fail 상태가 된다면, 이들을 클러스터에서 제외하고 NodeNum을 수정을 한다.

분산자원 관리 시스템 구성은 간단한 내용이 아니다. 따로 항목을 만들어서 자세히 기술 할 생각이다. 여기에서는 개요 정도만 소개한다.

분산자원 관리 시스템 구성을 위한 몇 가지 아키텍처가 있다. 이중 요즘 관심을 가지고 있는게 분산 커널(Distributed operation) 이다. 분산된 자원을 관리하는 운영체제라고 이해하면 된다. 운영체제라고 하지만 리눅스 커널과 같은 커널을 만드는 것과는 상관이 없다. 커널의 구성요소만 빌려서 좀 더 높은 차원에서 구현할 뿐이다. 즉

1. 자원 관리 : 일반 커널은 단일 컴퓨터의 자원을 관리한다. 하지만 분산 커널은 분산된 자원(보통 클러스터링된)을 관리한다. 메시지 처리 인프라를 위한 분산 커널은 서비스나 디바이스가 추가 됐을 때, 어떤 노드를 할당할지를 결정해야 하는데 이를 위해서 CPU, 네트워크, 메모리 정보를 주로 수집할 것이다.
2. 스케쥴링 : 일반 커널은 유저가 프로세스의 실행을 요청하면 적절한 크기의 메모리와 CPU를 할당한다. 분산 커널은 1의 자원 모니터링 정보를 토대로 적당한 노드(자원이 충분한 노드)를 선택해서 프로세스를 실행한다. 메시지 처리 인프라는 약간 다를 것이다. 디바이스 요청마다 채널 프로세스를 만들지 않고 하나의 프로세스가 모든 채널을 관리하기 때문이다. 그냥 자원이 충분한 노드를 선택해서 key를 할당하면 된다.
3. Service Discovery : 일반 커널에서 프로세스는 PID로 찾는다. 분산 커널의 경우 네트워크로 분리되기 때문에 PID로 프로세스를 찾을 수 없다. 기술적으로 보자면 노드이름과 PID를 조합하면 유일한 식별자를 만들 수 있을 거다. 하지만 보통 네트워크에서 표준으로 사용하고 있는 DNS를 이용해서 프로세스를 식별한다. 메시지 처리 인프라의 경우 굳이 프로세스를 식별할 필요는 없으므로 Service Discovery 시스템은 필요 없다.

일반적인 분산 운영체제보다는 단순한 형태로 만들 수 있을 것이다.

디바이스들은 NodeNum만 알고 있다면, 메시지 채널을 찾을 수 있다. 따라서 노드의 실패, 노드의 추가, 삭제에 의한 NodeNum의 정보를 디바이스들과 동기화하기 위한 매커니즘이 필요하다. 이 매커니즘은 분산 커널이 제공한다. 아래와 같이 구성할 수 있을 것이다. 디바이스에 Key를 할당하는 과정은 따로 다룬다. Key를 할당 받았다고 가정하고 기술한다.

Distributed OS는 Node들의 자원을 모니터링 하고, 상태를 관리한다. 이중 자원의 상태정보와 노드 갯수(Node_Num)은 Proxy와 공유한다. 디바이스가 다른 서비스로 메시지를 보낸다. 이 메시지는 프락시(proxy) 서버가 받아서 처리한다. 메시지에는 수신자의 메시지 채널 정보와 메시지 채널의 위치가 모두 포함되어 있기 때문에 단지 프락시만 하면 된다.

만약 목적지 노드가 메시지를 받을 수 없다면, 디바이스에 "백업 노드를 이용하라는" 에러메시지를 보낸다. 목적지와 백업 목적지를 찾기 위한 모든 연산을 디바이스에 맡기는 방식이다.

프락시에 맡기면 어떨까 ? 모든 key에 대한 백업 테이블을 미리 만드는 방법이 있다. 많은 메모리 공간이 필요하며, 특히 노드의 갯수가 바뀔 때 전체 테이블을 다시 만들어야 하는 문제점이 있다. 문제가 생긴 노드로 향하는 메시지에 대해서만 연산을 하는 방법도 있다. 노드에 문제가 생겼을 경우에만 연산을 하면 되며, 연산양도 그리 크지 않으므로 쓸만한 방법이다.