Merge pull request #1741 from Xylus/feature/issue-1708

Adds handling of queue service types

collector/src/main/java/com/navercorp/pinpoint/collector/handler/SpanHandler.java

@@ -86,26 +86,50 @@ public class SpanHandler implements SimpleHandler {
 
     private void insertSpanStat(TSpan span) {
         final ServiceType applicationServiceType = getApplicationServiceType(span);
+        final ServiceType spanServiceType = registry.findServiceType(span.getServiceType());
         // TODO consider to change span.isSetErr();
         final boolean isError = span.getErr() != 0;
         int bugCheck = 0;
         if (span.getParentSpanId() == -1) {
+            if (spanServiceType.isQueue()) {
+                // create virtual queue node
+                statisticsHandler.updateCaller(span.getAcceptorHost(), spanServiceType, span.getRemoteAddr(), span.getApplicationName(), applicationServiceType, span.getEndPoint(), span.getElapsed(), isError);
 
-            // create virtual user
-            statisticsHandler.updateCaller(span.getApplicationName(), ServiceType.USER, span.getAgentId(), span.getApplicationName(), applicationServiceType, span.getAgentId(), span.getElapsed(), isError);
+                statisticsHandler.updateCallee(span.getApplicationName(), applicationServiceType, span.getAcceptorHost(), spanServiceType, span.getAgentId(), span.getElapsed(), isError);
+            } else {
+                // create virtual user
+                statisticsHandler.updateCaller(span.getApplicationName(), ServiceType.USER, span.getAgentId(), span.getApplicationName(), applicationServiceType, span.getAgentId(), span.getElapsed(), isError);
 
-            // update the span information of the current node (self)
-            statisticsHandler.updateCallee(span.getApplicationName(), applicationServiceType, span.getApplicationName(), ServiceType.USER, span.getAgentId(), span.getElapsed(), isError);
+                // update the span information of the current node (self)
+                statisticsHandler.updateCallee(span.getApplicationName(), applicationServiceType, span.getApplicationName(), ServiceType.USER, span.getAgentId(), span.getElapsed(), isError);
+            }
             bugCheck++;
         }
 
         // save statistics info only when parentApplicationContext exists
         // when drawing server map based on statistics info, you must know the application name of the previous node.
         if (span.getParentApplicationName() != null) {
-            logger.debug("Received parent application name. {}", span.getParentApplicationName());
+            String parentApplicationName = span.getParentApplicationName();
+            logger.debug("Received parent application name. {}", parentApplicationName);
 
-            final ServiceType parentApplicationType = registry.findServiceType(span.getParentApplicationType());
-            statisticsHandler.updateCallee(span.getApplicationName(), applicationServiceType, span.getParentApplicationName(), parentApplicationType, span.getAgentId(), span.getElapsed(), isError);
+            ServiceType parentApplicationType = registry.findServiceType(span.getParentApplicationType());
+
+            // create virtual queue node if current' span's service type is a queue AND :
+            // 1. parent node's application service type is not a queue (it may have come from a queue that is traced)
+            // 2. current node's application service type is not a queue (current node may be a queue that is traced)
+            if (spanServiceType.isQueue()) {
+                if (!applicationServiceType.isQueue() && !parentApplicationType.isQueue()) {
+                    // emulate virtual queue node's accept Span and record it's acceptor host
+                    hostApplicationMapDao.insert(span.getRemoteAddr(), span.getAcceptorHost(), spanServiceType.getCode(), parentApplicationName, parentApplicationType.getCode());
+                    // emulate virtual queue node's send SpanEvent
+                    statisticsHandler.updateCaller(span.getAcceptorHost(), spanServiceType, span.getRemoteAddr(), span.getApplicationName(), applicationServiceType, span.getEndPoint(), span.getElapsed(), isError);
+
+                    parentApplicationName = span.getAcceptorHost();
+                    parentApplicationType = spanServiceType;
+                }
+            }
+
+            statisticsHandler.updateCallee(span.getApplicationName(), applicationServiceType, parentApplicationName, parentApplicationType, span.getAgentId(), span.getElapsed(), isError);
             bugCheck++;
         }
 
@@ -145,10 +169,10 @@ public class SpanHandler implements SimpleHandler {
             /**
              * save information to draw a server map based on statistics
              */
-            // save the information of caller (the spanevent that span called )
+            // save the information of caller (the spanevent that called span)
             statisticsHandler.updateCaller(span.getApplicationName(), applicationServiceType, span.getAgentId(), spanEvent.getDestinationId(), spanEventType, spanEvent.getEndPoint(), elapsed, hasException);
 
-            // save the information of callee (the span that called spanevent)
+            // save the information of callee (the span that spanevent called)
             statisticsHandler.updateCallee(spanEvent.getDestinationId(), spanEventType, span.getApplicationName(), applicationServiceType, span.getEndPoint(), elapsed, hasException);
         }
     }
@@ -165,7 +189,13 @@ public class SpanHandler implements SimpleHandler {
 
         final String parentApplicationName = span.getParentApplicationName();
         final short parentServiceType = span.getParentApplicationType();
-        hostApplicationMapDao.insert(acceptorHost, spanApplicationName, applicationServiceTypeCode, parentApplicationName, parentServiceType);
+
+        final ServiceType spanServiceType = registry.findServiceType(span.getServiceType());
+        if (spanServiceType.isQueue()) {
+            hostApplicationMapDao.insert(span.getEndPoint(), spanApplicationName, applicationServiceTypeCode, parentApplicationName, parentServiceType);
+        } else {
+            hostApplicationMapDao.insert(acceptorHost, spanApplicationName, applicationServiceTypeCode, parentApplicationName, parentServiceType);
+        }
     }
     
     private ServiceType getApplicationServiceType(TSpan span) {

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanBo.java

@@ -72,7 +72,7 @@ public class SpanBo implements Span {
     
     private Short applicationServiceType;
 
-    
+    private String acceptorHost;
     private String remoteAddr; // optional
 
     private byte loggingTransactionInfo; //optional
@@ -107,7 +107,8 @@ public class SpanBo implements Span {
         this.apiId = span.getApiId();
 
         this.errCode = span.getErr();
-        
+
+        this.acceptorHost = span.getAcceptorHost();
         this.remoteAddr = span.getRemoteAddr();
         
         this.loggingTransactionInfo = span.getLoggingTransactionInfo();
@@ -334,6 +335,14 @@ public class SpanBo implements Span {
         this.errCode = errCode;
     }
 
+    public String getAcceptorHost() {
+        return acceptorHost;
+    }
+
+    public void setAcceptorHost(String acceptorHost) {
+        this.acceptorHost = acceptorHost;
+    }
+
     public String getRemoteAddr() {
         return remoteAddr;
     }
@@ -456,6 +465,8 @@ public class SpanBo implements Span {
 
         buffer.put(loggingTransactionInfo);
 
+        buffer.putPrefixedString(acceptorHost);
+
         return buffer.getBuffer();
     }
 
@@ -511,6 +522,10 @@ public class SpanBo implements Span {
             this.loggingTransactionInfo = buffer.readByte();
         }
 
+        if (buffer.limit() > 0) {
+            this.acceptorHost = buffer.readPrefixedString();
+        }
+
         return buffer.getOffset();
     }
 
@@ -532,6 +547,7 @@ public class SpanBo implements Span {
         sb.append(", elapsed=").append(elapsed);
         sb.append(", rpc='").append(rpc).append('\'');
         sb.append(", serviceType=").append(serviceType);
+        sb.append(", acceptorHost=").append(acceptorHost);
         sb.append(", endPoint='").append(endPoint).append('\'');
         sb.append(", apiId=").append(apiId);
         sb.append(", annotationBoList=").append(annotationBoList);

commons/src/main/java/com/navercorp/pinpoint/common/trace/AnnotationKey.java

@@ -157,6 +157,8 @@ public interface AnnotationKey {
     AnnotationKey HTTP_IO = AnnotationKeyFactory.of(49, "http.io", VIEW_IN_RECORD_SET);
     // post method parameter of httpclient
 
+    AnnotationKey MESSAGE_QUEUE_URI = AnnotationKeyFactory.of(100, "message.queue.url");
+
     AnnotationKey ARGS0 = AnnotationKeyFactory.of(-1, "args[0]");
     AnnotationKey ARGS1 = AnnotationKeyFactory.of(-2, "args[1]");
     AnnotationKey ARGS2 = AnnotationKeyFactory.of(-3, "args[2]");

commons/src/main/java/com/navercorp/pinpoint/common/trace/DefaultServiceType.java

@@ -26,6 +26,7 @@ class DefaultServiceType implements ServiceType {
     private final String name;
     private final String desc;
     private final boolean terminal;
+    private final boolean queue;
 
     // FIXME record statistics of only rpc call currently. so is it all right to chane into isRecordRpc()
     private final boolean recordStatistics;
@@ -49,6 +50,7 @@ class DefaultServiceType implements ServiceType {
         this.category = ServiceTypeCategory.findCategory((short)code);
 
         boolean terminal = false;
+        boolean queue = false;
         boolean recordStatistics = false;
         boolean includeDestinationId = false;
         
@@ -57,6 +59,10 @@ class DefaultServiceType implements ServiceType {
             case TERMINAL:
                 terminal = true;
                 break;
+
+            case QUEUE:
+                queue = true;
+                break;
                 
             case RECORD_STATISTICS:
                 recordStatistics = true;
@@ -71,6 +77,7 @@ class DefaultServiceType implements ServiceType {
         }
         
         this.terminal = terminal;
+        this.queue = queue;
         this.recordStatistics = recordStatistics;
         this.includeDestinationId = includeDestinationId;
     }
@@ -123,6 +130,11 @@ class DefaultServiceType implements ServiceType {
         return terminal;
     }
 
+    @Override
+    public boolean isQueue() {
+        return queue;
+    }
+
     @Override
     public boolean isIncludeDestinationId() {
         return includeDestinationId;

commons/src/main/java/com/navercorp/pinpoint/common/trace/ServiceType.java

@@ -172,6 +172,8 @@ public interface ServiceType {
 
     boolean isTerminal();
 
+    boolean isQueue();
+
     boolean isIncludeDestinationId();
 
     ServiceTypeCategory getCategory();

commons/src/main/java/com/navercorp/pinpoint/common/trace/ServiceTypeProperty.java

@@ -21,6 +21,7 @@ package com.navercorp.pinpoint.common.trace;
  */
 public enum ServiceTypeProperty {
     TERMINAL,
+    QUEUE,
     RECORD_STATISTICS,
     INCLUDE_DESTINATION_ID
 }

web/src/main/java/com/navercorp/pinpoint/web/applicationmap/ApplicationMapBuilder.java

@@ -355,6 +355,10 @@ public class ApplicationMapBuilder {
             } else if (nodeType.isTerminal() || nodeType.isUnknown()) {
                 final NodeHistogram nodeHistogram = createTerminalNodeHistogram(node, linkList);
                 node.setNodeHistogram(nodeHistogram);
+            } else if (nodeType.isQueue()) {
+                // Virtual queue node - queues with agent installed will be handled above as a WAS node
+                final NodeHistogram nodeHistogram = createTerminalNodeHistogram(node, linkList);
+                node.setNodeHistogram(nodeHistogram);
             } else if (nodeType.isUser()) {
                 // for User nodes, find its source link and create the histogram
                 Application userNode = node.getApplication();
@@ -409,7 +413,7 @@ public class ApplicationMapBuilder {
         nodeHistogram.setApplicationTimeHistogram(applicationTimeHistogram);
 
         // for Terminal nodes, create AgentLevel histogram
-        if (nodeApplication.getServiceType().isTerminal()) {
+        if (nodeApplication.getServiceType().isTerminal() || nodeApplication.getServiceType().isQueue()) {
             final Map<String, Histogram> agentHistogramMap = new HashMap<>();
 
             for (Link link : toLinkList) {
@@ -454,7 +458,9 @@ public class ApplicationMapBuilder {
             return;
         }
 
-        if (nodeServiceType.isTerminal()) {
+        if (nodeServiceType.isWas()) {
+            agentInfoPopulator.addAgentInfos(node);
+        } else if (nodeServiceType.isTerminal() || nodeServiceType.isQueue()) {
             // extract information about the terminal node
             ServerBuilder builder = new ServerBuilder();
             for (LinkData linkData : linkDataDuplexMap.getSourceLinkDataList()) {
@@ -465,8 +471,6 @@ public class ApplicationMapBuilder {
             }
             ServerInstanceList serverInstanceList = builder.build();
             node.setServerInstanceList(serverInstanceList);
-        } else if (nodeServiceType.isWas()) {
-            agentInfoPopulator.addAgentInfos(node);
         } else {
             // add empty information
             node.setServerInstanceList(new ServerInstanceList());

web/src/main/java/com/navercorp/pinpoint/web/applicationmap/Link.java

@@ -93,6 +93,10 @@ public class Link {
         if (fromNode.getServiceType() == ServiceType.USER) {
             return toNode.getApplication();
         }
+        // same goes for virtual queue nodes
+        if (!fromNode.getServiceType().isWas() && fromNode.getServiceType().isQueue()) {
+            return toNode.getApplication();
+        }
         return fromNode.getApplication();
     }
 

web/src/main/java/com/navercorp/pinpoint/web/filter/LinkFilter.java

@@ -146,6 +146,8 @@ public class LinkFilter implements Filter {
         USER_TO_WAS,
         WAS_TO_UNKNOWN,
         WAS_TO_BACKEND,
+        WAS_TO_QUEUE,
+        QUEUE_TO_WAS,
         UNSUPPORTED
     }
 
@@ -165,6 +167,12 @@ public class LinkFilter implements Filter {
         if (includeWas(fromServiceDescList) && includeUnknown(toServiceDescList)) {
             return FilterType.WAS_TO_UNKNOWN;
         }
+        if (includeWas(fromServiceDescList) && includeQueue(toServiceDescList)) {
+            return FilterType.WAS_TO_QUEUE;
+        }
+        if (includeQueue(fromServiceDescList) && includeWas(toServiceDescList)) {
+            return FilterType.QUEUE_TO_WAS;
+        }
         // TODO toServiceDescList check logic not exist.
 //        if (includeWas(fromServiceDescList) && isBackEnd????()) {
         if (includeWas(fromServiceDescList)) {
@@ -214,6 +222,12 @@ public class LinkFilter implements Filter {
             case WAS_TO_WAS: {
                 return wasToWasFilter(transaction);
             }
+            case WAS_TO_QUEUE: {
+                return wasToQueueFilter(transaction);
+            }
+            case QUEUE_TO_WAS: {
+                return queueToWasFilter(transaction);
+            }
             case WAS_TO_BACKEND: {
                 return wasToBackendFilter(transaction);
             }
@@ -329,6 +343,30 @@ public class LinkFilter implements Filter {
         return fromBaseFilter(fromSpanList);
     }
 
+    /**
+     * WAS -> Queue (virtual)
+     * Should be the same as {@link #wasToBackendFilter}
+     */
+    private boolean wasToQueueFilter(List<SpanBo> transaction) {
+        return wasToBackendFilter(transaction);
+    }
+
+    /**
+     * Queue (virtual) -> WAS
+     */
+    private boolean queueToWasFilter(List<SpanBo> transaction) {
+        final List<SpanBo> toNode = findToNode(transaction);
+        logger.debug("matching toNode spans: {}", toNode);
+        for (SpanBo span : toNode) {
+            if (fromApplicationName.equals(span.getAcceptorHost())) {
+                if (checkResponseCondition(span.getElapsed(), isError(span))) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     private boolean fromBaseFilter(List<SpanBo> fromSpanList) {
         // from base filter. hint base filter
         // exceptional case
@@ -341,7 +379,7 @@ public class LinkFilter implements Filter {
             }
             for (SpanEventBo event : eventBoList) {
                 final ServiceType eventServiceType = serviceTypeRegistryService.findServiceType(event.getServiceType());
-                if (!eventServiceType.isRpcClient()) {
+                if (!eventServiceType.isRpcClient() || !eventServiceType.isQueue()) {
                     continue;
                 }
                 if (!eventServiceType.isRecordStatistics()) {
@@ -407,16 +445,16 @@ public class LinkFilter implements Filter {
     private List<SpanBo> findNode(List<SpanBo> nodeList, String findApplicationName, List<ServiceType> findServiceCode, AgentFilter agentFilter) {
         List<SpanBo> findList = null;
         for (SpanBo span : nodeList) {
-            final ServiceType spanServiceType = serviceTypeRegistryService.findServiceType(span.getServiceType());
-            if (findApplicationName.equals(span.getApplicationId()) && includeServiceType(findServiceCode, spanServiceType)) {
-                // apply preAgentFilter
-                if (agentFilter.accept(span.getAgentId())) {
-                    if (findList == null) {
-                        findList = new ArrayList<>();
+                final ServiceType applicationServiceType = serviceTypeRegistryService.findServiceType(span.getApplicationServiceType());
+                if (findApplicationName.equals(span.getApplicationId()) && includeServiceType(findServiceCode, applicationServiceType)) {
+                    // apply preAgentFilter
+                    if (agentFilter.accept(span.getAgentId())) {
+                        if (findList == null) {
+                            findList = new ArrayList<>();
+                        }
+                        findList.add(span);
                     }
-                    findList.add(span);
                 }
-            }
         }
         if (findList == null) {
             return Collections.emptyList();
@@ -448,6 +486,15 @@ public class LinkFilter implements Filter {
         return false;
     }
 
+    private boolean includeQueue(List<ServiceType> serviceTypeList) {
+        for (ServiceType serviceType : serviceTypeList) {
+            if (serviceType.isQueue()) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private boolean includeServiceType(List<ServiceType> serviceTypeList, ServiceType targetServiceType) {
         for (ServiceType serviceType : serviceTypeList) {
             if (serviceType == targetServiceType) {

web/src/main/java/com/navercorp/pinpoint/web/service/BFSLinkSelector.java

@@ -166,7 +166,7 @@ public class BFSLinkSelector implements LinkSelector {
     private List<LinkData> checkRpcCallAccepted(LinkData linkData, Range range) {
         // replace if the rpc client's destination has an agent installed and thus has an application name
         final Application toApplication = linkData.getToApplication();
-        if (!toApplication.getServiceType().isRpcClient()) {
+        if (!toApplication.getServiceType().isRpcClient() && !toApplication.getServiceType().isQueue()) {
             return Collections.singletonList(linkData);
         }
 
@@ -187,10 +187,15 @@ public class BFSLinkSelector implements LinkSelector {
                 return createVirtualLinkData(linkData, toApplication, acceptApplicationList);
             }
         } else {
-            final Application unknown = new Application(toApplication.getName(), ServiceType.UNKNOWN);
-            final LinkData unknownLinkData = new LinkData(linkData.getFromApplication(), unknown);
-            unknownLinkData.setLinkCallDataMap(linkData.getLinkCallDataMap());
-            return Collections.singletonList(unknownLinkData);
+            // for queues, accept application may not exist if no consumers have an agent installed
+            if (toApplication.getServiceType().isQueue()) {
+                return Collections.singletonList(linkData);
+            } else {
+                final Application unknown = new Application(toApplication.getName(), ServiceType.UNKNOWN);
+                final LinkData unknownLinkData = new LinkData(linkData.getFromApplication(), unknown);
+                unknownLinkData.setLinkCallDataMap(linkData.getLinkCallDataMap());
+                return Collections.singletonList(unknownLinkData);
+            }
         }
 
     }

web/src/main/java/com/navercorp/pinpoint/web/service/FilteredMapServiceImpl.java

@@ -416,10 +416,33 @@ public class FilteredMapServiceImpl implements FilteredMapService {
     private Application createParentApplication(SpanBo span, Map<Long, SpanBo> transactionSpanMap) {
         final SpanBo parentSpan = transactionSpanMap.get(span.getParentSpanId());
         if (span.isRoot() || parentSpan == null) {
-            String applicationName = span.getApplicationId();
-            ServiceType serviceType = ServiceType.USER;
-            return this.applicationFactory.createApplication(applicationName, serviceType);
+            ServiceType spanServiceType = this.registry.findServiceType(span.getServiceType());
+            if (spanServiceType.isQueue()) {
+                String applicationName = span.getAcceptorHost();
+                ServiceType serviceType = spanServiceType;
+                return this.applicationFactory.createApplication(applicationName, serviceType);
+            } else {
+                String applicationName = span.getApplicationId();
+                ServiceType serviceType = ServiceType.USER;
+                return this.applicationFactory.createApplication(applicationName, serviceType);
+            }
         } else {
+            // create virtual queue node if current' span's service type is a queue AND :
+            // 1. parent node's application service type is not a queue (it may have come from a queue that is traced)
+            // 2. current node's application service type is not a queue (current node may be a queue that is traced)
+            ServiceType spanServiceType = this.registry.findServiceType(span.getServiceType());
+            if (spanServiceType.isQueue()) {
+                ServiceType parentApplicationServiceType = this.registry.findServiceType(parentSpan.getApplicationServiceType());
+                ServiceType spanApplicationServiceType = this.registry.findServiceType(span.getApplicationServiceType());
+                if (!parentApplicationServiceType.isQueue() && !spanApplicationServiceType.isQueue()) {
+                    String parentApplicationName = span.getAcceptorHost();
+                    if (parentApplicationName == null) {
+                        parentApplicationName = span.getRemoteAddr();
+                    }
+                    short parentServiceType = span.getServiceType();
+                    return this.applicationFactory.createApplication(parentApplicationName, parentServiceType);
+                }
+            }
             String parentApplicationName = parentSpan.getApplicationId();
             short parentServiceType = parentSpan.getApplicationServiceType();
             return this.applicationFactory.createApplication(parentApplicationName, parentServiceType);

web/src/main/java/com/navercorp/pinpoint/web/view/NodeSerializer.java

@@ -57,6 +57,7 @@ public class NodeSerializer extends JsonSerializer<Node>  {
         jgen.writeStringField("serviceTypeCode", Short.toString(serviceType.getCode()));
 //        jgen.writeStringField("terminal", Boolean.toString(serviceType.isTerminal()));
         jgen.writeBooleanField("isWas", serviceType.isWas());  // for go.js
+        jgen.writeBooleanField("isQueue", serviceType.isQueue());
 
 
 
@@ -81,7 +82,8 @@ public class NodeSerializer extends JsonSerializer<Node>  {
     private void writeHistogram(JsonGenerator jgen, Node node) throws IOException {
         final ServiceType serviceType = node.getServiceType();
         final NodeHistogram nodeHistogram = node.getNodeHistogram();
-        if (serviceType.isWas() || serviceType.isTerminal() || serviceType.isUnknown() || serviceType.isUser()) {
+        // FIXME isn't this all ServiceTypes that can be a node?
+        if (serviceType.isWas() || serviceType.isTerminal() || serviceType.isUnknown() || serviceType.isUser() || serviceType.isQueue()) {
             Histogram applicationHistogram = nodeHistogram.getApplicationHistogram();
             if (applicationHistogram == null) {
                 writeEmptyObject(jgen, "histogram");
@@ -113,7 +115,8 @@ public class NodeSerializer extends JsonSerializer<Node>  {
         } else {
             jgen.writeBooleanField("hasAlert", false);  // for go.js
         }
-        if (serviceType.isWas() || serviceType.isUser() || serviceType.isTerminal() || serviceType.isUnknown()) {
+        // FIXME isn't this all ServiceTypes that can be a node?
+        if (serviceType.isWas() || serviceType.isUser() || serviceType.isTerminal() || serviceType.isUnknown() || serviceType.isQueue()) {
             List<ResponseTimeViewModel> applicationTimeSeriesHistogram = nodeHistogram.getApplicationTimeHistogram();
             if (applicationTimeSeriesHistogram == null) {
                 writeEmptyArray(jgen, "timeSeriesHistogram");

web/src/main/webapp/features/serverMap/server-map.directive.js

@@ -103,7 +103,7 @@
 	                 */
 	                extractMergeTypeList = function( serverMapData ) {
 	                	serverMapData.nodeDataArray.forEach( function( o ) {
-	                		if ( o.isWas === false && o.serviceType !== "USER" ) {
+	                		if ( o.isWas === false && o.isQueue === false && o.serviceType !== "USER" ) {
 	                			if ( angular.isUndefined( scope.mergeStatus[o.serviceType] ) ) {
 		                			scope.mergeTypeList.push( o.serviceType );
 		                			scope.mergeStatus[o.serviceType] = true;