From 3cec9281fe5f51a540388afd363f58ef7a13814a Mon Sep 17 00:00:00 2001 From: koo-taejin Date: Mon, 1 Sep 2014 18:10:58 +0900 Subject: [PATCH 1/4] =?UTF-8?q?#5=20Pinpoint=20=ED=81=B4=EB=9F=AC=EC=8A=A4?= =?UTF-8?q?=ED=84=B0=20=EA=B0=9C=EB=B0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Pinpoint Web Acceptor 등록 - Pinpoint Web Acceptor 등록시 등로된 Acceptor의 정보를 (쥬키퍼)클러스터에 저장하게 함 - 클러스터 저장은 다음과 같음 znode /pinpoint-cluster/web/ip:port contents ip1 ip2 ip3 - 해당 기능 설정은 pinpoint-web.properties를 통해 할수 있음 --- .../pinpoint/common/util/NetUtils.java | 158 ++++++++++++++ web/pom.xml | 8 + .../pinpoint/web/cluster/ClusterManager.java | 12 ++ .../cluster/zookeeper/ZookeeperClient.java | 165 +++++++++++++++ .../zookeeper/ZookeeperClusterManager.java | 198 ++++++++++++++++++ .../zookeeper/exception/AuthException.java | 23 ++ .../exception/BadOperationException.java | 23 ++ .../exception/ConnectionException.java | 23 ++ .../exception/PinpointZookeeperException.java | 23 ++ .../zookeeper/exception/TimeoutException.java | 23 ++ .../zookeeper/exception/UnknownException.java | 23 ++ .../pinpoint/web/config/WebConfig.java | 89 ++++++++ .../web/server/PinpointSocketManager.java | 156 ++++++++++++++ .../resources-dev/pinpoint-web.properties | 6 + .../resources-local/pinpoint-web.properties | 6 + .../resources-release/pinpoint-web.properties | 6 + .../resources-test/pinpoint-web.properties | 6 + web/src/main/resources/applicationContext.xml | 159 +++++++------- .../pinpoint/web/cluster/ClusterTest.java | 180 ++++++++++++++++ 19 files changed, 1212 insertions(+), 75 deletions(-) create mode 100644 commons/src/main/java/com/navercorp/pinpoint/common/util/NetUtils.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/ClusterManager.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClient.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/AuthException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/BadOperationException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/ConnectionException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/PinpointZookeeperException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/TimeoutException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/UnknownException.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java create mode 100644 web/src/main/java/com/navercorp/pinpoint/web/server/PinpointSocketManager.java create mode 100644 web/src/main/resources-dev/pinpoint-web.properties create mode 100644 web/src/main/resources-local/pinpoint-web.properties create mode 100644 web/src/main/resources-release/pinpoint-web.properties create mode 100644 web/src/main/resources-test/pinpoint-web.properties create mode 100644 web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java diff --git a/commons/src/main/java/com/navercorp/pinpoint/common/util/NetUtils.java b/commons/src/main/java/com/navercorp/pinpoint/common/util/NetUtils.java new file mode 100644 index 000000000..4d7922273 --- /dev/null +++ b/commons/src/main/java/com/navercorp/pinpoint/common/util/NetUtils.java @@ -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 + */ +public final class NetUtils { + + public static final String LOOPBACK_ADDRESS_V4 = "127.0.0.1"; + + private NetUtils() { + } + + public static List toInetSocketAddressLIst(List addressList) { + List inetSocketAddressList = new ArrayList(); + + 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 getLocalV4IpList() { + List result = new ArrayList(); + + Enumeration 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 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; + } + +} diff --git a/web/pom.xml b/web/pom.xml index 0a85f722c..0c6c1818f 100644 --- a/web/pom.xml +++ b/web/pom.xml @@ -612,6 +612,14 @@ org.aspectj aspectjweaver + + + org.apache.curator + curator-test + 2.6.0 + test + + diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/ClusterManager.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/ClusterManager.java new file mode 100644 index 000000000..7279225c6 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/ClusterManager.java @@ -0,0 +1,12 @@ +package com.nhn.pinpoint.web.cluster; + +/** + * @author koo.taejin + */ +public interface ClusterManager { + + boolean registerWebCluster(String zNodeName, byte[] contents); + + void close(); + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClient.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClient.java new file mode 100644 index 000000000..72ff09bbe --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClient.java @@ -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 + */ +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); + } + } + } + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java new file mode 100644 index 000000000..388aac8f2 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java @@ -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 + */ +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 job = new AtomicReference(); + + 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); + } + + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/AuthException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/AuthException.java new file mode 100644 index 000000000..31813ee26 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/AuthException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/BadOperationException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/BadOperationException.java new file mode 100644 index 000000000..68fd4d0a9 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/BadOperationException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/ConnectionException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/ConnectionException.java new file mode 100644 index 000000000..fed2285dc --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/ConnectionException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/PinpointZookeeperException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/PinpointZookeeperException.java new file mode 100644 index 000000000..82ad3fdfa --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/PinpointZookeeperException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/TimeoutException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/TimeoutException.java new file mode 100644 index 000000000..20266eee6 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/TimeoutException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/UnknownException.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/UnknownException.java new file mode 100644 index 000000000..a974f1b69 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/exception/UnknownException.java @@ -0,0 +1,23 @@ +package com.nhn.pinpoint.web.cluster.zookeeper.exception; + +/** + * @author koo.taejin + */ +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); + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java b/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java new file mode 100644 index 000000000..fa220b7d6 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java @@ -0,0 +1,89 @@ +package com.nhn.pinpoint.web.config; + +import javax.annotation.PostConstruct; + +import org.apache.commons.lang.StringUtils; +import org.springframework.beans.factory.annotation.Value; + +/** + * @author koo.taejin + */ +public class WebConfig { + + @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); + } + } + + 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; + } + +} diff --git a/web/src/main/java/com/navercorp/pinpoint/web/server/PinpointSocketManager.java b/web/src/main/java/com/navercorp/pinpoint/web/server/PinpointSocketManager.java new file mode 100644 index 000000000..f820d41e7 --- /dev/null +++ b/web/src/main/java/com/navercorp/pinpoint/web/server/PinpointSocketManager.java @@ -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 + */ +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 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 getCollectorChannelContext() { + return pinpointServerSocket.getDuplexCommunicationChannelContext(); + } + + private String getRepresentationLocalV4Ip() { + String ip = NetUtils.getLocalV4Ip(); + + if (!ip.equals(NetUtils.LOOPBACK_ADDRESS_V4)) { + return ip; + } + + // LOOPBACK Addess 다 제거하고 나옴 + List ipList = NetUtils.getLocalV4IpList(); + if (ipList.size() > 0) { + return ipList.get(0); + } + + return NetUtils.LOOPBACK_ADDRESS_V4; + } + + private byte[] convertIpListToBytes(List ipList, String delimeter) { + StringBuilder stringBuilder = new StringBuilder(); + + Iterator 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; + } + } + +} diff --git a/web/src/main/resources-dev/pinpoint-web.properties b/web/src/main/resources-dev/pinpoint-web.properties new file mode 100644 index 000000000..646ea1d30 --- /dev/null +++ b/web/src/main/resources-dev/pinpoint-web.properties @@ -0,0 +1,6 @@ +# local +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 \ No newline at end of file diff --git a/web/src/main/resources-local/pinpoint-web.properties b/web/src/main/resources-local/pinpoint-web.properties new file mode 100644 index 000000000..d529268c9 --- /dev/null +++ b/web/src/main/resources-local/pinpoint-web.properties @@ -0,0 +1,6 @@ +# local +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 \ No newline at end of file diff --git a/web/src/main/resources-release/pinpoint-web.properties b/web/src/main/resources-release/pinpoint-web.properties new file mode 100644 index 000000000..1a4e95d94 --- /dev/null +++ b/web/src/main/resources-release/pinpoint-web.properties @@ -0,0 +1,6 @@ +# local +cluster.enable=false +cluster.web.tcp.port=9995 +cluster.zookeeper.address=zk.pinpoint.nhncorp.com +cluster.zookeeper.sessiontimeout=3000 +cluster.zookeeper.retry.interval=60000 \ No newline at end of file diff --git a/web/src/main/resources-test/pinpoint-web.properties b/web/src/main/resources-test/pinpoint-web.properties new file mode 100644 index 000000000..d529268c9 --- /dev/null +++ b/web/src/main/resources-test/pinpoint-web.properties @@ -0,0 +1,6 @@ +# local +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 \ No newline at end of file diff --git a/web/src/main/resources/applicationContext.xml b/web/src/main/resources/applicationContext.xml index a2299b8af..0ad883872 100644 --- a/web/src/main/resources/applicationContext.xml +++ b/web/src/main/resources/applicationContext.xml @@ -1,75 +1,84 @@ - - - - - - - - - - - - - classpath:hbase.properties - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + classpath:hbase.properties + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java b/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java new file mode 100644 index 000000000..d340f1099 --- /dev/null +++ b/web/src/test/java/com/navercorp/pinpoint/web/cluster/ClusterTest.java @@ -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 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 + + } + } + +} From 4c928d8f4f787a1e47a4f22cec1617f68f2ce6bd Mon Sep 17 00:00:00 2001 From: koo-taejin Date: Mon, 1 Sep 2014 18:16:07 +0900 Subject: [PATCH 2/4] =?UTF-8?q?#5=20Pinpoint=20=ED=81=B4=EB=9F=AC=EC=8A=A4?= =?UTF-8?q?=ED=84=B0=20=EA=B0=9C=EB=B0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit log 추가 및 변경 --- .../web/cluster/zookeeper/ZookeeperClusterManager.java | 2 +- .../java/com/navercorp/pinpoint/web/config/WebConfig.java | 6 ++++++ 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java index 388aac8f2..b6284611e 100644 --- a/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java +++ b/web/src/main/java/com/navercorp/pinpoint/web/cluster/zookeeper/ZookeeperClusterManager.java @@ -56,7 +56,7 @@ public class ZookeeperClusterManager implements ClusterManager, Watcher { public boolean registerWebCluster(String zNodeName, byte[] contents) { String zNodePath = bindingPathAndZnode(PINPOINT_WEB_CLUSTER_PATh, zNodeName); - logger.info("Create Web Cluster Zookeeper UniqPath = " + zNodePath); + logger.info("Create Web Cluster Zookeeper UniqPath = {}", zNodePath); RegisterJob job = new RegisterJob(zNodePath, contents, retryInterval); if (!this.job.compareAndSet(null, job)) { diff --git a/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java b/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java index fa220b7d6..3f7a2b60a 100644 --- a/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java +++ b/web/src/main/java/com/navercorp/pinpoint/web/config/WebConfig.java @@ -3,6 +3,8 @@ 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; /** @@ -10,6 +12,8 @@ import org.springframework.beans.factory.annotation.Value; */ public class WebConfig { + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + @Value("#{pinpointWebProps['cluster.enable'] ?: false}") private boolean clusterEnable; @@ -35,6 +39,8 @@ public class WebConfig { assertPositiveNumber(clusterZookeeperSessionTimeout); assertPositiveNumber(clusterZookeeperRetryInterval); } + + logger.info("{}", toString()); } private boolean assertPort(int port) { From e1edaf20890a4a957f978202f8ce0fafa7f94bc3 Mon Sep 17 00:00:00 2001 From: koo-taejin Date: Mon, 1 Sep 2014 18:18:33 +0900 Subject: [PATCH 3/4] =?UTF-8?q?#5=20Pinpoint=20=ED=81=B4=EB=9F=AC=EC=8A=A4?= =?UTF-8?q?=ED=84=B0=20=EA=B0=9C=EB=B0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 환경별 설정 이름 변경 --- web/src/main/resources-dev/pinpoint-web.properties | 2 +- web/src/main/resources-release/pinpoint-web.properties | 2 +- web/src/main/resources-test/pinpoint-web.properties | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/web/src/main/resources-dev/pinpoint-web.properties b/web/src/main/resources-dev/pinpoint-web.properties index 646ea1d30..068760332 100644 --- a/web/src/main/resources-dev/pinpoint-web.properties +++ b/web/src/main/resources-dev/pinpoint-web.properties @@ -1,4 +1,4 @@ -# local +# dev cluster.enable=false cluster.web.tcp.port=9995 cluster.zookeeper.address=dev.zk.pinpoint.navercorp.com diff --git a/web/src/main/resources-release/pinpoint-web.properties b/web/src/main/resources-release/pinpoint-web.properties index 1a4e95d94..05258dd45 100644 --- a/web/src/main/resources-release/pinpoint-web.properties +++ b/web/src/main/resources-release/pinpoint-web.properties @@ -1,4 +1,4 @@ -# local +# release cluster.enable=false cluster.web.tcp.port=9995 cluster.zookeeper.address=zk.pinpoint.nhncorp.com diff --git a/web/src/main/resources-test/pinpoint-web.properties b/web/src/main/resources-test/pinpoint-web.properties index d529268c9..b85ad7907 100644 --- a/web/src/main/resources-test/pinpoint-web.properties +++ b/web/src/main/resources-test/pinpoint-web.properties @@ -1,4 +1,4 @@ -# local +# test cluster.enable=true cluster.web.tcp.port=9995 cluster.zookeeper.address=127.0.0.1:22213 From 4cfc8516292057bd3d96ea4be2cdc0b288897256 Mon Sep 17 00:00:00 2001 From: koo-taejin Date: Mon, 1 Sep 2014 18:40:01 +0900 Subject: [PATCH 4/4] =?UTF-8?q?#5=20Pinpoint=20=ED=81=B4=EB=9F=AC=EC=8A=A4?= =?UTF-8?q?=ED=84=B0=20=EA=B0=9C=EB=B0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. 로컬 설정을 되돌림 테스트시 실패가 가능하기 때문에 --- web/src/main/resources-local/pinpoint-web.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/web/src/main/resources-local/pinpoint-web.properties b/web/src/main/resources-local/pinpoint-web.properties index d529268c9..0b0fbffc4 100644 --- a/web/src/main/resources-local/pinpoint-web.properties +++ b/web/src/main/resources-local/pinpoint-web.properties @@ -1,5 +1,5 @@ # local -cluster.enable=true +cluster.enable=false cluster.web.tcp.port=9995 cluster.zookeeper.address=127.0.0.1:22213 cluster.zookeeper.sessiontimeout=3000