MQTT Cluser의 구성을 고민하고 있다. MQTT Cluster를 구성하는 각 노드들은 다른 노드들과 서로 독립적이기 때문에, 엄격한 노드관리가 필요하지는 않다. 해서 굳이 주키퍼와 같은 소프트웨어의 도움없이, Zabbix를 이용해서 클러스터를 관리하는 걸로 설계를 했다. 자빅스로 클러스터를 구성하는 노드를 모니터링 하다가 이상이 생기면, 삭제하고 이 정보를 REDIS에 저장하는 방식이다.
이 방식으로 관리하는데 별 문제는 없는 것 같은데, Zabbix가 single point failuer상태 인게 거슬린다. HA 구성을 해야 하는데, 귀찮아서 주키퍼를 도입해서 구성해보기로 했다.(굳이 도입할 필요는 없는 프로젝트인데, 핑계삼아서 공부하는 셈 치고 있다.)
대략 아래와 같이 사용하려 한다.
주키퍼로 MQTT Cluster를 구성하는 노드들이 제대로 작동하는지를 확인한다.
노드들이 죽는다면, znode를 삭제하는 식으로 현재 살아있는 노드의 정보를 유지한다.
MQTT 메시지를 라우팅 하는 프로세스는 이 노드정보를 읽어서 "메시지를 보내야 할지를 결정한다."
적용 테스트
테스트 환경
Docker : 3개의 서버로 주키퍼 클러스터를 구성하고, 테스트 노드들도 몇개 붙이려고 하니 VirtualBox로는 너무 무거울 것 같다. 가벼운 Docker를 이용하기로 했다.
운영체제 : Ubuntu server 리눅스 14.10
Zookeeper 환경 구성
설치
주키퍼 Server 설치
# apt-get install zookeeper
Init 제어를 위해서 zookeeperd를 설치했다.
# apt-get install zookeeperd
# service zookeeper start
# service zookeeper status
* zookeeper is running
주키퍼 클러스터 설정
주키퍼의 가용성을 위해서 3개의 주키퍼 서버로 만들어진 클러스터를 설정하기로 했다. /etc/zookeeper/conf/zoo.conf 파일을 수정한다.
두개의 포트가 필요하다. 첫번째 포트(2888)은 클러스터 리더의 port고, 두번째 포트는 followers(추종자)의 port다.
클러스터링에 참여하는 주키퍼 서버는 1~255까지의 고유한 id를 가지고 있어야 한다. 172.17.0.4는 1, 0.6은 2, 0.7은 3으로 결정했다. id는 /var/lib/zookeeper/myid에 설정해줘야 한다.
# cat /var/lib/zookeeper/myid
1
어떤 녀석이 리더인지 원시적으로 확인하기로 했다. 2888 포트를 바인드하고 있는 녀석을 찾으면 된다.
MQTT Cluster는 클라이언트에게 메시지를 전송하는 일을 한다. 클라이언트는 MQTT Cluster를 구성하는 여러 노드들 중 하나에 연결해 있으므로, 클라이언트가 붙어있는 노드를 관리하는 메시지 경로 설정 프로그램(GMP)이 메시지를 적당한 노드로 보내는 일을 한다.
작동중인 노드의 목록을 GMP와 공유하는게 목적이다.
구성
MQTT 클러스터의 노드를 "/mqtt/node"로 등록하고, Watcher Client로 /mqtt 밑에 있는 (연결된)노드의 정보를 공유하기로 했다. Znode는 Ephemeral 타입으로 등록하면 된다.
개발 환경 설정
(Java 언어를 이용하지 않기 때문에)Python Kazoo를 이용하기로 했다. Kazoo 모듈을 설치했다. 나는 apt-get으로 설치했다.(pip로 설치할 수도 있다.)
# apt-get install python-kazoo
Znode 구성
"/mqtt/node"로 간단하게 구성했다. MQTT 클러스터를 구성하는 node들은 /mqtt 밑에 위치한다.
/mqtt/node-1
/node-2
/node-3
Watcher 클라이언트
Watcher 클라이언트는 /mqtt 밑에 있는 znode들을 관리한다. znode는 Ephemeral로 등록, 연결이 끊어지면 삭제(delete)된다. Watcher는 delete된 Node를 확인해서, REDIS 테이블을 조작한다.
/mqtt의 자식 노드들을 watch하는 코드를 만들었다.
from kazoo.client import KazooClient
import sys
import time
zk = KazooClient(hosts="172.17.0.2:2181,172.17.0.3:2181,172.17.0.4")
zk.start()
if zk.exists('/mqtt'):
print '/mqtt exists.'
else:
zk.create('/mqtt')
@zk.ChildrenWatch("/mqtt")
def watch_children(children, func=my_func):
print children
while True:
time.sleep(5)
zk.stop()
테스트
GMP들은 위의 watch 코드를 이용해서, 메시지를 보내려는 노드가 있는지를 확인한다. 노드가 존재하지 않는다면, 이 메시지는 유저의 메시지 함에 쌓는다.
테스트는 작동하는지를 검증하는 수준에서 간단하게 진행했다.
from kazoo.client import KazooClient
import sys
import time
node_name = sys.argv[1]
print "MQTT Node Start"+node_name
zk = KazooClient(hosts="172.17.0.2:2181,172.17.0.3:2181,172.17.0.4:2181")
zk.start()
stat = zk.create("/mqtt/"+node_name, b"example",ephemeral=True)
time.sleep(20)
print "Client disconnected"
zk.stop()
프로그램은 20초후에 종료하도록 했다. 주키퍼에 연결하고 끊어지는 정보를 watch를 통해 확인할 수 있다.
평가 및 개선
코드는 잘 작동 하지만 서비스에 사용하려면 개선해야 할 점들이 눈에 보인다.
노드의 delete는 주키퍼 서버와 클라이언트의 연결여부로 판단한다. 주키퍼 서버와 클라이언트는 연결상태이지만 실제 노드의 서비스 즉 MQTT 서비스 데몬은 제대로 작동하지 않을 수 있다.
데몬은 떠있으나 요청을 받지 못하는 경우
요청은 받지만 처리에 많은 시간이 걸리는 경우
요청을 받아서 처리하는 것 같지만 에러가 발생하는 경우 : 데이터 베이스 연결이 불완전하다던지, 파일 시스템, 메모리가 꽉 찼다던지.. 기타 다양한 이유가 있을 수 있다.
이런 비정상 작동하는 노드들은 클러스터에서 제거해야 한다.
모니터링 솔류션과의 연동을 통한 개선
애플리케이션을 모니터링 하는 것으로 개선을 할 수 있겠다. 이 모니터링 프로그램의 아래의 것들을 모니터링 한다.
MQTT PING을 이용한 작동 여부 확인
MQTT 메시지 처리량 확인 : 메시지 카운팅을 하고 이 정보를 모니터링 한다.
작동방식은 다음과 같다.
Zabbix를 이용해서 모니터링 한다.
Node에 문제가 생기면, 문제의 노드를 Zookeeper에서 삭제한다.
우선 ELB에서 노드를 제외해서, 노드로 요청이 가지 않도록 한다.
문제의 노드는 더이상 메시지를 처리하지 않지만 여전히 작동 중이기 때문에, 문제의 원인을 찾는 활동을 계속할 수 있다.
Contents
Zookeeper를 이용한 데이터 공유
사용 목적
적용 테스트
테스트 환경
Zookeeper 환경 구성
설치
주키퍼 클러스터 설정
Watcher 개발
목적
구성
개발 환경 설정
Znode 구성
/mqtt/node-1 /node-2 /node-3Watcher 클라이언트
from kazoo.client import KazooClient import sys import time zk = KazooClient(hosts="172.17.0.2:2181,172.17.0.3:2181,172.17.0.4") zk.start() if zk.exists('/mqtt'): print '/mqtt exists.' else: zk.create('/mqtt') @zk.ChildrenWatch("/mqtt") def watch_children(children, func=my_func): print children while True: time.sleep(5) zk.stop()테스트
from kazoo.client import KazooClient import sys import time node_name = sys.argv[1] print "MQTT Node Start"+node_name zk = KazooClient(hosts="172.17.0.2:2181,172.17.0.3:2181,172.17.0.4:2181") zk.start() stat = zk.create("/mqtt/"+node_name, b"example",ephemeral=True) time.sleep(20) print "Client disconnected" zk.stop()평가 및 개선
- 데몬은 떠있으나 요청을 받지 못하는 경우
- 요청은 받지만 처리에 많은 시간이 걸리는 경우
- 요청을 받아서 처리하는 것 같지만 에러가 발생하는 경우 : 데이터 베이스 연결이 불완전하다던지, 파일 시스템, 메모리가 꽉 찼다던지.. 기타 다양한 이유가 있을 수 있다.
이런 비정상 작동하는 노드들은 클러스터에서 제거해야 한다.모니터링 솔류션과의 연동을 통한 개선
- MQTT PING을 이용한 작동 여부 확인
- MQTT 메시지 처리량 확인 : 메시지 카운팅을 하고 이 정보를 모니터링 한다.
작동방식은 다음과 같다.Recent Posts
Archive Posts
Tags