#82 collector에서 remote command 요청시 추가작업을 할수 있게 filter 코드 등록

1. route 시에 filter 부분 등록 가능하게 변경

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ChannelContextClusterPoint.java

@@ -55,13 +55,11 @@ public class ChannelContextClusterPoint implements TargetClusterPoint {
 
 	@Override
 	public String getApplicationName() {
-		// TODO Auto-generated method stub
 		return applicationName;
 	}
 
 	@Override
 	public String getAgentId() {
-		// TODO Auto-generated method stub
 		return agentId;
 	}
 

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouter.java

@@ -1,8 +1,5 @@
 package com.nhn.pinpoint.collector.cluster;
 
-import java.util.ArrayList;
-import java.util.List;
-
 import javax.annotation.PreDestroy;
 
 import org.apache.thrift.TBase;
@@ -11,9 +8,12 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 
-import com.nhn.pinpoint.collector.util.CollectorUtils;
-import com.nhn.pinpoint.rpc.Future;
-import com.nhn.pinpoint.rpc.ResponseMessage;
+import com.nhn.pinpoint.collector.cluster.route.DefaultRouteHandler;
+import com.nhn.pinpoint.collector.cluster.route.LoggingFilter;
+import com.nhn.pinpoint.collector.cluster.route.RequestEvent;
+import com.nhn.pinpoint.collector.cluster.route.RouteHandler;
+import com.nhn.pinpoint.collector.cluster.route.RouteResult;
+import com.nhn.pinpoint.collector.cluster.route.RouteStatus;
 import com.nhn.pinpoint.rpc.client.MessageListener;
 import com.nhn.pinpoint.rpc.packet.RequestPacket;
 import com.nhn.pinpoint.rpc.packet.ResponsePacket;
@@ -24,7 +24,6 @@ import com.nhn.pinpoint.thrift.io.DeserializerFactory;
 import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializer;
 import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializer;
 import com.nhn.pinpoint.thrift.io.SerializerFactory;
-import com.nhn.pinpoint.thrift.io.TCommandTypeVersion;
 import com.nhn.pinpoint.thrift.util.SerializationUtils;
 
 /**
@@ -36,6 +35,8 @@ public class ClusterPointRouter implements MessageListener {
 
 	private final ClusterPointRepository<TargetClusterPoint> targetClusterPointRepository;
 
+	private final RouteHandler routeHandler;
+	
 	@Autowired
 	private SerializerFactory<HeaderTBaseSerializer> commandSerializerFactory;
 
@@ -44,6 +45,11 @@ public class ClusterPointRouter implements MessageListener {
 
 	public ClusterPointRouter() {
 		this.targetClusterPointRepository = new ClusterPointRepository<TargetClusterPoint>();
+		this.routeHandler = new DefaultRouteHandler(targetClusterPointRepository);
+
+		LoggingFilter loggingFilter = new LoggingFilter();
+		this.routeHandler.addRequestFilter(loggingFilter.getRequestFilter());
+		this.routeHandler.addResponseFilter(loggingFilter.getResponseFilter());
 	}
 
 	@PreDestroy
@@ -67,33 +73,16 @@ public class ClusterPointRouter implements MessageListener {
 
 			channel.write(new ResponsePacket(requestPacket.getRequestId(), serialize(tResult)));
 		} else if (request instanceof TCommandTransfer) {
-
-			String applicationName = ((TCommandTransfer) request).getApplicationName();
-			String agentId = ((TCommandTransfer) request).getAgentId();
-			long startTimeStamp = ((TCommandTransfer) request).getStartTime();
-
 			byte[] payload = ((TCommandTransfer) request).getPayload();
-
-			List<TargetClusterPoint> clusterPointList = targetClusterPointRepository.getClusterPointList();
-			TargetClusterPoint clusterPoint = findClusterPoint(applicationName, agentId, startTimeStamp, clusterPointList);
-			if (clusterPoint == null) {
-				TResult result = new TResult(false);
-				result.setMessage(applicationName + "/" + agentId + " can't find suitable ChannelContext.");
-				channel.write(new ResponsePacket(requestPacket.getRequestId(), serialize(result)));
-				return;
-			}
-
 			TBase command = deserialize(payload);
-			TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(clusterPoint.gerVersion());
-			if (commandVersion.isSupportCommand(command)) {
-				Future<ResponseMessage> future = clusterPoint.request(payload);
-				future.await();
-				ResponseMessage responseMessage = future.getResult();
-
-				channel.write(new ResponsePacket(requestPacket.getRequestId(), responseMessage.getMessage()));
+			
+			RouteResult routeResult = routeHandler.onRoute(new RequestEvent((TCommandTransfer) request, requestPacket.getRequestId(), channel, command));
+			
+			if (RouteStatus.OK == routeResult.getStatus()) {
+				channel.write(new ResponsePacket(requestPacket.getRequestId(), routeResult.getResponseMessage().getMessage()));
 			} else {
 				TResult result = new TResult(false);
-				result.setMessage(applicationName + "/" + agentId + " unsupported command(" + command + ") type.");
+				result.setMessage(routeResult.getStatus().getReasonPhrase());
 
 				channel.write(new ResponsePacket(requestPacket.getRequestId(), serialize(result)));
 			}
@@ -105,7 +94,6 @@ public class ClusterPointRouter implements MessageListener {
 		}
 
 	}
-
 	
 	public ClusterPointRepository<TargetClusterPoint> getTargetClusterPointRepository() {
 		return targetClusterPointRepository;
@@ -119,37 +107,4 @@ public class ClusterPointRouter implements MessageListener {
 		return SerializationUtils.deserialize(objectData, commandDeserializerFactory, null);
 	}
 
-	
-	private TargetClusterPoint findClusterPoint(String applicationName, String agentId, long startTimeStamp, List<TargetClusterPoint> targetClusterPointList) {
-
-		List<TargetClusterPoint> result = new ArrayList<TargetClusterPoint>();
-		
-		for (TargetClusterPoint targetClusterPoint : targetClusterPointList) {
-			if (!targetClusterPoint.getApplicationName().equals(applicationName)) {
-				continue;
-			}
-			
-			if (!targetClusterPoint.getAgentId().equals(agentId)) {
-				continue;
-			}
-			
-			if (!(targetClusterPoint.getStartTimeStamp() == startTimeStamp)) {
-				continue;
-			}
-
-			result.add(targetClusterPoint);
-		}
-		
-		if (result.size() == 1) {
-			return result.get(0);
-		}
-		
-		if (result.size() > 1) {
-    		logger.warn("Ambiguous ClusterPoint {}, {}, {} (Valid Agent list={}).", applicationName, agentId, startTimeStamp, result);
-			return null;
-		}
-		
-		return null;
-	}
-
 }

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteEvent.java

@@ -0,0 +1,53 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import org.jboss.netty.channel.Channel;
+
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+
+public class DefaultRouteEvent implements RouteEvent {
+
+	private final TCommandTransfer deliveryCommand;
+	
+	private final int requestId;
+	
+	private final Channel sourceChannel;
+	
+	
+	public DefaultRouteEvent(TCommandTransfer deliveryCommand, int requestId, Channel sourceChannel) {
+		this.deliveryCommand = deliveryCommand;
+		
+		this.requestId = requestId;
+		
+		this.sourceChannel = sourceChannel;
+	}
+
+	@Override
+	public TCommandTransfer getDeliveryCommand() {
+		return deliveryCommand;
+	}
+
+	@Override
+	public int getRequestId() {
+		return requestId;
+	}
+
+	@Override
+	public Channel getSourceChannel() {
+		return sourceChannel;
+	}
+
+	@Override
+	public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append(this.getClass().getSimpleName());
+        sb.append("{");
+        sb.append("{sourceChannel=").append(sourceChannel).append(",");
+        sb.append("requestId=").append(requestId).append(",");
+        sb.append("applicationName=").append(deliveryCommand.getApplicationName()).append(",");
+        sb.append("agentId=").append(deliveryCommand.getAgentId()).append(",");
+        sb.append("startTimeStamp=").append(deliveryCommand.getStartTime());
+        sb.append('}');
+        return sb.toString();
+	}
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteFilterChain.java

@@ -0,0 +1,32 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultRouteFilterChain<T extends RouteEvent> implements RouteFilterChain<T> {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+	
+	private final CopyOnWriteArrayList<RouteFilter<T>> filterList = new CopyOnWriteArrayList<RouteFilter<T>>();
+	
+	@Override
+	public void addLast(RouteFilter<T> filter) {
+		filterList.add(filter);
+	}
+
+	@Override
+	public void doEvent(T event) {
+		for (RouteFilter<T> filter : filterList) {
+			try {
+				filter.doEvent(event);
+			} catch (Exception e) {
+				if (logger.isWarnEnabled()) {
+					logger.warn(filter.getClass().getSimpleName() + " filter occured exception. caused=" + e.getMessage() + ".", e);
+				}
+			}
+		}
+	}
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteHandler.java

@@ -0,0 +1,120 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.thrift.TBase;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.nhn.pinpoint.collector.cluster.ClusterPointLocator;
+import com.nhn.pinpoint.collector.cluster.TargetClusterPoint;
+import com.nhn.pinpoint.rpc.Future;
+import com.nhn.pinpoint.rpc.ResponseMessage;
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+import com.nhn.pinpoint.thrift.io.TCommandTypeVersion;
+
+public class DefaultRouteHandler implements RouteHandler {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	private final RouteFilterChain<RequestEvent> requestFilterChain;
+	private final RouteFilterChain<ResponseEvent> responseFilterChain;
+
+	private final ClusterPointLocator<TargetClusterPoint> targetClusterPointLocator;
+
+	public DefaultRouteHandler(ClusterPointLocator<TargetClusterPoint> targetClusterPointLocator) {
+		this.targetClusterPointLocator = targetClusterPointLocator;
+
+		this.requestFilterChain = new DefaultRouteFilterChain<RequestEvent>();
+		this.responseFilterChain = new DefaultRouteFilterChain<ResponseEvent>();
+	}
+
+	@Override
+	public void addRequestFilter(RouteFilter<RequestEvent> filter) {
+		this.requestFilterChain.addLast(filter);
+	}
+
+	@Override
+	public void addResponseFilter(RouteFilter<ResponseEvent> filter) {
+		this.responseFilterChain.addLast(filter);
+	}
+
+	@Override
+	public RouteResult onRoute(RequestEvent event) {
+		requestFilterChain.doEvent(event);
+
+		RouteResult routeResult = onRoute0(event);
+
+		responseFilterChain.doEvent(new ResponseEvent(event, routeResult));
+
+		return routeResult;
+
+	}
+
+	private RouteResult onRoute0(RequestEvent event) {
+		TBase requestObject = event.getRequestObject();
+		if (requestObject == null) {
+			return new RouteResult(RouteStatus.BAD_REQUEST);
+		}
+
+		TargetClusterPoint clusterPoint = findClusterPoint(event);
+		if (clusterPoint == null) {
+			return new RouteResult(RouteStatus.NOT_FOUND);
+		}
+
+		TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(clusterPoint.gerVersion());
+		if (!commandVersion.isSupportCommand(requestObject)) {
+			return new RouteResult(RouteStatus.NOT_ACCEPTABLE);
+		}
+
+		Future<ResponseMessage> future = clusterPoint.request(event.getDeliveryCommand().getPayload());
+		future.await();
+		ResponseMessage responseMessage = future.getResult();
+
+		if (responseMessage == null) {
+			return new RouteResult(RouteStatus.AGENT_TIMEOUT);
+		}
+
+		return new RouteResult(RouteStatus.OK, responseMessage);
+
+	}
+
+	private TargetClusterPoint findClusterPoint(RequestEvent event) {
+		TCommandTransfer deliveryCommand = event.getDeliveryCommand();
+
+		String applicationName = deliveryCommand.getApplicationName();
+		String agentId = deliveryCommand.getAgentId();
+		long startTimeStamp = deliveryCommand.getStartTime();
+
+		List<TargetClusterPoint> result = new ArrayList<TargetClusterPoint>();
+
+		for (TargetClusterPoint targetClusterPoint : targetClusterPointLocator.getClusterPointList()) {
+			if (!targetClusterPoint.getApplicationName().equals(applicationName)) {
+				continue;
+			}
+
+			if (!targetClusterPoint.getAgentId().equals(agentId)) {
+				continue;
+			}
+
+			if (!(targetClusterPoint.getStartTimeStamp() == startTimeStamp)) {
+				continue;
+			}
+
+			result.add(targetClusterPoint);
+		}
+
+		if (result.size() == 1) {
+			return result.get(0);
+		}
+
+		if (result.size() > 1) {
+			logger.warn("Ambiguous ClusterPoint {}, {}, {} (Valid Agent list={}).", applicationName, agentId, startTimeStamp, result);
+			return null;
+		}
+
+		return null;
+	}
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/LoggingFilter.java

@@ -0,0 +1,36 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class LoggingFilter {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	class RequestFilter implements RouteFilter<RequestEvent> {
+
+		@Override
+		public void doEvent(RequestEvent event) {
+			logger.warn("{} doEvent {}.", this.getClass().getSimpleName(), event);
+		}
+
+	}
+
+	class ResponseFilter implements RouteFilter<ResponseEvent> {
+
+		@Override
+		public void doEvent(ResponseEvent event) {
+			logger.warn("{} doEvent {}.", this.getClass().getSimpleName(), event);
+		}
+
+	}
+
+	public RequestFilter getRequestFilter() {
+		return new RequestFilter();
+	}
+
+	public ResponseFilter getResponseFilter() {
+		return new ResponseFilter();
+	}
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RequestEvent.java

@@ -0,0 +1,36 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import org.apache.thrift.TBase;
+import org.jboss.netty.channel.Channel;
+
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+
+public class RequestEvent extends DefaultRouteEvent {
+
+	private final TBase requestObject;
+
+	public RequestEvent(RouteEvent routeEvent, TBase requestObject) {
+		this(routeEvent.getDeliveryCommand(), routeEvent.getRequestId(), routeEvent.getSourceChannel(), requestObject);
+	}
+
+	public RequestEvent(TCommandTransfer deliveryCommand, int requestId, Channel sourceChannel, TBase requestObject) {
+		super(deliveryCommand, requestId, sourceChannel);
+
+		this.requestObject = requestObject;
+	}
+
+	public TBase getRequestObject() {
+		return requestObject;
+	}
+
+	@Override
+	public String toString() {
+		final StringBuilder sb = new StringBuilder(super.toString());
+        sb.append("{");
+        sb.append("requestObject=").append(requestObject);
+        sb.append("}");
+
+		return super.toString();
+	}
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/ResponseEvent.java

@@ -0,0 +1,35 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import org.jboss.netty.channel.Channel;
+
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+
+public class ResponseEvent extends DefaultRouteEvent {
+
+	private final RouteResult routeResult;
+
+	public ResponseEvent(RouteEvent routeEvent, RouteResult routeResult) {
+		this(routeEvent.getDeliveryCommand(), routeEvent.getRequestId(), routeEvent.getSourceChannel(), routeResult);
+	}
+
+	public ResponseEvent(TCommandTransfer deliveryCommand, int requestId, Channel sourceChannel, RouteResult routeResult) {
+		super(deliveryCommand, requestId, sourceChannel);
+		this.routeResult = routeResult;
+	}
+
+	public RouteResult getRouteResult() {
+		return routeResult;
+	}
+	
+	@Override
+	public String toString() {
+		final StringBuilder sb = new StringBuilder(super.toString());
+        sb.append("{");
+        sb.append("routeResult=").append(routeResult);
+        sb.append("}");
+
+		return super.toString();
+	}
+
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteEvent.java

@@ -0,0 +1,14 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import org.jboss.netty.channel.Channel;
+
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+
+public interface RouteEvent {
+
+	TCommandTransfer getDeliveryCommand();
+	
+	int getRequestId();
+	Channel getSourceChannel();
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilter.java

@@ -0,0 +1,7 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+public interface RouteFilter<T extends RouteEvent> {
+
+	void doEvent(T event);
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilterChain.java

@@ -0,0 +1,9 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+public interface RouteFilterChain<T extends RouteEvent> {
+
+	void addLast(RouteFilter<T> filter);
+	
+	void doEvent(T event);
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteHandler.java

@@ -0,0 +1,12 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+
+public interface RouteHandler {
+
+	void addRequestFilter(RouteFilter<RequestEvent> filter);
+	
+	void addResponseFilter(RouteFilter<ResponseEvent> filter);
+	
+	RouteResult onRoute(RequestEvent event);
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteResult.java

@@ -0,0 +1,38 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+import com.nhn.pinpoint.rpc.ResponseMessage;
+import com.nhn.pinpoint.rpc.util.AssertUtils;
+
+public class RouteResult {
+
+	private final RouteStatus status;
+	private final ResponseMessage responseMessage;
+
+	public RouteResult(RouteStatus status) {
+		this(status, null);
+	}
+
+	public RouteResult(RouteStatus status, ResponseMessage responseMessage) {
+		this.status = status;
+		this.responseMessage = responseMessage;
+
+		if (RouteStatus.OK == status) {
+			AssertUtils.assertNotNull(responseMessage, "ResponseMessage may not be null.");
+		}
+	}
+
+	public RouteStatus getStatus() {
+		return status;
+	}
+
+	public ResponseMessage getResponseMessage() {
+		return responseMessage;
+	}
+
+	
+	@Override
+	public String toString() {
+		return status.toString();
+	}
+	
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteStatus.java

@@ -0,0 +1,43 @@
+package com.nhn.pinpoint.collector.cluster.route;
+
+public enum RouteStatus {
+
+	OK(0, "OK"),
+
+	BAD_REQUEST(400, "Bad Request"),
+
+	NOT_FOUND(404, " Target Route Agent Not Found."),
+
+	NOT_ACCEPTABLE(406, "Target Route Agent Not Acceptable Command."),
+	
+	AGENT_TIMEOUT(504, "Target Route Agent Timeout");
+
+	private final int value;
+
+	private final String reasonPhrase;
+
+	private RouteStatus(int value, String reasonPhrase) {
+		this.value = value;
+		this.reasonPhrase = reasonPhrase;
+	}
+
+	public int getValue() {
+		return value;
+	}
+
+	public String getReasonPhrase() {
+		return reasonPhrase;
+	}
+	
+	@Override
+	public String toString() {
+        final StringBuilder sb = new StringBuilder();
+        sb.append(this.getClass().getSimpleName());
+        sb.append("{");
+        sb.append("code=").append(getValue()).append(",");
+        sb.append("message=").append(getReasonPhrase());
+        sb.append('}');
+        return sb.toString();
+	}
+
+}

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java

@@ -38,16 +38,16 @@ public class ZookeeperProfilerClusterManager implements SocketChannelStateChange
 
 	private final WorkerStateContext workerState;
 
-	private final ClusterPointRepository clusterPointRepository;
+	private final ClusterPointRepository profileCluster;
 
 	// 단순하게 하자 그냥 RUN이면 등록 FINISHED면 경우 삭제 그외 skip
 	// 만약 상태가 안맞으면(?) 보정 들어가야 하는데 leak detector 같은걸 worker내부에 둘 까도 고민중 
 	//
 	// RUN_DUPLEX에서만 생성할수 있게 해야한다.
 	// 지금은 RUN 상대방의 상태를 알수 없는 상태이기 때문에 이상황에서 등록
-	public ZookeeperProfilerClusterManager(ZookeeperClient client, String serverIdentifier, ClusterPointRepository clusterPoint) {
+	public ZookeeperProfilerClusterManager(ZookeeperClient client, String serverIdentifier, ClusterPointRepository profileCluster) {
 		this.workerState = new WorkerStateContext();
-		this.clusterPointRepository = clusterPoint;
+		this.profileCluster = profileCluster;
 		
 		this.client = client;
 		this.worker = new ZookeeperLatestJobWorker(client, serverIdentifier);
@@ -120,12 +120,12 @@ public class ZookeeperProfilerClusterManager implements SocketChannelStateChange
 				UpdateJob job = new UpdateJob(channelContext, new byte[0]);
 				worker.putJob(job);
 				
-				clusterPointRepository.addClusterPoint(new ChannelContextClusterPoint(channelContext));
+				profileCluster.addClusterPoint(new ChannelContextClusterPoint(channelContext));
 			} else if (PinpointServerSocketStateCode.isFinished(stateCode)) {
 				DeleteJob job = new DeleteJob(channelContext);
 				worker.putJob(job);
 
-				clusterPointRepository.removeClusterPoint(new ChannelContextClusterPoint(channelContext));
+				profileCluster.removeClusterPoint(new ChannelContextClusterPoint(channelContext));
 			} 
 		} else {
 			WorkerState state = this.workerState.getCurrentState();

collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperWebClusterManager.java

@@ -33,7 +33,7 @@ public class ZookeeperWebClusterManager implements Runnable {
 	private final StopTask stopTask = new StopTask();
 
 	private final ZookeeperClient client;
-	private final WebCluster webClusterPoint;
+	private final WebCluster webCluster;
 	private final String zNodePath;
 
 	private final AtomicBoolean retryMode = new AtomicBoolean(false);
@@ -49,10 +49,10 @@ public class ZookeeperWebClusterManager implements Runnable {
 	// Job이 포함되면 실행. Job성공시 이후 Job 모두 삭제
 	// 먼가 이상한 형태의 자료구조가 필요한거 같은데....
 
-	public ZookeeperWebClusterManager(ZookeeperClient client, String zookeeperClusterPath, String serverIdentifier, WebCluster clusterPoint) {
+	public ZookeeperWebClusterManager(ZookeeperClient client, String zookeeperClusterPath, String serverIdentifier, WebCluster webCluster) {
 		this.client = client;
 
-		this.webClusterPoint = clusterPoint;
+		this.webCluster = webCluster;
 		this.zNodePath = zookeeperClusterPath;
 
 		this.workerState = new WorkerStateContext();
@@ -188,19 +188,19 @@ public class ZookeeperWebClusterManager implements Runnable {
 				List<String> childNodeList = client.getChildrenNode(zNodePath, true);
 				List<InetSocketAddress> clusterAddressList = NetUtils.toInetSocketAddressLIst(childNodeList);
 
-				List<InetSocketAddress> addressList = webClusterPoint.getWebClusterList();
+				List<InetSocketAddress> addressList = webCluster.getWebClusterList();
 
 				logger.info("Handle register and remove Task. Current Address List = {}, Cluster Address List = {}", addressList, clusterAddressList);
 				
 				for (InetSocketAddress clusterAddress : clusterAddressList) {
 					if (!addressList.contains(clusterAddress)) {
-						webClusterPoint.connectPointIfAbsent(clusterAddress);
+						webCluster.connectPointIfAbsent(clusterAddress);
 					}
 				}
 
 				for (InetSocketAddress address : addressList) {
 					if (!clusterAddressList.contains(address)) {
-						webClusterPoint.disconnectPoint(address);
+						webCluster.disconnectPoint(address);
 					}
 				}
 

collector/src/test/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouterTest2.java

@@ -0,0 +1,209 @@
+package com.nhn.pinpoint.collector.cluster;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.thrift.TException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import com.nhn.pinpoint.collector.receiver.tcp.AgentProperties;
+import com.nhn.pinpoint.collector.util.CollectorUtils;
+import com.nhn.pinpoint.rpc.DefaultFuture;
+import com.nhn.pinpoint.rpc.Future;
+import com.nhn.pinpoint.rpc.PinpointSocketException;
+import com.nhn.pinpoint.rpc.ResponseMessage;
+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.server.ChannelContext;
+import com.nhn.pinpoint.rpc.server.PinpointServerSocket;
+import com.nhn.pinpoint.rpc.server.ServerMessageListener;
+import com.nhn.pinpoint.rpc.server.SocketChannel;
+import com.nhn.pinpoint.thrift.dto.command.TCommandEcho;
+import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer;
+import com.nhn.pinpoint.thrift.io.DeserializerFactory;
+import com.nhn.pinpoint.thrift.io.SerializerFactory;
+import com.nhn.pinpoint.thrift.util.SerializationUtils;
+
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration("classpath:applicationContext-test.xml")
+public class ClusterPointRouterTest2 {
+
+	private static final int DEFAULT_ACCEPTOR_SOCKET_PORT = 22215;
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	private final long currentTime = System.currentTimeMillis();
+
+	@Autowired
+	ClusterPointRouter clusterPointRouter;
+
+	@Autowired
+	private SerializerFactory commandSerializerFactory;
+
+	@Autowired
+	private DeserializerFactory commandDeserializerFactory;
+
+	@Test
+	public void profilerClusterPointtest() throws TException, InterruptedException {
+		WebCluster webCluster = null;
+		try {
+			webCluster = new WebCluster(CollectorUtils.getServerIdentifier(), clusterPointRouter);
+			
+			PinpointServerSocket pinpointServerSocket = createServerSocket("127.0.0.1", DEFAULT_ACCEPTOR_SOCKET_PORT);
+
+			InetSocketAddress address = new InetSocketAddress("127.0.0.1", DEFAULT_ACCEPTOR_SOCKET_PORT);
+			webCluster.connectPointIfAbsent(address);
+
+			// profiler쪽 clusterPoint 생성
+			SocketChannel socketChannel = mock(SocketChannel.class);
+			ClusterPoint clusterPoint = new ChannelContextClusterPoint(createChannelContext(socketChannel));
+
+			ClusterPointRepository clusterPointRepository = clusterPointRouter.getTargetClusterPointRepository();
+			clusterPointRepository.addClusterPoint(clusterPoint);
+
+			byte[] echoPayload = createEchoPayload("hello");
+			when(socketChannel.sendRequestMessage(echoPayload)).thenReturn(createExpectedFuture(echoPayload));
+
+			byte[] commandDeliveryPayload = createDeliveryCommandPayload("application", "agent", currentTime, echoPayload);
+
+			List<ChannelContext> contextList = pinpointServerSocket.getDuplexCommunicationChannelContext();
+			ChannelContext context = contextList.get(0);
+			Future<ResponseMessage> future = context.getSocketChannel().sendRequestMessage(commandDeliveryPayload);
+			future.await();
+
+			TCommandEcho base = (TCommandEcho) SerializationUtils.deserialize(future.getResult().getMessage(), commandDeserializerFactory);
+			
+			Assert.assertEquals(base.getMessage(), "hello");
+		} finally {
+			if (webCluster != null) {
+				webCluster.close();
+			}
+		}
+	}
+	
+	private PinpointServerSocket createServerSocket(String host, int port) {
+		PinpointServerSocket pinpointServerSocket = new PinpointServerSocket();
+		pinpointServerSocket.setMessageListener(new PinpointSocketManagerHandler());
+		pinpointServerSocket.bind(host, port);
+
+		
+		return pinpointServerSocket;
+	}
+	
+	private ChannelContext createChannelContext(SocketChannel socketChannel) {
+		ChannelContext channelContext = new ChannelContext(socketChannel, null);
+		channelContext.setChannelProperties(getParams());
+
+		return channelContext;
+	}
+	
+	private DefaultFuture createExpectedFuture(byte[] payload) {
+		ResponseMessage responseMessage = new ResponseMessage();
+		responseMessage.setMessage(payload);
+
+		DefaultFuture future = new DefaultFuture();
+		future.setResult(responseMessage);
+
+		return future;
+	}
+
+	private byte[] createEchoPayload(String message) throws TException {
+		TCommandEcho echo = new TCommandEcho();
+		echo.setMessage("hello");
+
+		byte[] payload = SerializationUtils.serialize(echo, commandSerializerFactory);
+		return payload;
+	}
+	
+	private byte[] createDeliveryCommandPayload(String application, String agent, long currentTime, byte[] echoPayload) throws TException {
+		TCommandTransfer commandTransfer = new TCommandTransfer();
+		commandTransfer.setApplicationName("application");
+		commandTransfer.setAgentId("agent");
+		commandTransfer.setStartTime(currentTime);
+		commandTransfer.setPayload(echoPayload);
+
+		byte[] payload = SerializationUtils.serialize(commandTransfer, commandSerializerFactory);
+		return payload;
+	}
+
+	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 int handleEnableWorker(Map properties) {
+			logger.warn("do handleEnableWorker {}", properties);
+			return ControlEnableWorkerConfirmPacket.SUCCESS;
+		}
+	}
+
+	private Map<Object, Object> getParams() {
+		Map<Object, Object> properties = new HashMap<Object, Object>();
+
+		properties.put(AgentProperties.KEY_AGENTID, "agent");
+		properties.put(AgentProperties.KEY_APPLICATION_NAME, "application");
+		properties.put(AgentProperties.KEY_HOSTNAME, "hostname");
+		properties.put(AgentProperties.KEY_IP, "ip");
+		properties.put(AgentProperties.KEY_PID, 1111);
+		properties.put(AgentProperties.KEY_SERVICE_TYPE, 10);
+		properties.put(AgentProperties.KEY_START_TIME_MILLIS, currentTime);
+		properties.put(AgentProperties.KEY_VERSION, "1.0.3-SNAPSHOT");
+
+		return properties;
+	}
+
+	private TargetClusterPoint findClusterPoint(String applicationName, String agentId, long startTimeStamp, List<TargetClusterPoint> targetClusterPointList) {
+
+		List<TargetClusterPoint> result = new ArrayList<TargetClusterPoint>();
+
+		for (TargetClusterPoint targetClusterPoint : targetClusterPointList) {
+			if (!targetClusterPoint.getApplicationName().equals(applicationName)) {
+				continue;
+			}
+
+			if (!targetClusterPoint.getAgentId().equals(agentId)) {
+				continue;
+			}
+
+			if (!(targetClusterPoint.getStartTimeStamp() == startTimeStamp)) {
+				continue;
+			}
+
+			result.add(targetClusterPoint);
+		}
+
+		if (result.size() == 1) {
+			return result.get(0);
+		}
+
+		if (result.size() > 1) {
+			logger.warn("Ambiguous ClusterPoint {}, {}, {} (Valid Agent list={}).", applicationName, agentId, startTimeStamp, result);
+			return null;
+		}
+
+		return null;
+	}
+
+}