[pinpoint-apm#9666] Applied maxAge for collector stream

Author: smilu97

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

@@ -22,9 +22,8 @@
 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 org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -47,7 +46,10 @@ protected ClusterPoint<?> findClusterPoint(TCommandTransfer deliveryCommand) {
         String agentId = deliveryCommand.getAgentId();
         long startTimeStamp = deliveryCommand.getStartTime();
         final ClusterKey sourceKey = new ClusterKey(applicationName, agentId, startTimeStamp);
+        return findClusterPoint(sourceKey);
+    }
 
+    public ClusterPoint<?> findClusterPoint(ClusterKey sourceKey) {
         List<ClusterPoint<?>> result = new ArrayList<>();
 
         for (ClusterPoint<?> targetClusterPoint : targetClusterPointLocator.getClusterPointList()) {
@@ -62,7 +64,7 @@ protected ClusterPoint<?> findClusterPoint(TCommandTransfer deliveryCommand) {
         }
 
         if (result.size() > 1) {
-            logger.warn("Ambiguous ClusterPoint {}, {}, {} (Valid Agent list={}).", applicationName, agentId, startTimeStamp, result);
+            logger.warn("Ambiguous ClusterPoint {} (Valid Agent list={}).", sourceKey, result);
             return null;
         }
 

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

@@ -18,7 +18,6 @@
 
 import com.navercorp.pinpoint.rpc.stream.ServerStreamChannel;
 import com.navercorp.pinpoint.thrift.dto.command.TCommandTransfer;
-
 import org.apache.thrift.TBase;
 
 import java.util.Objects;
@@ -30,16 +29,27 @@ public class StreamEvent extends DefaultRouteEvent {
 
     private final ServerStreamChannel serverStreamChannel;
     private final TBase<?, ?> requestObject;
+    private final long maxAgeMs;
 
     public StreamEvent(RouteEvent routeEvent, ServerStreamChannel serverStreamChannel, TBase<?, ?> requestObject) {
         this(routeEvent.getDeliveryCommand(), serverStreamChannel, requestObject);
     }
 
     public StreamEvent(TCommandTransfer deliveryCommand, ServerStreamChannel serverStreamChannel, TBase<?, ?> requestObject) {
+        this(deliveryCommand, serverStreamChannel, requestObject, Long.MAX_VALUE);
+    }
+
+    public StreamEvent(
+            TCommandTransfer deliveryCommand,
+            ServerStreamChannel serverStreamChannel,
+            TBase<?, ?> requestObject,
+            long maxAgeMs
+    ) {
         super(deliveryCommand, serverStreamChannel.getRemoteAddress());
 
         this.serverStreamChannel = Objects.requireNonNull(serverStreamChannel, "serverStreamChannel");
         this.requestObject = Objects.requireNonNull(requestObject, "requestObject");
+        this.maxAgeMs = maxAgeMs;
     }
 
     public ServerStreamChannel getStreamChannel() {
@@ -53,21 +63,24 @@ public int getStreamChannelId() {
     public TBase<?, ?> getRequestObject() {
         return requestObject;
     }
-    
+
+    public long getMaxAgeMs() {
+        return maxAgeMs;
+    }
+
     @Override
     public String toString() {
-        final StringBuilder sb = new StringBuilder();
-        sb.append(this.getClass().getSimpleName());
-        sb.append("{");
-        sb.append("{remoteAddress=").append(getRemoteAddress()).append(", ");
-        sb.append("applicationName=").append(getDeliveryCommand().getApplicationName()).append(", ");
-        sb.append("agentId=").append(getDeliveryCommand().getAgentId()).append(", ");
-        sb.append("startTimeStamp=").append(getDeliveryCommand().getStartTime()).append(", ");
-        sb.append("serverStreamChannel=").append(getStreamChannel()).append(", ");
-        sb.append("streamChannelId=").append(getStreamChannelId()).append(", ");
-        sb.append("requestObject=").append(requestObject);
-        sb.append('}');
-        return sb.toString();
+        return this.getClass().getSimpleName() +
+                "{" +
+                "{remoteAddress=" + getRemoteAddress() + ", " +
+                "applicationName=" + getDeliveryCommand().getApplicationName() + ", " +
+                "agentId=" + getDeliveryCommand().getAgentId() + ", " +
+                "startTimeStamp=" + getDeliveryCommand().getStartTime() + ", " +
+                "serverStreamChannel=" + getStreamChannel() + ", " +
+                "streamChannelId=" + getStreamChannelId() + ", " +
+                "requestObject=" + requestObject +
+                "maxAgeMs=" + maxAgeMs +
+                '}';
     }
 
 }

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

@@ -35,13 +35,17 @@
 import com.navercorp.pinpoint.thrift.io.HeaderTBaseSerializer;
 import com.navercorp.pinpoint.thrift.io.SerializerFactory;
 import com.navercorp.pinpoint.thrift.util.SerializationUtils;
-
-import org.apache.thrift.TBase;
-import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.apache.thrift.TBase;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.lang.NonNull;
 
 import java.util.Objects;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author koo.taejin
@@ -51,6 +55,7 @@ public class StreamRouteHandler extends AbstractRouteHandler<StreamEvent> {
     public static final String ATTACHMENT_KEY = StreamRouteManager.class.getSimpleName();
 
     private final Logger logger = LogManager.getLogger(this.getClass());
+    private final ScheduledExecutorService streamExpireExecutor = createExecutor("Stream-Expire");
 
     private final RouteFilterChain<StreamEvent> streamCreateFilterChain;
     private final RouteFilterChain<ResponseEvent> responseFilterChain;
@@ -71,6 +76,15 @@ public StreamRouteHandler(ClusterPointLocator<ClusterPoint<?>> targetClusterPoin
         this.commandSerializerFactory = Objects.requireNonNull(commandSerializerFactory, "commandSerializerFactory");
     }
 
+    private static ScheduledExecutorService createExecutor(String name) {
+        return Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+            @Override
+            public Thread newThread(@NonNull Runnable r) {
+                return new Thread(r, name);
+            }
+        });
+    }
+
     @Override
     public void addRequestFilter(RouteFilter<StreamEvent> filter) {
         this.streamCreateFilterChain.addLast(filter);
@@ -89,8 +103,7 @@ public void addCloseFilter(RouteFilter<StreamRouteCloseEvent> filter) {
     public TCommandTransferResponse onRoute(StreamEvent event) {
         streamCreateFilterChain.doEvent(event);
 
-        TCommandTransferResponse routeResult = onRoute0(event);
-        return routeResult;
+        return onRoute0(event);
     }
 
     private TCommandTransferResponse onRoute0(StreamEvent event) {
@@ -118,6 +131,7 @@ private TCommandTransferResponse onRoute0(StreamEvent event) {
 
                 ClientStreamChannel producerStreamChannel = createStreamChannel((ThriftAgentConnection) clusterPoint, event.getDeliveryCommand().getPayload(), routeManager);
                 routeManager.setProducer(producerStreamChannel);
+                scheduleExpire(producerStreamChannel, event.getMaxAgeMs());
                 return createResponse(TRouteResult.OK);
             } else if (clusterPoint instanceof GrpcAgentConnection) {
                 StreamRouteManager routeManager = new StreamRouteManager(event);
@@ -127,6 +141,7 @@ private TCommandTransferResponse onRoute0(StreamEvent event) {
 
                 ClientStreamChannel producerStreamChannel = ((GrpcAgentConnection) clusterPoint).openStream(event.getRequestObject(), routeManager);
                 routeManager.setProducer(producerStreamChannel);
+                scheduleExpire(producerStreamChannel, event.getMaxAgeMs());
                 return createResponse(TRouteResult.OK);
             } else {
                 return createResponse(TRouteResult.NOT_SUPPORTED_SERVICE);
@@ -143,6 +158,13 @@ private TCommandTransferResponse onRoute0(StreamEvent event) {
         return createResponse(TRouteResult.UNKNOWN);
     }
 
+    private void scheduleExpire(ClientStreamChannel channel, long maxAgeMs) {
+        if (maxAgeMs < Long.MAX_VALUE) {
+            final Runnable expireTask = () -> channel.close();
+            streamExpireExecutor.schedule(expireTask, maxAgeMs, TimeUnit.MILLISECONDS);
+        }
+    }
+
     private ClientStreamChannel createStreamChannel(ThriftAgentConnection clusterPoint, byte[] payload, ClientStreamChannelEventHandler streamChannelEventHandler) throws StreamException {
         PinpointServer pinpointServer = clusterPoint.getPinpointServer();
         return pinpointServer.openStreamAndAwait(payload, streamChannelEventHandler, 3000);