diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ChannelContextClusterPoint.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ChannelContextClusterPoint.java index 8ecded808..2846ecd2c 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ChannelContextClusterPoint.java +++ b/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; } diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouter.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouter.java index 8d4b71c5b..966f77ea1 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouter.java +++ b/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 targetClusterPointRepository; + private final RouteHandler routeHandler; + @Autowired private SerializerFactory commandSerializerFactory; @@ -44,6 +45,11 @@ public class ClusterPointRouter implements MessageListener { public ClusterPointRouter() { this.targetClusterPointRepository = new ClusterPointRepository(); + 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 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 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 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 targetClusterPointList) { - - List result = new ArrayList(); - - 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; - } - } diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteEvent.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteEvent.java new file mode 100644 index 000000000..b92c51db9 --- /dev/null +++ b/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(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteFilterChain.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteFilterChain.java new file mode 100644 index 000000000..2977499bd --- /dev/null +++ b/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 implements RouteFilterChain { + + private final Logger logger = LoggerFactory.getLogger(this.getClass()); + + private final CopyOnWriteArrayList> filterList = new CopyOnWriteArrayList>(); + + @Override + public void addLast(RouteFilter filter) { + filterList.add(filter); + } + + @Override + public void doEvent(T event) { + for (RouteFilter filter : filterList) { + try { + filter.doEvent(event); + } catch (Exception e) { + if (logger.isWarnEnabled()) { + logger.warn(filter.getClass().getSimpleName() + " filter occured exception. caused=" + e.getMessage() + ".", e); + } + } + } + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteHandler.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteHandler.java new file mode 100644 index 000000000..acb1abf3a --- /dev/null +++ b/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 requestFilterChain; + private final RouteFilterChain responseFilterChain; + + private final ClusterPointLocator targetClusterPointLocator; + + public DefaultRouteHandler(ClusterPointLocator targetClusterPointLocator) { + this.targetClusterPointLocator = targetClusterPointLocator; + + this.requestFilterChain = new DefaultRouteFilterChain(); + this.responseFilterChain = new DefaultRouteFilterChain(); + } + + @Override + public void addRequestFilter(RouteFilter filter) { + this.requestFilterChain.addLast(filter); + } + + @Override + public void addResponseFilter(RouteFilter 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 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 result = new ArrayList(); + + 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; + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/LoggingFilter.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/LoggingFilter.java new file mode 100644 index 000000000..b4451efe2 --- /dev/null +++ b/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 { + + @Override + public void doEvent(RequestEvent event) { + logger.warn("{} doEvent {}.", this.getClass().getSimpleName(), event); + } + + } + + class ResponseFilter implements RouteFilter { + + @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(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RequestEvent.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RequestEvent.java new file mode 100644 index 000000000..1ce28091e --- /dev/null +++ b/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(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/ResponseEvent.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/ResponseEvent.java new file mode 100644 index 000000000..31ae1c453 --- /dev/null +++ b/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(); + } + + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteEvent.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteEvent.java new file mode 100644 index 000000000..145847807 --- /dev/null +++ b/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(); + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilter.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilter.java new file mode 100644 index 000000000..5310ddf15 --- /dev/null +++ b/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 { + + void doEvent(T event); + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilterChain.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteFilterChain.java new file mode 100644 index 000000000..afa00afa8 --- /dev/null +++ b/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 { + + void addLast(RouteFilter filter); + + void doEvent(T event); + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteHandler.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteHandler.java new file mode 100644 index 000000000..3c60f7058 --- /dev/null +++ b/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 filter); + + void addResponseFilter(RouteFilter filter); + + RouteResult onRoute(RequestEvent event); + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteResult.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteResult.java new file mode 100644 index 000000000..00b5faa61 --- /dev/null +++ b/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(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteStatus.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/route/RouteStatus.java new file mode 100644 index 000000000..93a8a90ac --- /dev/null +++ b/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(); + } + +} diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java index 251e17a85..9cf9f92bf 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java +++ b/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(); diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperWebClusterManager.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperWebClusterManager.java index 53d516fbb..65eedb72e 100644 --- a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperWebClusterManager.java +++ b/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 childNodeList = client.getChildrenNode(zNodePath, true); List clusterAddressList = NetUtils.toInetSocketAddressLIst(childNodeList); - List addressList = webClusterPoint.getWebClusterList(); + List 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); } } diff --git a/collector/src/test/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouterTest2.java b/collector/src/test/java/com/navercorp/pinpoint/collector/cluster/ClusterPointRouterTest2.java new file mode 100644 index 000000000..b3f6cee0e --- /dev/null +++ b/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 contextList = pinpointServerSocket.getDuplexCommunicationChannelContext(); + ChannelContext context = contextList.get(0); + Future 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 getParams() { + Map properties = new HashMap(); + + 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 targetClusterPointList) { + + List result = new ArrayList(); + + 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; + } + +}