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 493ad7838..8d4b71c5b 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 @@ -21,6 +21,8 @@ import com.nhn.pinpoint.rpc.packet.SendPacket; import com.nhn.pinpoint.thrift.dto.TResult; import com.nhn.pinpoint.thrift.dto.command.TCommandTransfer; 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; 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 index 8fdc42dff..af9c3e847 100644 --- 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 @@ -34,7 +34,9 @@ import com.nhn.pinpoint.rpc.server.ServerMessageListener; import com.nhn.pinpoint.rpc.server.SocketChannel; import com.nhn.pinpoint.thrift.io.DeserializerFactory; import com.nhn.pinpoint.thrift.io.Header; +import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializer; 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.L4Packet; import com.nhn.pinpoint.thrift.io.SerializerFactory; diff --git a/collector/src/main/resources/applicationContext-collector.xml b/collector/src/main/resources/applicationContext-collector.xml index 78c4394c5..2166726cc 100644 --- a/collector/src/main/resources/applicationContext-collector.xml +++ b/collector/src/main/resources/applicationContext-collector.xml @@ -89,7 +89,7 @@ - + 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 447afd7ac..8f2b71421 100644 --- a/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java +++ b/profiler/src/main/java/com/navercorp/pinpoint/profiler/DefaultAgent.java @@ -36,6 +36,7 @@ import com.nhn.pinpoint.profiler.sender.BufferedUdpDataSender; import com.nhn.pinpoint.profiler.sender.DataSender; import com.nhn.pinpoint.profiler.sender.EnhancedDataSender; import com.nhn.pinpoint.profiler.sender.TcpDataSender; +import com.nhn.pinpoint.profiler.sender.UdpDataSender; import com.nhn.pinpoint.profiler.util.ApplicationServerTypeResolver; import com.nhn.pinpoint.profiler.util.PreparedStatementUtils; import com.nhn.pinpoint.profiler.util.RuntimeMXBeanUtils; @@ -127,12 +128,12 @@ public class DefaultAgent implements Agent { this.tcpDataSender = createTcpDataSender(socket); - this.spanDataSender = createUdpDataSender(this.profilerConfig.getCollectorUdpSpanServerPort(), "Pinpoint-UdpSpanDataExecutor", + this.spanDataSender = createBufferedUdpDataSender(this.profilerConfig.getCollectorUdpSpanServerPort(), "Pinpoint-UdpSpanDataExecutor", this.profilerConfig.getSpanDataSenderWriteQueueSize(), this.profilerConfig.getSpanDataSenderSocketTimeout(), this.profilerConfig.getSpanDataSenderSocketSendBufferSize(), this.profilerConfig.getSpanDataSenderChunkSize()); this.statDataSender = createUdpDataSender(this.profilerConfig.getCollectorUdpServerPort(), "Pinpoint-UdpStatDataExecutor", this.profilerConfig.getStatDataSenderWriteQueueSize(), this.profilerConfig.getStatDataSenderSocketTimeout(), - this.profilerConfig.getStatDataSenderSocketSendBufferSize(), this.profilerConfig.getStatDataSenderChunkSize()); + this.profilerConfig.getStatDataSenderSocketSendBufferSize()); this.traceContext = createTraceContext(agentInformation.getServerType()); @@ -282,7 +283,11 @@ public class DefaultAgent implements Agent { return new TcpDataSender(socket); } - protected DataSender createUdpDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize, int chunkSize) { + protected DataSender createUdpDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize) { + return new UdpDataSender(this.profilerConfig.getCollectorServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize); + } + + protected DataSender createBufferedUdpDataSender(int port, String threadName, int writeQueueSize, int timeout, int sendBufferSize, int chunkSize) { return new BufferedUdpDataSender(this.profilerConfig.getCollectorServerIp(), port, threadName, writeQueueSize, timeout, sendBufferSize, chunkSize); } 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 2152027fe..7102969be 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 @@ -20,7 +20,9 @@ import com.nhn.pinpoint.rpc.packet.SendPacket; import com.nhn.pinpoint.rpc.util.AssertUtils; import com.nhn.pinpoint.thrift.dto.TResult; import com.nhn.pinpoint.thrift.io.DeserializerFactory; +import com.nhn.pinpoint.thrift.io.HeaderTBaseDeserializer; 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; diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ChunkHeaderBufferedTBaseSerializer.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ChunkHeaderBufferedTBaseSerializer.java index 0926d0ab7..04168c8e6 100644 --- a/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ChunkHeaderBufferedTBaseSerializer.java +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/io/ChunkHeaderBufferedTBaseSerializer.java @@ -22,7 +22,7 @@ import com.nhn.pinpoint.thrift.dto.TSpanEvent; public class ChunkHeaderBufferedTBaseSerializer { private static final String FIELD_NAME_SPAN_EVENT_LIST = "spanEventList"; private static final int DEFAULT_CHUNK_SIZE = 1024 * 16; - + // span event list serialized buffer private final TBaseStream eventStream; // header @@ -115,7 +115,7 @@ public class ChunkHeaderBufferedTBaseSerializer { base.write(protocol); - if (needAuthFlush()) { + if (isNeedFlush()) { flush(); } } @@ -132,12 +132,12 @@ public class ChunkHeaderBufferedTBaseSerializer { base.write(protocol); - if (needAuthFlush()) { + if (isNeedFlush()) { flush(); } } - private boolean needAuthFlush() { + private boolean isNeedFlush() { return flushHandler != null && transport.getBufferPosition() > chunkSize; } @@ -181,7 +181,7 @@ public class ChunkHeaderBufferedTBaseSerializer { public TTransport getTransport() { return transport; } - + public int getChunkSize() { return chunkSize; } diff --git a/thrift/src/main/java/com/navercorp/pinpoint/thrift/util/SerializationUtils.java b/thrift/src/main/java/com/navercorp/pinpoint/thrift/util/SerializationUtils.java index 19b417e55..be06efdbf 100644 --- a/thrift/src/main/java/com/navercorp/pinpoint/thrift/util/SerializationUtils.java +++ b/thrift/src/main/java/com/navercorp/pinpoint/thrift/util/SerializationUtils.java @@ -12,89 +12,89 @@ import com.nhn.pinpoint.thrift.io.SerializerFactory; public final class SerializationUtils { - private static final Logger LOGGER = LoggerFactory.getLogger(SerializationUtils.class); - - private SerializationUtils() { - } + private static final Logger LOGGER = LoggerFactory.getLogger(SerializationUtils.class); - public static byte[] serialize(TBase object, SerializerFactory factory) throws TException { - assertNotNull(factory, "SerializerFactory may note be null."); + private SerializationUtils() { + } - return serialize(object, factory.createSerializer()); - } + public static byte[] serialize(TBase object, SerializerFactory factory) throws TException { + assertNotNull(factory, "SerializerFactory may note be null."); - public static byte[] serialize(TBase object, HeaderTBaseSerializer serializer) throws TException { - assertNotNull(object, "TBase may note be null."); - assertNotNull(serializer, "Serializer may note be null."); + return serialize(object, factory.createSerializer()); + } - return serializer.serialize(object); - } + public static byte[] serialize(TBase object, HeaderTBaseSerializer serializer) throws TException { + assertNotNull(object, "TBase may note be null."); + assertNotNull(serializer, "Serializer may note be null."); - public static byte[] serialize(TBase object, SerializerFactory factory, byte[] defaultValue) { - try { - return serialize(object, factory); + return serializer.serialize(object); + } + + public static byte[] serialize(TBase object, SerializerFactory factory, byte[] defaultValue) { + try { + return serialize(object, factory); } catch (Exception e) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("Serialize " + object + " fail. caused=" + e.getMessage(), e); - } + if (LOGGER.isWarnEnabled()) { + LOGGER.warn("Serialize " + object + " fail. caused=" + e.getMessage(), e); + } } - - return defaultValue; - } - public static byte[] serialize(TBase object, HeaderTBaseSerializer serializer, byte[] defaultValue) { - try { - return serialize(object, serializer); + return defaultValue; + } + + public static byte[] serialize(TBase object, HeaderTBaseSerializer serializer, byte[] defaultValue) { + try { + return serialize(object, serializer); } catch (Exception e) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("Serialize " + object + " fail. caused=" + e.getMessage(), e); - } + if (LOGGER.isWarnEnabled()) { + LOGGER.warn("Serialize " + object + " fail. caused=" + e.getMessage(), e); + } } - - return defaultValue; - } - public static TBase deserialize(byte[] objectData, DeserializerFactory factory) throws TException { - assertNotNull(factory, "DeserializerFactory may note be null."); + return defaultValue; + } - return deserialize(objectData, factory.createDeserializer()); - } + public static TBase deserialize(byte[] objectData, DeserializerFactory factory) throws TException { + assertNotNull(factory, "DeserializerFactory may note be null."); - public static TBase deserialize(byte[] objectData, HeaderTBaseDeserializer deserializer) throws TException { - assertNotNull(objectData, "TBase may note be null."); - assertNotNull(deserializer, "Deserializer may note be null."); + return deserialize(objectData, factory.createDeserializer()); + } - return deserializer.deserialize(objectData); - } + public static TBase deserialize(byte[] objectData, HeaderTBaseDeserializer deserializer) throws TException { + assertNotNull(objectData, "TBase may note be null."); + assertNotNull(deserializer, "Deserializer may note be null."); - public static TBase deserialize(byte[] objectData, DeserializerFactory factory, TBase defaultValue) { - try { - return deserialize(objectData, factory); + return deserializer.deserialize(objectData); + } + + public static TBase deserialize(byte[] objectData, DeserializerFactory factory, TBase defaultValue) { + try { + return deserialize(objectData, factory); } catch (Exception e) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("Deserialize fail. caused=" + e.getMessage(), e); - } + if (LOGGER.isWarnEnabled()) { + LOGGER.warn("Deserialize fail. caused=" + e.getMessage(), e); + } } - - return defaultValue; - } - public static TBase deserialize(byte[] objectData, HeaderTBaseDeserializer deserializer, TBase defaultValue) { - try { - return deserialize(objectData, deserializer); + return defaultValue; + } + + public static TBase deserialize(byte[] objectData, HeaderTBaseDeserializer deserializer, TBase defaultValue) { + try { + return deserialize(objectData, deserializer); } catch (Exception e) { - if (LOGGER.isWarnEnabled()) { - LOGGER.warn("Deserialize fail. caused=" + e.getMessage(), e); - } + if (LOGGER.isWarnEnabled()) { + LOGGER.warn("Deserialize fail. caused=" + e.getMessage(), e); + } } - - return defaultValue; - } - private static void assertNotNull(Object object, String message) { - - if (object == null) { - throw new NullPointerException(message); - } - } + return defaultValue; + } + + private static void assertNotNull(Object object, String message) { + + if (object == null) { + throw new NullPointerException(message); + } + } } diff --git a/web/src/main/java/com/navercorp/pinpoint/web/controller/CommandController.java b/web/src/main/java/com/navercorp/pinpoint/web/controller/CommandController.java index 6658e3e9a..948b22a06 100644 --- a/web/src/main/java/com/navercorp/pinpoint/web/controller/CommandController.java +++ b/web/src/main/java/com/navercorp/pinpoint/web/controller/CommandController.java @@ -28,6 +28,8 @@ import com.nhn.pinpoint.thrift.dto.command.TMonitorInfo; import com.nhn.pinpoint.thrift.dto.command.TThreadDump; import com.nhn.pinpoint.thrift.dto.command.TThreadState; 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.util.SerializationUtils; import com.nhn.pinpoint.web.server.PinpointSocketManager;