diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java index a23007abc..72a84a8ab 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java @@ -30,6 +30,8 @@ import com.nhn.pinpoint.profiler.interceptor.bci.JavaAssistByteCodeInstrumentor; import com.nhn.pinpoint.profiler.logging.Slf4jLoggerBinder; import com.nhn.pinpoint.profiler.monitor.AgentStatMonitor; import com.nhn.pinpoint.profiler.receiver.CommandDispatcher; +import com.nhn.pinpoint.profiler.receiver.service.EchoService; +import com.nhn.pinpoint.profiler.receiver.service.ThreadDumpService; import com.nhn.pinpoint.profiler.sampler.SamplerFactory; import com.nhn.pinpoint.profiler.sender.BufferedUdpDataSender; import com.nhn.pinpoint.profiler.sender.DataSender; @@ -119,7 +121,11 @@ public class DefaultAgent implements Agent { this.agentInformation = agentInformationFactory.createAgentInformation(typeResolver.getServerType()); logger.info("agentInformation:{}", agentInformation); - this.factory = createPinpointSocketFactory(this.profilerConfig.isTcpDataSenderCommandAcceptEnable()); + CommandDispatcher commandDispatcher = new CommandDispatcher(); + commandDispatcher.registerCommandService(new ThreadDumpService()); + commandDispatcher.registerCommandService(new EchoService()); + + this.factory = createPinpointSocketFactory(commandDispatcher); this.socket = createPinpointSocket(this.profilerConfig.getCollectorTcpServerIp(), this.profilerConfig.getCollectorTcpServerPort(), factory); this.serverMetaDataHolder = createServerMetaDataHolder(); @@ -247,15 +253,17 @@ public class DefaultAgent implements Agent { return serverMetaDataHolder; } - protected PinpointSocketFactory createPinpointSocketFactory(boolean isSupportServerMode) { - + protected PinpointSocketFactory createPinpointSocketFactory(CommandDispatcher commandDispatcher) { PinpointSocketFactory pinpointSocketFactory = new PinpointSocketFactory(); pinpointSocketFactory.setTimeoutMillis(1000 * 5); Map properties = this.agentInformation.toMap(); + + boolean isSupportServerMode = this.profilerConfig.isTcpDataSenderCommandAcceptEnable(); + if (isSupportServerMode) { - CommandDispatcher.Builder builder = new CommandDispatcher.Builder(); - pinpointSocketFactory.setMessageListener(builder.build()); + pinpointSocketFactory.setMessageListener(commandDispatcher); + pinpointSocketFactory.setServerStreamChannelMessageListener(commandDispatcher); properties.put(AgentHandshakePropertyType.SUPPORT_SERVER.getName(), true); } else { diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/CommandDispatcher.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/CommandDispatcher.java index 5c157f16e..f19112e9f 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/CommandDispatcher.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/CommandDispatcher.java @@ -1,8 +1,5 @@ package com.nhn.pinpoint.profiler.receiver; -import java.util.ArrayList; -import java.util.List; - import org.apache.thrift.TBase; import org.apache.thrift.protocol.TCompactProtocol; import org.apache.thrift.protocol.TProtocolFactory; @@ -11,12 +8,15 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.nhn.pinpoint.common.Version; -import com.nhn.pinpoint.profiler.receiver.service.EchoService; -import com.nhn.pinpoint.profiler.receiver.service.ThreadDumpService; import com.nhn.pinpoint.rpc.client.MessageListener; import com.nhn.pinpoint.rpc.packet.RequestPacket; import com.nhn.pinpoint.rpc.packet.ResponsePacket; import com.nhn.pinpoint.rpc.packet.SendPacket; +import com.nhn.pinpoint.rpc.packet.stream.StreamClosePacket; +import com.nhn.pinpoint.rpc.packet.stream.StreamCreateFailPacket; +import com.nhn.pinpoint.rpc.packet.stream.StreamCreatePacket; +import com.nhn.pinpoint.rpc.stream.ServerStreamChannelContext; +import com.nhn.pinpoint.rpc.stream.ServerStreamChannelMessageListener; import com.nhn.pinpoint.rpc.util.AssertUtils; import com.nhn.pinpoint.thrift.dto.TResult; import com.nhn.pinpoint.thrift.io.DeserializerFactory; @@ -25,121 +25,108 @@ import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializerFactory; import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializer; import com.nhn.pinpoint.thrift.io.HeaderTBaseSerializerFactory; import com.nhn.pinpoint.thrift.io.SerializerFactory; -import com.nhn.pinpoint.thrift.io.TBaseLocator; import com.nhn.pinpoint.thrift.io.TCommandRegistry; import com.nhn.pinpoint.thrift.io.TCommandTypeVersion; import com.nhn.pinpoint.thrift.io.ThreadLocalHeaderTBaseDeserializerFactory; import com.nhn.pinpoint.thrift.io.ThreadLocalHeaderTBaseSerializerFactory; import com.nhn.pinpoint.thrift.util.SerializationUtils; -public class CommandDispatcher implements MessageListener { +public class CommandDispatcher implements MessageListener, ServerStreamChannelMessageListener { - private final Logger logger = LoggerFactory.getLogger(this.getClass()); + private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final ProfilerCommandServiceLocator locator; + private final ProfilerCommandServiceRegistry commandServiceRegistry = new ProfilerCommandServiceRegistry(); + + private final SerializerFactory serializerFactory; + private final DeserializerFactory deserializerFactory; - private final SerializerFactory serializerFactory; - private final DeserializerFactory deserializerFactory; + public CommandDispatcher() { + this(Version.VERSION); + } - public CommandDispatcher(Builder builder) { - ProfilerCommandServiceRegistry registry = new ProfilerCommandServiceRegistry(); - for (ProfilerCommandService service : builder.serviceList) { - registry.addService(service); - } - this.locator = registry; - - SerializerFactory serializerFactory = new HeaderTBaseSerializerFactory(true, builder.serializationMaxSize, builder.protocolFactory, builder.commandTbaseLocator); - this.serializerFactory = wrappedThreadLocalSerializerFactory(serializerFactory); - AssertUtils.assertNotNull(this.serializerFactory); - - DeserializerFactory deserializerFactory = new HeaderTBaseDeserializerFactory(builder.protocolFactory, builder.commandTbaseLocator); - this.deserializerFactory = wrappedThreadLocalDeserializerFactory(deserializerFactory); - AssertUtils.assertNotNull(this.deserializerFactory); - } + public CommandDispatcher(String pinpointVersion) { + this(pinpointVersion, HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE); + } - private SerializerFactory wrappedThreadLocalSerializerFactory(SerializerFactory serializerFactory) { - return new ThreadLocalHeaderTBaseSerializerFactory(serializerFactory); - } - - private DeserializerFactory wrappedThreadLocalDeserializerFactory(DeserializerFactory deserializerFactory) { - return new ThreadLocalHeaderTBaseDeserializerFactory(deserializerFactory); - } - - @Override - public void handleSend(SendPacket sendPacket, Channel channel) { - logger.info("MessageReceive {} {}", sendPacket, channel); - } + public CommandDispatcher(String pinpointVersion, int serializationMaxSize) { + TProtocolFactory protocolFactory = new TCompactProtocol.Factory(); + TCommandRegistry commandTbaseRegistry = new TCommandRegistry(TCommandTypeVersion.getVersion(pinpointVersion)); + + SerializerFactory serializerFactory = new HeaderTBaseSerializerFactory(true, serializationMaxSize, protocolFactory, commandTbaseRegistry); + this.serializerFactory = wrappedThreadLocalSerializerFactory(serializerFactory); + AssertUtils.assertNotNull(this.serializerFactory); - - @Override - public void handleRequest(RequestPacket requestPacket, Channel channel) { - logger.info("MessageReceive {} {}", requestPacket, channel); + DeserializerFactory deserializerFactory = new HeaderTBaseDeserializerFactory(protocolFactory, commandTbaseRegistry); + this.deserializerFactory = wrappedThreadLocalDeserializerFactory(deserializerFactory); + AssertUtils.assertNotNull(this.deserializerFactory); + } - TBase request = SerializationUtils.deserialize(requestPacket.getPayload(), deserializerFactory, null); - - TBase response = null; - if (request == null) { - TResult tResult = new TResult(false); - tResult.setMessage("Unsupported Type."); - - response = tResult; - } else { - ProfilerRequestCommandService service = locator.getRequestService(request); - - if (service == null) { - TResult tResult = new TResult(false); - tResult.setMessage("Unsupported Listener."); + @Override + public void handleSend(SendPacket sendPacket, Channel channel) { + logger.info("MessageReceive {} {}", sendPacket, channel); + } - response = tResult; - } else { - response = service.requestCommandService(request); - } - } - - byte[] payload = SerializationUtils.serialize(response, serializerFactory, null); - - if (payload != null) { - channel.write(new ResponsePacket(requestPacket.getRequestId(), payload)); - } - } + @Override + public void handleRequest(RequestPacket requestPacket, Channel channel) { + logger.info("MessageReceive {} {}", requestPacket, channel); - public static class Builder { - private List serviceList = new ArrayList(); + TBase request = SerializationUtils.deserialize(requestPacket.getPayload(), deserializerFactory, null); + + TBase response = null; + if (request == null) { + TResult tResult = new TResult(false); + tResult.setMessage("Unsupported Type."); + + response = tResult; + } else { + ProfilerRequestCommandService service = commandServiceRegistry.getRequestService(request); + + if (service == null) { + TResult tResult = new TResult(false); + tResult.setMessage("Unsupported Listener."); - private int serializationMaxSize = HeaderTBaseSerializerFactory.DEFAULT_UDP_STREAM_MAX_SIZE; - private TProtocolFactory protocolFactory = new TCompactProtocol.Factory(); - private TBaseLocator commandTbaseLocator = new TCommandRegistry(TCommandTypeVersion.getVersion(Version.VERSION)); - - public Builder() { - serviceList.add(new ThreadDumpService()); - serviceList.add(new EchoService()); - } - - public void addService(ProfilerCommandService service) { - serviceList.add(service); - } + response = tResult; + } else { + response = service.requestCommandService(request); + } + } + + byte[] payload = SerializationUtils.serialize(response, serializerFactory, null); + if (payload != null) { + channel.write(new ResponsePacket(requestPacket.getRequestId(), payload)); + } + } - public void setProtocolFactory(TProtocolFactory protocolFactory) { - this.protocolFactory = protocolFactory; - } + @Override + public short handleStreamCreate(ServerStreamChannelContext streamChannelContext, StreamCreatePacket packet) { + logger.info("MessageReceived handleStreamCreate {} {}", packet, streamChannelContext); - public void setCommandTbaseLocator(TBaseLocator commandTbaseLocator) { - this.commandTbaseLocator = commandTbaseLocator; - } + TBase request = SerializationUtils.deserialize(packet.getPayload(), deserializerFactory, null); + + ProfilerStreamCommandService service = commandServiceRegistry.getStreamService(request); + if (service == null) { + return StreamCreateFailPacket.PACKET_UNSUPPORT; + } + + service.streamCommandService(request, streamChannelContext); + + return StreamCreatePacket.SUCCESS; + } - public void setSerializationMaxSize(int serializationMaxSize) { - this.serializationMaxSize = serializationMaxSize; - } + @Override + public void handleStreamClose(ServerStreamChannelContext streamChannelContext, StreamClosePacket packet) { + } - public CommandDispatcher build() { - AssertUtils.assertNotNull(protocolFactory, "protocolFactory may note be null."); - AssertUtils.assertNotNull(commandTbaseLocator, "commandTbaseLocator may note be null."); - AssertUtils.assertTrue(serializationMaxSize > 0, "serializationMaxSize must grater than zero."); - AssertUtils.assertTrue(serviceList.size() > 0, "serializationMaxSize must grater than zero."); - - return new CommandDispatcher(this); - } - - } + public boolean registerCommandService(ProfilerCommandService commandService) { + return this.commandServiceRegistry.addService(commandService); + } + + private SerializerFactory wrappedThreadLocalSerializerFactory(SerializerFactory serializerFactory) { + return new ThreadLocalHeaderTBaseSerializerFactory(serializerFactory); + } + + private DeserializerFactory wrappedThreadLocalDeserializerFactory(DeserializerFactory deserializerFactory) { + return new ThreadLocalHeaderTBaseDeserializerFactory(deserializerFactory); + } } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceLocator.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceLocator.java index 96369870b..29639230a 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceLocator.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceLocator.java @@ -12,5 +12,7 @@ public interface ProfilerCommandServiceLocator { ProfilerSimpleCommandService getSimpleService(TBase tBase); ProfilerRequestCommandService getRequestService(TBase tBase); + + ProfilerStreamCommandService getStreamService(TBase tBase); } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceRegistry.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceRegistry.java index 46b80d6b7..7cc190174 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceRegistry.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerCommandServiceRegistry.java @@ -1,7 +1,6 @@ package com.nhn.pinpoint.profiler.receiver; -import java.util.HashMap; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import org.apache.thrift.TBase; import org.slf4j.Logger; @@ -14,27 +13,24 @@ public class ProfilerCommandServiceRegistry implements ProfilerCommandServiceLoc private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final Map, ProfilerCommandService> profilerCommandServiceRepository; + private final ConcurrentHashMap, ProfilerCommandService> profilerCommandServiceRepository; public ProfilerCommandServiceRegistry() { - profilerCommandServiceRepository = new HashMap, ProfilerCommandService>(); + profilerCommandServiceRepository = new ConcurrentHashMap, ProfilerCommandService>(); } - - /** - * not guarantee thread safe. - */ public boolean addService(ProfilerCommandService service) { return addService(service.getCommandClazz(), service); } public boolean addService(Class clazz, ProfilerCommandService service) { - if (profilerCommandServiceRepository.containsKey(clazz)) { - logger.warn("Already Register Type({}).", clazz.getName()); - return false; - } - - profilerCommandServiceRepository.put(clazz, service); + ProfilerCommandService inValue = profilerCommandServiceRepository.putIfAbsent(clazz, service); + + if (inValue != null) { + logger.warn("Already Register Type({}).", clazz.getName()); + return false; + } + return true; } @@ -64,5 +60,16 @@ public class ProfilerCommandServiceRegistry implements ProfilerCommandServiceLoc return null; } + + @Override + public ProfilerStreamCommandService getStreamService(TBase tBase) { + ProfilerCommandService service = profilerCommandServiceRepository.get(tBase.getClass()); + + if (service instanceof ProfilerStreamCommandService) { + return (ProfilerStreamCommandService) service; + } + + return null; + } } diff --git a/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerStreamCommandService.java b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerStreamCommandService.java new file mode 100644 index 000000000..87ceed0cf --- /dev/null +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/receiver/ProfilerStreamCommandService.java @@ -0,0 +1,11 @@ +package com.nhn.pinpoint.profiler.receiver; + +import org.apache.thrift.TBase; + +import com.nhn.pinpoint.rpc.stream.ServerStreamChannelContext; + +public interface ProfilerStreamCommandService extends ProfilerCommandService { + + short streamCommandService(TBase tBase, ServerStreamChannelContext streamChannelContext); + +} diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TCommandRegistry.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TCommandRegistry.java index 0bd432add..0457c9306 100644 --- a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TCommandRegistry.java +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/TCommandRegistry.java @@ -1,9 +1,8 @@ package com.nhn.pinpoint.thrift.io; import java.util.Collection; -import java.util.HashMap; import java.util.List; -import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; import org.apache.thrift.TBase; import org.apache.thrift.TException; @@ -13,15 +12,13 @@ import org.apache.thrift.TException; */ public class TCommandRegistry implements TBaseLocator { - private final Map commandTBaseRepository; + private final ConcurrentHashMap commandTBaseRepository = new ConcurrentHashMap(); public TCommandRegistry(TCommandTypeVersion version) { this(version.getSupportCommandList()); } public TCommandRegistry(List supportCommandList) { - commandTBaseRepository = new HashMap(supportCommandList.size()); - for (TCommandType type : supportCommandList) { commandTBaseRepository.put(type.getType(), type); } @@ -30,7 +27,6 @@ public class TCommandRegistry implements TBaseLocator { @Override public TBase tBaseLookup(short type) throws TException { TCommandType commandTBaseType = commandTBaseRepository.get(type); - if (commandTBaseType == null) { throw new TException("Unsupported type:" + type); }