오픈소스를 이용한 사내 데이터 분배 시스템 개발 (Feat. Thrift, Zookeeper)

IDINCU Engineering

오픈소스를 이용한 사내 데이터 분배 시스템 개발 (Feat. Thrift, Zookeeper)

2014/09/02 Irvin 1 Comment

1. 필요성

요새 그렇지 않은 서비스가 어디있겠냐마는, 저희 서비스 역시 하루에 수십~수백만 건의 로그가 군데 군데 쌓이고 있으며, 수십~수백만 건의 설문 응답 데이터 등이 중앙 RDBMS에 저장되고 있습니다. 그런데 최근 서비스 기반 아키텍쳐 (SOA)로 시스템을 점차 전환하면서 각각의 분산 서버에서는 기존의 중앙 RDBMS에 직접 접근하지 않게 되자, 제각기 발생하는 엄청나게 다양한 로그 및 데이터를 효율적으로 관리/분배할 수 있는 시스템이 필요하게 되었습니다. 대표적으로 Netflix 사에서 공개한 Suro라는 오픈소스 프로젝트가 있었지만, 데이터를 전송하는 transport layer부터 데이터를 저장하는 sink 단까지 저희 서비스에 딱 맞게 입맛대로 구성하고 싶었고 (non-java client 지원, metadata 등을 능동적으로 읽어 오기도 하는 시스템에 융합된 sink 개발, …) 새로이 개발하는 공수가 대단히 클 것 같지는 않았기에 직접 만들어 보게 되었습니다. 라고 쓰고 시켜서 만들었다고 읽습니다 /ㅁ/

2. 사용된 오픈소스

1) Thrift
가장 많은 트래픽을 감당해야 하는 시스템이니만큼 속도와 안정성 측면에서 충분한 검증을 받은 라이브러리를 transport layer로 선택하여야 했습니다. Protocol Buffer나 Avro에 대한 고려도 해 보았지만, 역시나 Facebook의 이름값에 넘어가 비교적 가장 많이들 사용하고 있는 Thrift를 사용하기로 결정하였습니다.

2) Zookeeper
데이터를 어떻게 분배할 것인가에 대한 rule이나, 현재 동작 중인 서버들의 상태 등을 관리하기 위해 사용하였습니다. 무난하게 RDBMS를 사용할 수도 있지만 매번 rule을 조회하기에는 부하가 너무 크고, cache를 사용하기에는 rule 변경 시 바로 적용되지도 않을뿐더러 서버마다 cache가 만료되는 시점이 달라 문제가 생길 수가 있습니다. 하지만 Zookeeper를 사용할 경우 rule의 변경 사항이 생길 때 곧바로 푸시를 해 줄 수 있고, 커넥션이 끊어질 경우 데이터가 휘발되게 하는 옵션이 있어 서버의 동작 여부를 알아 보기에도 매우 용이합니다.

3) Curator
원래도 아주 간단한 Zookeeper이긴 하지만, 그 Zookeeper를 보다 더 간편하게 사용할 수 있도록 해 주는 라이브러리입니다.

3. 프로젝트 구성

본 프로젝트는 thrift-lib, server, coordinator, client, 총 4개의 서브 프로젝트로 구성되어 있습니다. 각 서브 프로젝트의 역할과 함께 Thrift, Zookeeper의 활용 방법도 소개드리겠습니다.

1) thrift-lib
말 그대로 서비스 전반에서 thrift를 기반으로 통신하기 위해 데이터 및 호출 스펙을 정의한 서브 프로젝트입니다. 다른 모든 서브 프로젝트에서 본 서브 프로젝트를 import하여 사용합니다.

namespace java com.idincu.blog struct Blog { 1: string type, 2: string sender, 3: i64 timestamp, 4: optional string level, 5: optional string message, 6: optional string formattedMessage, 7: optional list argumentArray, 8: optional string loggerName, 9: optional string threadName, 10: optional list callerData, 11: optional map<string, string> extraData } service ServerService { void ping(), byte send(1: Blog blog) } service CoordinatorService { void ping(), list<string> getServerList() }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

namespace java com.idincu.blog   struct Blog {     1: string type,     2: string sender,     3: i64 timestamp,     4: optional string level,     5: optional string message,     6: optional string formattedMessage,     7: optional list argumentArray,     8: optional string loggerName,     9: optional string threadName,     10: optional list callerData,     11: optional map<string, string> extraData }   service ServerService {     void ping(),     byte send(1: Blog blog) }   service CoordinatorService {     void ping(),     list<string> getServerList() }

위와 같은 형식으로 [filename].thrift 파일을 생성한 후

thrift –gen java [filename].thrift

와 같이 명령을 내리면 gen-java 디렉터리 안에 java에서 사용할 수 있는 통신 라이브러리를 자동으로 생성해 줍니다. 해당 코드를 사용하여 아주 간편하게 통신을 할 수 있습니다. 물론 java 외에도 수많은 언어를 지원하기 때문에 나중에 다른 언어의 server나 client를 개발하기에도 아주 용이합니다.

2) server
server는 client로부터 다량의 데이터를 받아 큐에 쌓아 두고, 별개의 thread에서는 큐를 읽어 정해진 rule에 따라 데이터를 여러 sink에 분배합니다. 로컬 파일로 저장하는 sink, graylog로 전송하는 sink, email로 전송하는 sink, 자체적으로 관리하는 storage 서비스에 전송하는 sink 등 다양한 sink를 만들어 붙일 수 있겠습니다. 앞에서 언급했던 Zookeeper는 다음과 같이 활용하고 있습니다.

// connect to zookeeper RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); CuratorFramework client = CuratorFrameworkFactory.newClient("[connectString(list of zookeeper servers)]", retryPolicy); client.start(); // rule 모니터링 final NodeCache nc = new NodeCache(client, "[rule이 기록 된 node의 path]"); nc.start(); nc.getListenable().addListener(new NodeCacheListener() { @Override public void nodeChanged() throws Exception { // rule 갱신 로직 } });

1 2 3 4 5 6 7 8 9 10 11 12 13 14

// connect to zookeeper         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);         CuratorFramework client = CuratorFrameworkFactory.newClient("[connectString(list of zookeeper servers)]", retryPolicy);         client.start();           // rule 모니터링         final NodeCache nc = new NodeCache(client, "[rule이 기록 된 node의 path]");         nc.start();         nc.getListenable().addListener(new NodeCacheListener() {             @Override             public void nodeChanged() throws Exception {                 // rule 갱신 로직             }         });

위와 같이 특정 node를 watch하여 변경 사항이 생길 시 listener를 통해 즉시 반영할 수 있습니다. 또한 아래와 같이 Zookeeper에 노드를 생성할 수도 있으며, CreateMode.EPHEMERAL로 해당 커넥션이 종료될 경우 노드가 자동으로 삭제되게 할 수도 있습니다. 참고로 위의 node watch 기능이 거의 latency 없이 작동했던 반면, 커넥션이 종료된 후 node가 사라지는 데까지는 수십 초 가량의 시간이 걸려, 서버 재시작 시 기존 node가 살아 있어 익셉션이 나는 것을 방지하기 위해 retry 처리를 해 주어야 했습니다.

// zookeeper에 노드 생성 while (true) { try { client.getData().forPath("/status/" + hostname); } catch (KeeperException.NoNodeException e) { break; } System.out.println("Retrying registration..."); Thread.sleep(5000); } client.create().withMode(CreateMode.EPHEMERAL).forPath("/status/" + hostname);

1 2 3 4 5 6 7 8 9 10 11

// zookeeper에 노드 생성         while (true) {             try {                 client.getData().forPath("/status/" + hostname);             } catch (KeeperException.NoNodeException e) {                 break;             }             System.out.println("Retrying registration...");             Thread.sleep(5000);         }         client.create().withMode(CreateMode.EPHEMERAL).forPath("/status/" + hostname);

3) coordinator
coodinator는 위에서 server들이 Zookeeper에 생성에 놓은 node들을 토대로 client에게 현재 서버들의 상태를 알려 주는 아주 간단한 역할을 맡고 있습니다. HAProxy 같은 load balancer를 사용할 수도 있겠지만 이처럼 로그와 같은 대량의 데이터가 실시간으로 몰리는 시스템에서는 load balancer 그 자체가 single point of failure가 될 수 있기 때문에, 맨 처음 접속 시나 장애 시에만 coordinator로부터 서버들의 정보를 받아온 후로는 각 서버에 바로 접속하는 구조를 사용하게 되었습니다. 다음은 앞서 1)에서 설명했던 thrift-lib을 이용하여 서버를 구동하는 코드입니다. 보시는 바와 같이 아주 간단하게 적용할 수 있으며, Nonblocking 서버 외에도 ThreadPool 서버나 Simple 서버 등을 제공합니다.

public class CoordinatorServer { public static CoordinatorHandler handler; public static CoordinatorService.Processor processor; public static void main(String[] args) throws Exception { handler = new CoordinatorHandler(); processor = new CoordinatorService.Processor(handler); Runnable simple = new Runnable() { public void run() { simple(processor); } }; new Thread(simple).start(); } public static void simple(CoordinatorService.Processor processor) { try { TNonblockingServerTransport serverTransport = new TNonblockingServerSocket([port]); TServer server = new TNonblockingServer(new TNonblockingServer.Args(serverTransport).processor(processor)); System.out.println("Starting the coordinator server..."); server.serve(); } catch (Exception e) { e.printStackTrace(); } } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

public class CoordinatorServer {     public static CoordinatorHandler handler;     public static CoordinatorService.Processor processor;       public static void main(String[] args) throws Exception {         handler = new CoordinatorHandler();         processor = new CoordinatorService.Processor(handler);           Runnable simple = new Runnable() {             public void run() {                 simple(processor);             }         };           new Thread(simple).start();     }       public static void simple(CoordinatorService.Processor processor) {         try {             TNonblockingServerTransport serverTransport = new TNonblockingServerSocket([port]);             TServer server = new TNonblockingServer(new TNonblockingServer.Args(serverTransport).processor(processor));               System.out.println("Starting the coordinator server...");             server.serve();         } catch (Exception e) {             e.printStackTrace();         }     } }

 

public class CoordinatorHandler implements CoordinatorService.Iface { private CuratorFramework client; public CoordinatorHandler() throws Exception { RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); client = CuratorFrameworkFactory.newClient(prop.getProperty("[connectString(list of zookeeper servers)]"), retryPolicy); client.start(); } @Override public void ping() throws TException { } @Override public List<String> getServerList() throws TException { List<String> serverList = null; try { serverList = client.getChildren().forPath("/status"); } catch (Exception e) { e.printStackTrace(); } Collections.shuffle(serverList); System.out.println(serverList + " served."); return serverList; } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

public class CoordinatorHandler implements CoordinatorService.Iface {     private CuratorFramework client;       public CoordinatorHandler() throws Exception {         RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);         client = CuratorFrameworkFactory.newClient(prop.getProperty("[connectString(list of zookeeper servers)]"), retryPolicy);         client.start();     }       @Override     public void ping() throws TException {     }       @Override     public List<String> getServerList() throws TException {         List<String> serverList = null;         try {             serverList = client.getChildren().forPath("/status");         } catch (Exception e) {             e.printStackTrace();         }         Collections.shuffle(serverList);         System.out.println(serverList + " served.");         return serverList;     } }

4) client
client는 구동시 coorinator에 한 번 접속하여 server list를 받아온 후, 한 server에 접속하여 데이터를 전송할 수 있게 해 줍니다.

TTransport coordinatorTransport = new TFramedTransport(new TSocket(coordinatorUrl, [port])); coordinatorTransport.open(); TProtocol coordinatorProtocol = new TBinaryProtocol(coordinatorTransport); CoordinatorService.Client client = new CoordinatorService.Client(coordinatorProtocol); serverList = client.getServerList();

1 2 3 4 5

TTransport coordinatorTransport = new TFramedTransport(new TSocket(coordinatorUrl, [port]));             coordinatorTransport.open();             TProtocol coordinatorProtocol = new TBinaryProtocol(coordinatorTransport);             CoordinatorService.Client client = new CoordinatorService.Client(coordinatorProtocol);             serverList = client.getServerList();

client 측 코드 역시 Thrift를 사용하여 간단하게 작성할 수 있으며, logback을 통해 로그 또한 간편하게 전송할 수 있도록 새로운 appender도 만들어 주었습니다.

public class BlogAppender extends AppenderBase<ILoggingEvent> { private String coordinatorUrl; private String sender; public String getCoordinatorUrl() { return coordinatorUrl; } public void setCoordinatorUrl(String coordinatorUrl) { this.coordinatorUrl = coordinatorUrl; } public String getSender() { return sender; } public void setSender(String sender) { this.sender = sender; } @Override public void start() { super.start(); // coordinator, server 접속 등 초기화 } @Override protected void append(ILoggingEvent event) { // server로 로그 전송 } }

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31

public class BlogAppender extends AppenderBase<ILoggingEvent> {     private String coordinatorUrl;     private String sender;       public String getCoordinatorUrl() {         return coordinatorUrl;     }       public void setCoordinatorUrl(String coordinatorUrl) {         this.coordinatorUrl = coordinatorUrl;     }       public String getSender() {         return sender;     }       public void setSender(String sender) {         this.sender = sender;     }       @Override     public void start() {         super.start();         // coordinator, server 접속 등 초기화     }       @Override     protected void append(ILoggingEvent event) {         // server로 로그 전송     } }

이렇게 appender를 만들고 난 후 logback.xml에 추가하면 @Slf4j 어노테이션을 통해 간편하게 로그를 전송할 수 있게 됩니다.

<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE configuration> <configuration> <appender name="console" class="ch.qos.logback.core.ConsoleAppender"> <encoder> <charset>UTF-8</charset> <pattern>%date [%level] %msg - %logger [%file : %line] %X{RequestURL} %X{ParameterMap} %X{RequestedSessionId} %X{RemoteHost}%n</pattern> </encoder> </appender> <appender name="canal" class="com.idincu.blog.BlogAppender"> <coordinatorUrl>[coordiantor url]</coordinatorUrl> <sender>blog</sender> </appender> <logger name="com.idincu"> <level value="debug"/> </logger> <root> <level value="info"/> <appender-ref ref="console"/> <appender-ref ref="canal"/> </root> </configuration>

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

<?xml version="1.0" encoding="UTF-8"?> <!DOCTYPE configuration> <configuration> <appender name="console" class="ch.qos.logback.core.ConsoleAppender"> <encoder> <charset>UTF-8</charset> <pattern>%date [%level] %msg - %logger [%file : %line] %X{RequestURL} %X{ParameterMap} %X{RequestedSessionId} %X{RemoteHost}%n</pattern> </encoder> </appender> <appender name="canal" class="com.idincu.blog.BlogAppender"> <coordinatorUrl>[coordiantor url]</coordinatorUrl> <sender>blog</sender> </appender>   <logger name="com.idincu"> <level value="debug"/> </logger>   <root> <level value="info"/> <appender-ref ref="console"/> <appender-ref ref="canal"/> </root> </configuration>

 

4. 마치며

데이터 분배 시스템 구성도

결과적으로 이런 시스템이 완성되었습니다! 저희는 현재 본 시스템을 사용하여 데이터를 분배하고 있으며, 이 외에도 사내 클라우드 시스템 또한 오픈소스를 잘 사용하여 개발, 사용하고 있습니다.  오픈소스를 이용하는 중에 문제가 보이면 직접 contribute하는 분들도 계시고요. 이미 다들 그러시겠지만, 혹 아직 그렇지 않은 분들도 계시다면 편리하고 보람찬 오픈소스 생활을 누려 봅시다 : D