Change the cluster api management method. #2262

Change the cluster api management method.
pinpoint-apm · Dec 12, 2016 · 0fd04e0 · 0fd04e0
1 parent 099fe66
commit 0fd04e0
Show file tree

Hide file tree

Showing 28 changed files with 334 additions and 346 deletions.
diff --git a/...or/src/main/java/com/navercorp/pinpoint/collector/cluster/PinpointServerClusterPoint.java b/...or/src/main/java/com/navercorp/pinpoint/collector/cluster/PinpointServerClusterPoint.java
@@ -16,15 +16,20 @@
 
 package com.navercorp.pinpoint.collector.cluster;
 
-import java.util.Map;
-
-import org.apache.commons.lang3.StringUtils;
-
-import com.navercorp.pinpoint.collector.receiver.tcp.AgentHandshakePropertyType;
 import com.navercorp.pinpoint.rpc.Future;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.rpc.util.AssertUtils;
 import com.navercorp.pinpoint.rpc.util.MapUtils;
+import com.navercorp.pinpoint.thrift.io.TCommandType;
+import com.navercorp.pinpoint.thrift.io.TCommandTypeVersion;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.thrift.TBase;
+import org.springframework.util.NumberUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
 
 /**
  * @author koo.taejin
@@ -38,22 +43,36 @@ public class PinpointServerClusterPoint implements TargetClusterPoint {
     private final long startTimeStamp;
 
     private final String version;
+    private final List<TCommandType> supportCommandList;
 
     public PinpointServerClusterPoint(PinpointServer pinpointServer) {
         AssertUtils.assertNotNull(pinpointServer, "pinpointServer may not be null.");
         this.pinpointServer = pinpointServer;
 
         Map<Object, Object> properties = pinpointServer.getChannelProperties();
-        this.version = MapUtils.getString(properties, AgentHandshakePropertyType.VERSION.getName());
+        this.version = MapUtils.getString(properties, HandshakePropertyType.VERSION.getName());
         AssertUtils.assertTrue(!StringUtils.isBlank(version), "Version may not be null or empty.");
 
-        this.applicationName = MapUtils.getString(properties, AgentHandshakePropertyType.APPLICATION_NAME.getName());
+        this.supportCommandList = new ArrayList<>();
+        Object supportCommandCodeList = properties.get(HandshakePropertyType.SUPPORT_COMMAND_LIST.getName());
+        if (supportCommandCodeList instanceof List) {
+            for (Object supportCommandCode : (List)supportCommandCodeList) {
+                if (supportCommandCode instanceof Number) {
+                    TCommandType commandType = TCommandType.getType(NumberUtils.convertNumberToTargetClass((Number) supportCommandCode, Short.class));
+                    if (commandType != null) {
+                        supportCommandList.add(commandType);
+                    }
+                }
+            }
+        }
+
+        this.applicationName = MapUtils.getString(properties, HandshakePropertyType.APPLICATION_NAME.getName());
         AssertUtils.assertTrue(!StringUtils.isBlank(applicationName), "ApplicationName may not be null or empty.");
 
-        this.agentId = MapUtils.getString(properties, AgentHandshakePropertyType.AGENT_ID.getName());
+        this.agentId = MapUtils.getString(properties, HandshakePropertyType.AGENT_ID.getName());
         AssertUtils.assertTrue(!StringUtils.isBlank(agentId), "AgentId may not be null or empty.");
 
-        this.startTimeStamp = MapUtils.getLong(properties, AgentHandshakePropertyType.START_TIMESTAMP.getName());
+        this.startTimeStamp = MapUtils.getLong(properties, HandshakePropertyType.START_TIMESTAMP.getName());
         AssertUtils.assertTrue(startTimeStamp > 0, "StartTimeStamp is must greater than zero.");
     }
 
@@ -86,6 +105,22 @@ public String gerVersion() {
         return version;
     }
 
+    @Override
+    public boolean isSupportCommand(TBase command) {
+        for (TCommandType supportCommand : supportCommandList) {
+            if (supportCommand.getClazz() == command.getClass()) {
+                return true;
+            }
+        }
+
+        TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(version);
+        if (commandVersion.isSupportCommand(command)) {
+            return true;
+        }
+
+        return false;
+    }
+
     public PinpointServer getPinpointServer() {
         return pinpointServer;
     }
@@ -103,6 +138,8 @@ public String toString() {
         log.append(")");
         log.append(", version:");
         log.append(version);
+        log.append(", supportCommandList:");
+        log.append(supportCommandList);
         log.append(", pinpointServer:");
         log.append(pinpointServer);
 

diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/TargetClusterPoint.java b/collector/src/main/java/com/navercorp/pinpoint/collector/cluster/TargetClusterPoint.java
@@ -16,6 +16,8 @@
 
 package com.navercorp.pinpoint.collector.cluster;
 
+import org.apache.thrift.TBase;
+
 public interface TargetClusterPoint extends ClusterPoint {
 
     String getApplicationName();
@@ -26,4 +28,6 @@ public interface TargetClusterPoint extends ClusterPoint {
 
     String gerVersion();
 
+    boolean isSupportCommand(TBase command);
+
 }
diff --git a/...tor/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteHandler.java b/...tor/src/main/java/com/navercorp/pinpoint/collector/cluster/route/DefaultRouteHandler.java
@@ -23,7 +23,6 @@
 import com.navercorp.pinpoint.rpc.ResponseMessage;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransferResponse;
 import com.navercorp.pinpoint.thrift.dto.command.TRouteResult;
-import com.navercorp.pinpoint.thrift.io.TCommandTypeVersion;
 import org.apache.thrift.TBase;
 
 /**
@@ -76,8 +75,7 @@ private TCommandTransferResponse onRoute0(RequestEvent event) {
             return createResponse(TRouteResult.NOT_FOUND);
         }
 
-        TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(clusterPoint.gerVersion());
-        if (!commandVersion.isSupportCommand(requestObject)) {
+        if (!clusterPoint.isSupportCommand(requestObject)) {
             return createResponse(TRouteResult.NOT_SUPPORTED_REQUEST);
         }
 

diff --git a/...ctor/src/main/java/com/navercorp/pinpoint/collector/cluster/route/StreamRouteHandler.java b/...ctor/src/main/java/com/navercorp/pinpoint/collector/cluster/route/StreamRouteHandler.java
@@ -24,12 +24,17 @@
 import com.navercorp.pinpoint.rpc.packet.stream.StreamClosePacket;
 import com.navercorp.pinpoint.rpc.packet.stream.StreamResponsePacket;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
-import com.navercorp.pinpoint.rpc.stream.*;
+import com.navercorp.pinpoint.rpc.stream.ClientStreamChannel;
+import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelContext;
+import com.navercorp.pinpoint.rpc.stream.ClientStreamChannelMessageListener;
+import com.navercorp.pinpoint.rpc.stream.ServerStreamChannel;
+import com.navercorp.pinpoint.rpc.stream.ServerStreamChannelContext;
+import com.navercorp.pinpoint.rpc.stream.StreamChannelStateChangeEventHandler;
+import com.navercorp.pinpoint.rpc.stream.StreamChannelStateCode;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransferResponse;
 import com.navercorp.pinpoint.thrift.dto.command.TRouteResult;
 import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializer;
 import com.navercorp.pinpoint.thrift.io.SerializerFactory;
-import com.navercorp.pinpoint.thrift.io.TCommandTypeVersion;
 import com.navercorp.pinpoint.thrift.util.SerializationUtils;
 import org.apache.thrift.TBase;
 import org.slf4j.Logger;
@@ -96,8 +101,7 @@ private TCommandTransferResponse onRoute0(StreamEvent event) {
             return createResponse(TRouteResult.NOT_FOUND);
         }
 
-        TCommandTypeVersion commandVersion = TCommandTypeVersion.getVersion(clusterPoint.gerVersion());
-        if (!commandVersion.isSupportCommand(requestObject)) {
+        if (!clusterPoint.isSupportCommand(requestObject)) {
             return createResponse(TRouteResult.NOT_SUPPORTED_REQUEST);
         }
 

diff --git a/...ain/java/com/navercorp/pinpoint/collector/cluster/zookeeper/PinpointServerRepository.java b/...ain/java/com/navercorp/pinpoint/collector/cluster/zookeeper/PinpointServerRepository.java
@@ -16,7 +16,7 @@
 
 package com.navercorp.pinpoint.collector.cluster.zookeeper;
 
-import com.navercorp.pinpoint.collector.receiver.tcp.AgentHandshakePropertyType;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.rpc.util.MapUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -88,9 +88,9 @@ public List<PinpointServer> getValues() {
 
     private String getKey(PinpointServer pinpointServer) {
         Map<Object, Object> properties = pinpointServer.getChannelProperties();
-        final String applicationName = MapUtils.getString(properties, AgentHandshakePropertyType.APPLICATION_NAME.getName());
-        final String agentId = MapUtils.getString(properties, AgentHandshakePropertyType.AGENT_ID.getName());
-        final Long startTimeStamp = MapUtils.getLong(properties, AgentHandshakePropertyType.START_TIMESTAMP.getName());
+        final String applicationName = MapUtils.getString(properties, HandshakePropertyType.APPLICATION_NAME.getName());
+        final String agentId = MapUtils.getString(properties, HandshakePropertyType.AGENT_ID.getName());
+        final Long startTimeStamp = MapUtils.getLong(properties, HandshakePropertyType.START_TIMESTAMP.getName());
 
         if (StringUtils.isBlank(applicationName) || StringUtils.isBlank(agentId) || startTimeStamp == null || startTimeStamp <= 0) {
             return StringUtils.EMPTY;

diff --git a/.../src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperJobWorker.java b/.../src/main/java/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperJobWorker.java
@@ -17,9 +17,9 @@
 package com.navercorp.pinpoint.collector.cluster.zookeeper;
 
 import com.navercorp.pinpoint.collector.cluster.zookeeper.job.ZookeeperJob;
-import com.navercorp.pinpoint.collector.receiver.tcp.AgentHandshakePropertyType;
 import com.navercorp.pinpoint.common.server.util.concurrent.CommonStateContext;
 import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.rpc.util.ClassUtils;
 import com.navercorp.pinpoint.rpc.util.MapUtils;
@@ -371,9 +371,9 @@ private String removeIfExistContents(String contents, String removeContents) {
 
     private String getKey(PinpointServer pinpointServer) {
         Map<Object, Object> properties = pinpointServer.getChannelProperties();
-        final String applicationName = MapUtils.getString(properties, AgentHandshakePropertyType.APPLICATION_NAME.getName());
-        final String agentId = MapUtils.getString(properties, AgentHandshakePropertyType.AGENT_ID.getName());
-        final Long startTimeStamp = MapUtils.getLong(properties, AgentHandshakePropertyType.START_TIMESTAMP.getName());
+        final String applicationName = MapUtils.getString(properties, HandshakePropertyType.APPLICATION_NAME.getName());
+        final String agentId = MapUtils.getString(properties, HandshakePropertyType.AGENT_ID.getName());
+        final Long startTimeStamp = MapUtils.getLong(properties, HandshakePropertyType.START_TIMESTAMP.getName());
 
         if (StringUtils.isBlank(applicationName) || StringUtils.isBlank(agentId) || startTimeStamp == null || startTimeStamp <= 0) {
             return StringUtils.EMPTY;

diff --git a/...a/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java b/...a/com/navercorp/pinpoint/collector/cluster/zookeeper/ZookeeperProfilerClusterManager.java
@@ -19,8 +19,8 @@
 import com.navercorp.pinpoint.collector.cluster.ClusterPointRepository;
 import com.navercorp.pinpoint.collector.cluster.PinpointServerClusterPoint;
 import com.navercorp.pinpoint.common.server.util.concurrent.CommonStateContext;
-import com.navercorp.pinpoint.collector.receiver.tcp.AgentHandshakePropertyType;
 import com.navercorp.pinpoint.rpc.common.SocketStateCode;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.rpc.server.handler.ServerStateChangeEventHandler;
 import com.navercorp.pinpoint.rpc.util.MapUtils;
@@ -172,8 +172,8 @@ public void initZookeeperClusterData() {
     }
 
     private boolean skipAgent(Map<Object, Object> agentProperties) {
-        String applicationName = MapUtils.getString(agentProperties, AgentHandshakePropertyType.APPLICATION_NAME.getName());
-        String agentId = MapUtils.getString(agentProperties, AgentHandshakePropertyType.AGENT_ID.getName());
+        String applicationName = MapUtils.getString(agentProperties, HandshakePropertyType.APPLICATION_NAME.getName());
+        String agentId = MapUtils.getString(agentProperties, HandshakePropertyType.AGENT_ID.getName());
 
         if (StringUtils.isBlank(applicationName) || StringUtils.isBlank(agentId)) {
             return true;

diff --git a/...c/main/java/com/navercorp/pinpoint/collector/receiver/tcp/AgentHandshakePropertyType.java b/...c/main/java/com/navercorp/pinpoint/collector/receiver/tcp/AgentHandshakePropertyType.java
diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/tcp/TCPReceiver.java b/collector/src/main/java/com/navercorp/pinpoint/collector/receiver/tcp/TCPReceiver.java
@@ -29,6 +29,7 @@
 import com.navercorp.pinpoint.common.util.ExecutorFactory;
 import com.navercorp.pinpoint.common.util.PinpointThreadFactory;
 import com.navercorp.pinpoint.rpc.PinpointSocket;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.packet.HandshakeResponseCode;
 import com.navercorp.pinpoint.rpc.packet.HandshakeResponseType;
 import com.navercorp.pinpoint.rpc.packet.PingPacket;
@@ -186,12 +187,12 @@ public HandshakeResponseCode handleHandshake(Map properties) {
                     return HandshakeResponseType.ProtocolError.PROTOCOL_ERROR;
                 }
 
-                boolean hasAllType = AgentHandshakePropertyType.hasAllType(properties);
-                if (!hasAllType) {
+                boolean hasRequiredKeys = HandshakePropertyType.hasRequiredKeys(properties);
+                if (!hasRequiredKeys) {
                     return HandshakeResponseType.PropertyError.PROPERTY_ERROR;
                 }
 
-                boolean supportServer = MapUtils.getBoolean(properties, AgentHandshakePropertyType.SUPPORT_SERVER.getName(), true);
+                boolean supportServer = MapUtils.getBoolean(properties, HandshakePropertyType.SUPPORT_SERVER.getName(), true);
                 if (supportServer) {
                     return HandshakeResponseType.Success.DUPLEX_COMMUNICATION;
                 } else {

diff --git a/collector/src/main/java/com/navercorp/pinpoint/collector/rpc/handler/AgentEventHandler.java b/collector/src/main/java/com/navercorp/pinpoint/collector/rpc/handler/AgentEventHandler.java
@@ -16,31 +16,29 @@
 
 package com.navercorp.pinpoint.collector.rpc.handler;
 
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.Executor;
-
-import javax.annotation.Resource;
-
-import org.apache.commons.collections.MapUtils;
-import org.apache.thrift.TException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import com.navercorp.pinpoint.collector.cluster.route.ResponseEvent;
 import com.navercorp.pinpoint.collector.dao.AgentEventDao;
-import com.navercorp.pinpoint.collector.receiver.tcp.AgentHandshakePropertyType;
 import com.navercorp.pinpoint.common.server.bo.AgentEventBo;
 import com.navercorp.pinpoint.common.server.util.AgentEventMessageSerializer;
 import com.navercorp.pinpoint.common.server.util.AgentEventType;
 import com.navercorp.pinpoint.common.server.util.AgentEventTypeCategory;
+import com.navercorp.pinpoint.rpc.packet.HandshakePropertyType;
 import com.navercorp.pinpoint.rpc.server.PinpointServer;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransfer;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransferResponse;
 import com.navercorp.pinpoint.thrift.dto.command.TRouteResult;
 import com.navercorp.pinpoint.thrift.io.DeserializerFactory;
 import com.navercorp.pinpoint.thrift.io.HeaderTBaseDeserializer;
 import com.navercorp.pinpoint.thrift.util.SerializationUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Resource;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executor;
 
 /**
  * @author HyunGil Jeong
@@ -79,9 +77,9 @@ public void handleEvent(PinpointServer pinpointServer, long eventTimestamp, Agen
 
         Map<Object, Object> channelProperties = pinpointServer.getChannelProperties();
 
-        final String agentId = MapUtils.getString(channelProperties, AgentHandshakePropertyType.AGENT_ID.getName());
+        final String agentId = MapUtils.getString(channelProperties, HandshakePropertyType.AGENT_ID.getName());
         final long startTimestamp = MapUtils.getLong(channelProperties,
-                AgentHandshakePropertyType.START_TIMESTAMP.getName());
+                HandshakePropertyType.START_TIMESTAMP.getName());
 
         this.executor.execute(new AgentEventHandlerDispatch(agentId, startTimestamp, eventTimestamp, eventType,
                 eventMessage));