Merge branch 'master' of sunsh318/pinpoint

from pull-request 7 * refs/heads/master: #5 Pinpoint 클러스터 개발 #5 Pinpoint 클러스터 개발 #5 Pinpoint 클러스터 개발 #5 Pinpoint 클러스터 개발
2026-08-19 01:35:58 +10:00 · 2014-09-04 18:52:23 +09:00
parent b1f2a8ab84 4cfc851629
commit 959b65cd0f
19 changed files with 1218 additions and 75 deletions
@@ -0,0 +1,158 @@
+package com.nhn.pinpoint.common.util;
+
+import java.net.Inet4Address;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.NetworkInterface;
+import java.net.SocketException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Enumeration;
+import java.util.List;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public final class NetUtils {
+
+	public static final String LOOPBACK_ADDRESS_V4 = "127.0.0.1";
+	
+	private NetUtils() {
+	}
+
+	public static List<InetSocketAddress> toInetSocketAddressLIst(List<String> addressList) {
+		List<InetSocketAddress> inetSocketAddressList = new ArrayList<InetSocketAddress>();
+
+		for (String address : addressList) {
+			InetSocketAddress inetSocketAddress = toInetSocketAddress(address);
+			if (inetSocketAddress != null) {
+				inetSocketAddressList.add(inetSocketAddress);
+			}
+		}
+
+		return inetSocketAddressList;
+	}
+
+	public static InetSocketAddress toInetSocketAddress(String address) {
+		try {
+			URI uri = new URI("pinpoint://" + address);
+
+			return new InetSocketAddress(uri.getHost(), uri.getPort());
+		} catch (URISyntaxException e) {
+		}
+
+		return null;
+	}
+
+	public static String getLocalV4Ip() {
+		try {
+			InetAddress localHost = InetAddress.getLocalHost();
+			String localIp = localHost.getHostAddress();
+			if (validationIpV4FormatAddress(localIp)) {
+				return localIp;
+			}
+		} catch (UnknownHostException e) {
+		}
+		return LOOPBACK_ADDRESS_V4;
+	}
+	
+	/**
+	 * 가지고 있는 외부에서 접근할수 있는 ip를 모두 반환합니다. 
+	 * 만약 로컬 ip가 획득하지 못할 경우 Empty List를 반환합니다. 
+	 */
+	public static List<String> getLocalV4IpList() {
+		List<String> result = new ArrayList<String>();
+		
+		Enumeration<NetworkInterface> interfaces = null;
+		try {
+			interfaces = NetworkInterface.getNetworkInterfaces();
+		} catch (SocketException e) {
+		}
+
+		if (interfaces == null) {
+			return Collections.EMPTY_LIST;
+		}
+		
+		while (interfaces.hasMoreElements()) {
+			NetworkInterface current = interfaces.nextElement();
+			if (isSkipIp(current)) {
+				continue;
+			}
+
+			Enumeration<InetAddress> addresses = current.getInetAddresses();
+			while (addresses.hasMoreElements()) {
+				InetAddress address = addresses.nextElement();
+				if (address.isLoopbackAddress() || !(address instanceof Inet4Address)) {
+					continue;
+				}
+				
+				if (validationIpV4FormatAddress(address.getHostAddress())) {
+					result.add(address.getHostAddress());
+				}
+			}
+		}
+
+		return result;
+	}
+	
+	private static boolean isSkipIp(NetworkInterface networkInterface) {
+		try {
+			if (!networkInterface.isUp() || networkInterface.isLoopback() || networkInterface.isVirtual()) {
+				return true;
+			}
+			return false;
+		} catch (Exception e) {
+		}
+		return true;
+	}
+	
+
+	public static boolean validationIpPortV4FormatAddress(String address) {
+		try {
+			int splitIndex = address.indexOf(':');
+
+			if (splitIndex == -1 || splitIndex + 1 >= address.length()) {
+				return false;
+			}
+
+			String ip = address.substring(0, splitIndex);
+			
+			if (!validationIpV4FormatAddress(ip)) {
+				return false;
+			}
+
+			String port = address.substring(splitIndex + 1, address.length());
+			if (Integer.parseInt(port) > 65535) {
+				return false;
+			}
+
+			return true;
+		} catch (Exception e) {
+		}
+
+		return false;
+	}
+	
+	public static boolean validationIpV4FormatAddress(String address) {
+		try {
+			String[] eachDotAddress = address.split("\\.");
+			if (eachDotAddress.length != 4) {
+				return false;
+			}
+
+			for (String eachAddress : eachDotAddress) {
+				if (Integer.parseInt(eachAddress) > 255) {
+					return false;
+				}
+			}
+			return true;
+		} catch (NumberFormatException e) {
+		}
+
+		return false;
+	}
+
+}
@@ -612,6 +612,14 @@
            <groupId>org.aspectj</groupId>
            <artifactId>aspectjweaver</artifactId>
        </dependency>
+		<!-- for zookeeper test -->
+        <dependency>
+            <groupId>org.apache.curator</groupId>
+            <artifactId>curator-test</artifactId>
+            <version>2.6.0</version>
+            <scope>test</scope>
+        </dependency>
+        
    </dependencies>

    <build>
@@ -0,0 +1,12 @@
+package com.nhn.pinpoint.web.cluster;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public interface ClusterManager {
+
+	boolean registerWebCluster(String zNodeName, byte[] contents);
+
+	void close();
+
+}
@@ -0,0 +1,165 @@
+package com.nhn.pinpoint.web.cluster.zookeeper;
+
+
+import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.AuthException;
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.BadOperationException;
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.ConnectionException;
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.PinpointZookeeperException;
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.TimeoutException;
+import com.nhn.pinpoint.web.cluster.zookeeper.exception.UnknownException;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class ZookeeperClient {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	// 쥬키퍼 클라이언트는 스레드 세이프함
+	private final ZookeeperClusterManager manager;
+
+	private final ZooKeeper zookeeper;
+	private final AtomicBoolean clientState = new AtomicBoolean(true);
+
+	// 데이터를 이친구가 다가지고 있어야 할 거 같은데;
+	public ZookeeperClient(String hostPort, int sessionTimeout, ZookeeperClusterManager manager) throws KeeperException, IOException, InterruptedException {
+		this.manager = manager;
+		zookeeper = new ZooKeeper(hostPort, sessionTimeout, this.manager); // server
+	}
+	
+	/**
+	 * path의 가장마지막에 있는 node는 생성하지 않는다. 
+	 * 
+	 * @throws PinpointZookeeperException 
+	 * @throws InterruptedException 
+	 */
+	public void createPath(String path) throws PinpointZookeeperException, InterruptedException {
+		checkState();
+
+		int pos = 1;
+		do {
+			pos = path.indexOf('/', pos + 1);
+
+			if (pos == -1) {
+				pos = path.length();
+				return;
+			}
+
+			try {
+				String subPath = path.substring(0, pos);
+				if (zookeeper.exists(subPath, false) != null) {
+					continue;
+				}
+
+				zookeeper.create(subPath, new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
+			} catch (KeeperException exception) {
+				if (exception.code() != Code.NODEEXISTS) {
+					handleException(exception);
+				} 
+			}
+
+		} while (pos < path.length());
+	}
+
+	// 정확히 동일한 노드가 생성되어 있는지 확인하려면
+	// 내부의 컨텐츠 검사도 해야됨
+	public String createNode(String znodePath, byte[] data, CreateMode createMode) throws PinpointZookeeperException, InterruptedException {
+		checkState();
+
+		try {
+			if (zookeeper.exists(znodePath, false) != null) {
+				return znodePath;
+			}
+
+			String pathName = zookeeper.create(znodePath, data, Ids.OPEN_ACL_UNSAFE, createMode);
+			return pathName;
+		} catch (KeeperException exception) {
+			if (exception.code() != Code.NODEEXISTS) {
+				handleException(exception);
+			} 
+		}
+		return znodePath;
+	}
+	
+	public void delete(String path) throws PinpointZookeeperException, InterruptedException {
+		checkState();
+
+		try {
+			zookeeper.delete(path, -1);
+		} catch (KeeperException exception) {
+			if (exception.code() != Code.NONODE) {
+				handleException(exception);
+			} 
+		}
+	}
+
+	public boolean exists(String path) throws PinpointZookeeperException, InterruptedException {
+		checkState();
+
+		try {
+			Stat stat = zookeeper.exists(path, false);
+			if (stat == null) {
+				return false;
+			}
+		} catch (KeeperException exception) {
+			if (exception.code() != Code.NODEEXISTS) {
+				handleException(exception);
+			} 
+		}
+		return true;
+	}
+
+	private void checkState() throws PinpointZookeeperException {
+		if (!this.manager.isConnected() || !clientState.get()) {
+			throw new ConnectionException("instance must be connected.");
+		}
+	}
+
+	private void handleException(KeeperException keeperException) throws PinpointZookeeperException {
+		switch (keeperException.code()) {
+			case CONNECTIONLOSS:
+			case SESSIONEXPIRED:
+				throw new ConnectionException(keeperException.getMessage(), keeperException);
+			case AUTHFAILED:
+			case INVALIDACL:
+			case NOAUTH:
+				throw new AuthException(keeperException.getMessage(), keeperException);
+			case BADARGUMENTS:
+			case BADVERSION:
+			case NOCHILDRENFOREPHEMERALS:
+			case NOTEMPTY:
+			case NODEEXISTS:
+			case NONODE:
+				throw new BadOperationException(keeperException.getMessage(), keeperException);
+			case OPERATIONTIMEOUT:
+				throw new TimeoutException(keeperException.getMessage(), keeperException);
+			default:
+				throw new UnknownException(keeperException.getMessage(), keeperException);
+			}
+	}
+	
+	public void close() {
+		if (clientState.compareAndSet(true, false)) {
+			if (zookeeper != null) {
+				try {
+					zookeeper.close();
+				} catch (InterruptedException ignore) {
+					logger.debug(ignore.getMessage(), ignore);
+				}
+			}
+		}
+	}
+
+}
@@ -0,0 +1,198 @@
+package com.nhn.pinpoint.web.cluster.zookeeper;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.jboss.netty.util.HashedWheelTimer;
+import org.jboss.netty.util.Timeout;
+import org.jboss.netty.util.Timer;
+import org.jboss.netty.util.TimerTask;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.nhn.pinpoint.rpc.util.TimerFactory;
+import com.nhn.pinpoint.web.cluster.ClusterManager;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class ZookeeperClusterManager implements ClusterManager, Watcher {
+
+	private static final String PINPOINT_CLUSTER_PATH = "/pinpoint-cluster";
+	private static final String PINPOINT_WEB_CLUSTER_PATh = PINPOINT_CLUSTER_PATH + "/web";
+
+	private static final String PATH_SEPERATOR = "/";
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	private final AtomicBoolean connected = new AtomicBoolean(false);
+	private final ZookeeperClient client;
+
+	private final int retryInterval;
+	
+	private final Timer timer;
+
+	private final AtomicReference<RegisterJob> job = new AtomicReference<ZookeeperClusterManager.RegisterJob>();
+
+	public ZookeeperClusterManager(String zookeeperAddress, int sessionTimeout, int retryInterval) throws KeeperException, IOException, InterruptedException {
+		this.client = new ZookeeperClient(zookeeperAddress, sessionTimeout, this);
+		this.retryInterval = retryInterval;
+		// 등록이 실패하였을때 생성하게 하는게 나을수도 있음
+		this.timer = createTimer();
+	}
+
+	// 등록이 실패해도 계속 시도 (주기는 기본 1분)
+	// 크게 부하가 가는 작업이 아니며, 
+	// 실패할 경우 계속 로그를 출력
+	@Override
+	public boolean registerWebCluster(String zNodeName, byte[] contents) {
+		String zNodePath = bindingPathAndZnode(PINPOINT_WEB_CLUSTER_PATh, zNodeName);
+
+		logger.info("Create Web Cluster Zookeeper UniqPath = {}", zNodePath);
+
+		RegisterJob job = new RegisterJob(zNodePath, contents, retryInterval);
+		if (!this.job.compareAndSet(null, job)) {
+			logger.warn("Already Register Web Cluster Node.");
+			return false;
+		}
+
+		// 스케쥴로 라도 등록하면 성공
+		registerWebCluster(job);
+		return true;
+	}
+
+	@Override
+	public void process(WatchedEvent event) {
+		KeeperState state = event.getState();
+		EventType eventType = event.getType();
+
+		// 상태가 되면 ephemeral 노드가 사라짐
+		// 문서에 따라 자동으로 연결이 되고, 연결되는 이벤트는 process에서 감지가 됨
+		if (state == KeeperState.Disconnected || state == KeeperState.Expired) {
+			connected.compareAndSet(true, false);
+			return;
+		}
+
+		if ((state == KeeperState.SyncConnected || state == KeeperState.NoSyncConnected) && eventType == EventType.None) {
+			// 이전상태가 RUN일수 있기 때문에 유지해도 됨
+			boolean changed = connected.compareAndSet(false, true);
+			if (changed) {
+				RegisterJob job = this.job.get();
+				if (job != null) {
+					registerWebCluster(job);
+				}
+			}
+			return;
+		}
+	}
+
+	@Override
+	public void close() {
+		if (timer != null) {
+			timer.stop();
+		}
+
+		if (client != null) {
+			this.client.close();
+		}
+	}
+
+	private Timer createTimer() {
+		HashedWheelTimer timer = TimerFactory.createHashedWheelTimer("Pinpoint-Web-Cluster-Timer", 100, TimeUnit.MILLISECONDS, 512);
+		timer.start();
+		return timer;
+	}
+
+	private boolean registerWebCluster(RegisterJob job) {
+		String zNodePath = job.getZnodePath();
+		byte[] contents = job.getContents();
+
+		if (!isConnected()) {
+			logger.info("Web Cluster Zookeeper is Disconnected. This job retry when reconnected. Path={}", zNodePath);
+			return false;
+		}
+
+		try {
+			if (!client.exists(zNodePath)) {
+				client.createPath(zNodePath);
+			}
+
+			// 쥬키퍼의 zNode는 ip:port 형태의 이름으로 만들수 있음
+			String nodeName = client.createNode(zNodePath, contents, CreateMode.EPHEMERAL);
+			logger.info("Register Web Cluster Zookeeper UniqPath = {}.", zNodePath);
+			return true;
+		} catch (Exception e) {
+			logger.warn(e.getMessage(), e);
+		}
+		reservationRegisterWebCluster(job);
+		return false;
+	}
+
+	private void reservationRegisterWebCluster(RegisterJob job) {
+		timer.newTimeout(job, job.getRetryInterval(), TimeUnit.MILLISECONDS);
+	}
+
+	public boolean isConnected() {
+		return connected.get();
+	}
+
+	private String bindingPathAndZnode(String path, String znodeName) {
+		StringBuilder fullPath = new StringBuilder();
+
+		fullPath.append(path);
+		if (!path.endsWith(PATH_SEPERATOR)) {
+			fullPath.append(PATH_SEPERATOR);
+		}
+		fullPath.append(znodeName);
+
+		return fullPath.toString();
+	}
+
+	class RegisterJob implements TimerTask {
+		private final String znodeName;
+		private final byte[] contents;
+		private final int retryInterval;
+		
+		public RegisterJob(String znodeName, byte[] contents, int retryInterval) {
+			this.znodeName = znodeName;
+			this.contents = contents;
+			this.retryInterval = retryInterval;
+		}
+
+		public String getZnodePath() {
+			return znodeName;
+		}
+
+		public byte[] getContents() {
+			return contents;
+		}
+
+		public int getRetryInterval() {
+			return retryInterval;
+		}
+		
+		@Override
+		public String toString() {
+			StringBuilder toString = new StringBuilder();
+			toString.append(this.getClass().getSimpleName());
+			toString.append(", Znode=" + getZnodePath());
+
+			return toString.toString();
+		}
+
+		@Override
+		public void run(Timeout timeout) throws Exception {
+			registerWebCluster(this);
+		}
+
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class AuthException extends PinpointZookeeperException {
+
+	public AuthException() {
+	}
+
+	public AuthException(String message) {
+		super(message);
+	}
+
+	public AuthException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public AuthException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class BadOperationException extends PinpointZookeeperException {
+
+	public BadOperationException() {
+	}
+
+	public BadOperationException(String message) {
+		super(message);
+	}
+
+	public BadOperationException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public BadOperationException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class ConnectionException extends PinpointZookeeperException {
+
+	public ConnectionException() {
+	}
+
+	public ConnectionException(String message) {
+		super(message);
+	}
+
+	public ConnectionException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public ConnectionException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class PinpointZookeeperException extends Exception {
+	
+	public PinpointZookeeperException() {
+	}
+
+	public PinpointZookeeperException(String message) {
+		super(message);
+	}
+
+	public PinpointZookeeperException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public PinpointZookeeperException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class TimeoutException extends PinpointZookeeperException {
+
+	public TimeoutException() {
+	}
+
+	public TimeoutException(String message) {
+		super(message);
+	}
+
+	public TimeoutException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public TimeoutException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.web.cluster.zookeeper.exception;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class UnknownException extends PinpointZookeeperException {
+
+	public UnknownException() {
+	}
+
+	public UnknownException(String message) {
+		super(message);
+	}
+
+	public UnknownException(String message, Throwable cause) {
+		super(message, cause);
+	}
+
+	public UnknownException(Throwable cause) {
+		super(cause);
+	}
+
+}
@@ -0,0 +1,95 @@
+package com.nhn.pinpoint.web.config;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class WebConfig {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+	
+	@Value("#{pinpointWebProps['cluster.enable'] ?: false}")
+	private boolean clusterEnable;
+	
+	@Value("#{pinpointWebProps['cluster.web.tcp.port'] ?: 0}")
+	private int clusterTcpPort;
+
+	@Value("#{pinpointWebProps['cluster.zookeeper.address'] ?: ''}")
+	private String clusterZookeeperAddress;
+
+	@Value("#{pinpointWebProps['cluster.zookeeper.sessiontimeout'] ?: -1}")
+	private int clusterZookeeperSessionTimeout;
+
+	@Value("#{pinpointWebProps['cluster.zookeeper.retry.interval'] ?: 60000}")
+	private int clusterZookeeperRetryInterval;
+	
+	@PostConstruct
+	public void validation() {
+		if (isClusterEnable()) {
+			assertPort(clusterTcpPort);
+			if(StringUtils.isEmpty(clusterZookeeperAddress)) {
+				throw new IllegalArgumentException("clusterZookeeperAddress may not be empty =" + clusterZookeeperAddress);
+			}
+			assertPositiveNumber(clusterZookeeperSessionTimeout);
+			assertPositiveNumber(clusterZookeeperRetryInterval);
+		}
+		
+		logger.info("{}", toString());
+	}
+
+	private boolean assertPort(int port) {
+		if (port > 0 && 65535 > port) {
+			return true;
+		}
+		
+		throw new IllegalArgumentException("Invalid Port =" + port);
+	}
+	
+	private boolean assertPositiveNumber(int number) {
+		if (number >= 0) {
+			return true;
+		}
+		
+		throw new IllegalArgumentException("Invalid Positive Number =" + number);
+	}
+
+	public boolean isClusterEnable() {
+		return clusterEnable;
+	}
+
+	public int getClusterTcpPort() {
+		return clusterTcpPort;
+	}
+
+	public String getClusterZookeeperAddress() {
+		return clusterZookeeperAddress;
+	}
+
+	public int getClusterZookeeperSessionTimeout() {
+		return clusterZookeeperSessionTimeout;
+	}
+	
+	@Override
+	public String toString() {
+		return "WebConfig [clusterEnable=" + clusterEnable
+				+ ", clusterTcpPort=" + clusterTcpPort
+				+ ", clusterZookeeperAddress=" + clusterZookeeperAddress
+				+ ", clusterZookeeperSessionTimeout="
+				+ clusterZookeeperSessionTimeout + "]";
+	}
+
+	public int getClusterZookeeperRetryInterval() {
+		return clusterZookeeperRetryInterval;
+	}
+
+	public void setClusterZookeeperRetryInterval(int clusterZookeeperRetryInterval) {
+		this.clusterZookeeperRetryInterval = clusterZookeeperRetryInterval;
+	}
+
+}
@@ -0,0 +1,156 @@
+package com.nhn.pinpoint.web.server;
+
+import java.io.IOException;
+import java.net.SocketException;
+import java.nio.charset.Charset;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+
+import org.apache.zookeeper.KeeperException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.nhn.pinpoint.common.util.NetUtils;
+import com.nhn.pinpoint.rpc.packet.ControlEnableWorkerConfirmPacket;
+import com.nhn.pinpoint.rpc.packet.RequestPacket;
+import com.nhn.pinpoint.rpc.packet.SendPacket;
+import com.nhn.pinpoint.rpc.packet.StreamPacket;
+import com.nhn.pinpoint.rpc.server.ChannelContext;
+import com.nhn.pinpoint.rpc.server.PinpointServerSocket;
+import com.nhn.pinpoint.rpc.server.ServerMessageListener;
+import com.nhn.pinpoint.rpc.server.ServerStreamChannel;
+import com.nhn.pinpoint.rpc.server.SocketChannel;
+import com.nhn.pinpoint.web.cluster.ClusterManager;
+import com.nhn.pinpoint.web.cluster.zookeeper.ZookeeperClusterManager;
+import com.nhn.pinpoint.web.config.WebConfig;
+
+/**
+ * @author koo.taejin <kr14910>
+ */
+public class PinpointSocketManager {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass().getName());
+	private final Charset charset = Charset.forName("UTF-8");
+
+	// 로컬 ip
+	// @Value("#{pinpointWebProps['web.tcpListenI']}")
+	private String representationLocalIp;
+	private List<String> localIpList;
+
+	private WebConfig config;
+
+	private final PinpointServerSocket pinpointServerSocket;
+
+	private ClusterManager clusterManager;
+
+	public PinpointSocketManager(WebConfig config) {
+		this.config = config;
+		this.pinpointServerSocket = new PinpointServerSocket();
+	}
+
+	@PostConstruct
+	public void start() throws KeeperException, IOException, InterruptedException {
+		logger.info("{} enable {}.", this.getClass().getSimpleName(), config.isClusterEnable());
+		if (config.isClusterEnable()) {
+			this.representationLocalIp = getRepresentationLocalV4Ip();
+			this.localIpList = NetUtils.getLocalV4IpList();
+
+			logger.info("Representation_Ip = {}, Ip_List = {}", representationLocalIp, localIpList);
+			
+			// 옵션으로 지정할수 있게 하면 좋을듯 뛰울껀지 말껀지
+			if (representationLocalIp.equals(NetUtils.LOOPBACK_ADDRESS_V4) || localIpList.size() == 0) {
+				throw new SocketException("Can't find Local Ip.");
+			}
+			
+			String nodeName = representationLocalIp + ":" + config.getClusterTcpPort();
+			if (!NetUtils.validationIpPortV4FormatAddress(nodeName)) {
+				throw new SocketException("Unexpected LocalAddress. LocalAddress format must be ip:port (" + nodeName + ").");
+			}
+
+			this.pinpointServerSocket.setMessageListener(new PinpointSocketManagerHandler());
+			this.pinpointServerSocket.bind(representationLocalIp, config.getClusterTcpPort());
+
+			this.clusterManager = new ZookeeperClusterManager(config.getClusterZookeeperAddress(), config.getClusterZookeeperSessionTimeout(), config.getClusterZookeeperRetryInterval());
+
+			// json list는 표준규칙이 아니기 때문에 ip\r\n으로 저장
+			this.clusterManager.registerWebCluster(nodeName, convertIpListToBytes(localIpList, "\r\n"));
+		}
+	}
+
+	@PreDestroy
+	public void stop() {
+		if (config.isClusterEnable()) {
+			if (clusterManager != null) {
+				clusterManager.close();
+			}
+			
+			if (pinpointServerSocket != null) {
+				pinpointServerSocket.close();
+			}
+		}
+	}
+
+	public List<ChannelContext> getCollectorChannelContext() {
+		return pinpointServerSocket.getDuplexCommunicationChannelContext();
+	}
+	
+	private String getRepresentationLocalV4Ip() {
+		String ip = NetUtils.getLocalV4Ip();
+		
+		if (!ip.equals(NetUtils.LOOPBACK_ADDRESS_V4)) {
+			return ip;
+		}
+		
+		// LOOPBACK Addess 다 제거하고 나옴
+		List<String> ipList = NetUtils.getLocalV4IpList();
+		if (ipList.size() > 0) {
+			return ipList.get(0);
+		}
+		
+		return NetUtils.LOOPBACK_ADDRESS_V4;
+	}
+
+	private byte[] convertIpListToBytes(List<String> ipList, String delimeter) {
+		StringBuilder stringBuilder = new StringBuilder();
+		
+		Iterator<String> ipIterator = ipList.iterator();
+		while (ipIterator.hasNext()) {
+			String eachIp = ipIterator.next();
+			stringBuilder.append(eachIp);
+			
+			if (ipIterator.hasNext()) {
+				stringBuilder.append(delimeter);
+			}
+		}
+		
+		return stringBuilder.toString().getBytes(charset);
+	}
+	
+	private class PinpointSocketManagerHandler implements ServerMessageListener {
+		@Override
+		public void handleSend(SendPacket sendPacket, SocketChannel channel) {
+			logger.warn("Unsupport send received {} {}", sendPacket, channel);
+		}
+
+		@Override
+		public void handleRequest(RequestPacket requestPacket, SocketChannel channel) {
+			logger.warn("Unsupport request received {} {}", requestPacket, channel);
+		}
+
+		@Override
+		public void handleStream(StreamPacket streamPacket, ServerStreamChannel streamChannel) {
+			logger.warn("unsupported streamPacket received {}", streamPacket);
+		}
+
+		@Override
+		public int handleEnableWorker(Map properties) {
+			logger.warn("do handleEnableWorker {}", properties);
+			return ControlEnableWorkerConfirmPacket.SUCCESS;
+		}
+	}
+
+}
@@ -0,0 +1,6 @@
+# dev
+cluster.enable=false
+cluster.web.tcp.port=9995
+cluster.zookeeper.address=dev.zk.pinpoint.navercorp.com
+cluster.zookeeper.sessiontimeout=3000
+cluster.zookeeper.retry.interval=60000
@@ -0,0 +1,6 @@
+# local
+cluster.enable=false
+cluster.web.tcp.port=9995
+cluster.zookeeper.address=127.0.0.1:22213
+cluster.zookeeper.sessiontimeout=3000
+cluster.zookeeper.retry.interval=5000
@@ -0,0 +1,6 @@
+# release
+cluster.enable=false
+cluster.web.tcp.port=9995
+cluster.zookeeper.address=zk.pinpoint.nhncorp.com
+cluster.zookeeper.sessiontimeout=3000
+cluster.zookeeper.retry.interval=60000
@@ -0,0 +1,6 @@
+# test
+cluster.enable=true
+cluster.web.tcp.port=9995
+cluster.zookeeper.address=127.0.0.1:22213
+cluster.zookeeper.sessiontimeout=3000
+cluster.zookeeper.retry.interval=5000
@@ -1,75 +1,84 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="http://www.springframework.org/schema/beans"
-       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xmlns:context="http://www.springframework.org/schema/context"
-       xmlns:lang="http://www.springframework.org/schema/lang"
-       xmlns:beans="http://www.springframework.org/schema/beans"
-       xmlns:util="http://www.springframework.org/schema/util" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
-       xmlns:hdp="http://www.springframework.org/schema/hadoop"
-       xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.0.xsd
-		http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang.xsd
-
-		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
-		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
-		http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
-
-    <context:annotation-config/>
-
-	<!--
-    <context:component-scan base-package="com.nhn.pinpoint.web.dao.hbase,
-    									 com.nhn.pinpoint.web.dao.mysql,
-                                         com.nhn.pinpoint.web.service,
-                                         com.nhn.pinpoint.web.mapper,
-                                         com.nhn.pinpoint.web.filter" />
-	-->
-
-    <context:component-scan base-package="com.nhn.pinpoint.web.dao.hbase,
-                                         com.nhn.pinpoint.web.service,
-                                         com.nhn.pinpoint.web.mapper,
-                                         com.nhn.pinpoint.web.filter" />
-
-    <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
-        <property name="locations">
-            <list>
-                <value>classpath:hbase.properties</value>
-                <!-- 
-                <value>classpath:jdbc.properties</value>
-                -->
-            </list>
-        </property>
-    </bean>
-    
-    <!-- 
-    <util:properties id="dataProps" location="classpath:data.properties"/>
-    -->
- 
-    <import resource="classpath:applicationContext-hbase.xml" />
-    
-    <!-- 
-    <import resource="classpath:applicationContext-datasource.xml" />
-    <import resource="classpath:applicationContext-dao-config.xml" />
-    <import resource="classpath:applicationContext-scheduler.xml" />
-    -->
-    <!--<import resource="classpath:applicationContext-cache.xml" />-->
-
-    <bean id="spanMapper" class="com.nhn.pinpoint.web.mapper.SpanMapper"></bean>
-    <bean id="annotationMapper" class="com.nhn.pinpoint.web.mapper.AnnotationMapper"></bean>
-    <bean id="spanAnnotationMapper" class="com.nhn.pinpoint.web.mapper.SpanMapper">
-        <property name="annotationMapper" ref="annotationMapper"/>
-    </bean>
-    
-    <bean id="jsonObjectMapper" class="com.fasterxml.jackson.databind.ObjectMapper">
-    </bean>
-
-    <bean id="rangeFactory" class="com.nhn.pinpoint.web.vo.RangeFactory">
-    </bean>
-
-    <bean id="timeSlot" class="com.nhn.pinpoint.common.util.DefaultTimeSlot">
-    </bean>
-
-	<!--
-    <bean id="mailResource" class="com.nhn.pinpoint.web.alarm.resource.MailResourceImpl" />
-    <bean id="smsResource" class="com.nhn.pinpoint.web.alarm.resource.SmsResourceImpl" />
-	-->   
-</beans>
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:context="http://www.springframework.org/schema/context"
+       xmlns:lang="http://www.springframework.org/schema/lang"
+       xmlns:beans="http://www.springframework.org/schema/beans"
+       xmlns:util="http://www.springframework.org/schema/util" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
+       xmlns:hdp="http://www.springframework.org/schema/hadoop"
+       xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-3.0.xsd
+		http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang.xsd
+
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/hadoop http://www.springframework.org/schema/hadoop/spring-hadoop.xsd">
+
+    <context:annotation-config/>
+
+	<!--
+    <context:component-scan base-package="com.nhn.pinpoint.web.dao.hbase,
+    									 com.nhn.pinpoint.web.dao.mysql,
+                                         com.nhn.pinpoint.web.service,
+                                         com.nhn.pinpoint.web.mapper,
+                                         com.nhn.pinpoint.web.filter" />
+	-->
+
+    <context:component-scan base-package="com.nhn.pinpoint.web.dao.hbase,
+                                         com.nhn.pinpoint.web.service,
+                                         com.nhn.pinpoint.web.mapper,
+                                         com.nhn.pinpoint.web.filter" />
+
+    <bean id="propertyConfigurer" class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
+        <property name="locations">
+            <list>
+                <value>classpath:hbase.properties</value>
+                <!-- 
+                <value>classpath:jdbc.properties</value>
+                -->
+            </list>
+        </property>
+    </bean>
+    
+    <util:properties id="pinpointWebProps" location="classpath:pinpoint-web.properties"/>
+    
+    <!-- 
+    <util:properties id="dataProps" location="classpath:data.properties"/>
+    -->
+ 
+    <import resource="classpath:applicationContext-hbase.xml" />
+    
+    <!-- 
+    <import resource="classpath:applicationContext-datasource.xml" />
+    <import resource="classpath:applicationContext-dao-config.xml" />
+    <import resource="classpath:applicationContext-scheduler.xml" />
+    -->
+    <!--<import resource="classpath:applicationContext-cache.xml" />-->
+
+    <bean id="spanMapper" class="com.nhn.pinpoint.web.mapper.SpanMapper"></bean>
+    <bean id="annotationMapper" class="com.nhn.pinpoint.web.mapper.AnnotationMapper"></bean>
+    <bean id="spanAnnotationMapper" class="com.nhn.pinpoint.web.mapper.SpanMapper">
+        <property name="annotationMapper" ref="annotationMapper"/>
+    </bean>
+    
+    <bean id="jsonObjectMapper" class="com.fasterxml.jackson.databind.ObjectMapper">
+    </bean>
+
+    <bean id="rangeFactory" class="com.nhn.pinpoint.web.vo.RangeFactory">
+    </bean>
+
+    <bean id="timeSlot" class="com.nhn.pinpoint.common.util.DefaultTimeSlot">
+    </bean>
+
+    <bean id="config" class="com.nhn.pinpoint.web.config.WebConfig">
+    </bean>
+
+    <bean id="pinpointSocketManager" class="com.nhn.pinpoint.web.server.PinpointSocketManager">
+    	<constructor-arg ref="config" />
+    </bean>
+
+	<!--
+    <bean id="mailResource" class="com.nhn.pinpoint.web.alarm.resource.MailResourceImpl" />
+    <bean id="smsResource" class="com.nhn.pinpoint.web.alarm.resource.SmsResourceImpl" />
+	-->   
+</beans>
@@ -0,0 +1,180 @@
+package com.nhn.pinpoint.web.cluster;
+
+import java.io.IOException;
+import java.util.List;
+
+import junit.framework.Assert;
+
+import org.apache.curator.test.TestingServer;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.ZooKeeper;
+import org.jboss.netty.channel.Channel;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.nhn.pinpoint.common.util.NetUtils;
+import com.nhn.pinpoint.rpc.client.MessageListener;
+import com.nhn.pinpoint.rpc.client.PinpointSocket;
+import com.nhn.pinpoint.rpc.client.PinpointSocketFactory;
+import com.nhn.pinpoint.rpc.packet.RequestPacket;
+import com.nhn.pinpoint.rpc.packet.SendPacket;
+import com.nhn.pinpoint.web.server.PinpointSocketManager;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration("classpath:applicationContext.xml")
+public class ClusterTest {
+
+	private static final int DEFAULT_ACCEPTOR_PORT = 9995;
+	private static final int DEFAULT_ZOOKEEPER_PORT = 22213;
+
+	private static final String DEFAULT_IP = NetUtils.getLocalV4Ip();
+
+	private static final String CLUSTER_NODE_PATH = "/pinpoint-cluster/web/" + DEFAULT_IP + ":" + DEFAULT_ACCEPTOR_PORT;
+
+	private static TestingServer ts = null;
+
+	@Autowired
+	PinpointSocketManager socketManager;
+
+	@BeforeClass
+	public static void setUp() throws Exception {
+		ts = createZookeeperServer(DEFAULT_ZOOKEEPER_PORT);
+	}
+
+	@AfterClass
+	public static void tearDown() throws Exception {
+		closeZookeeperServer(ts);
+	}
+
+	@Before
+	public void before() throws IOException {
+		ts.stop();
+	}
+
+	// ApplicationContext 설정에 맞게 등록이 되는지
+	@Test
+	public void clusterTest1() throws Exception {
+		ts.restart();
+		Thread.sleep(5000);
+
+		ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:22213", 5000, null);
+		getNodeAndCompareContents(zookeeper);
+
+	}
+
+	// ApplicationContext 설정에 맞게 등록이 되는지
+	@Test
+	public void clusterTest2() throws Exception {
+		ts.restart();
+		Thread.sleep(5000);
+
+		ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:22213", 5000, null);
+		getNodeAndCompareContents(zookeeper);
+
+		ts.stop();
+
+		Thread.sleep(5000);
+		try {
+			zookeeper.getData(CLUSTER_NODE_PATH, null, null);
+			Assert.fail();
+		} catch (KeeperException e) {
+			Assert.assertEquals(KeeperException.Code.CONNECTIONLOSS, e.code());
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+
+		ts.restart();
+
+		getNodeAndCompareContents(zookeeper);
+
+	}
+
+	// ApplicationContext 설정에 맞게 등록이 되는지
+	@Test
+	public void clusterTest3() throws Exception {
+		ts.restart();
+
+		PinpointSocketFactory factory = null;
+		PinpointSocket socket = null;
+		try {
+			Thread.sleep(5000);
+
+			ZooKeeper zookeeper = new ZooKeeper("127.0.0.1:22213", 5000, null);
+			getNodeAndCompareContents(zookeeper);
+
+			Assert.assertEquals(0, socketManager.getCollectorChannelContext().size());
+
+			factory = new PinpointSocketFactory();
+			socket = factory.connect(DEFAULT_IP, DEFAULT_ACCEPTOR_PORT, new SimpleListener());
+
+			Thread.sleep(1000);
+
+			Assert.assertEquals(1, socketManager.getCollectorChannelContext().size());
+
+		} finally {
+			closePinpointSocket(factory, socket);
+
+		}
+	}
+
+	private static TestingServer createZookeeperServer(int port) throws Exception {
+		TestingServer mockZookeeperServer = new TestingServer(port);
+		mockZookeeperServer.start();
+
+		return mockZookeeperServer;
+	}
+
+	private static void closeZookeeperServer(TestingServer mockZookeeperServer) throws Exception {
+		try {
+			if (mockZookeeperServer != null) {
+				mockZookeeperServer.close();
+			}
+		} catch (Exception e) {
+			e.printStackTrace();
+		}
+	}
+
+	private void getNodeAndCompareContents(ZooKeeper zookeeper) throws KeeperException, InterruptedException {
+		byte[] conetents = zookeeper.getData(CLUSTER_NODE_PATH, null, null);
+
+		String[] registeredIplist = new String(conetents).split("\r\n");
+
+		List<String> ipList = NetUtils.getLocalV4IpList();
+
+		Assert.assertEquals(registeredIplist.length, ipList.size());
+
+		for (String ip : registeredIplist) {
+			Assert.assertTrue(ipList.contains(ip));
+		}
+	}
+
+	private void closePinpointSocket(PinpointSocketFactory factory, PinpointSocket socket) {
+		if (socket != null) {
+			socket.close();
+		}
+
+		if (factory != null) {
+			factory.release();
+		}
+	}
+
+	class SimpleListener implements MessageListener {
+		@Override
+		public void handleSend(SendPacket sendPacket, Channel channel) {
+
+		}
+
+		@Override
+		public void handleRequest(RequestPacket requestPacket, Channel channel) {
+			// TODO Auto-generated method stub
+
+		}
+	}
+
+}