Fix endpoint dependency bug in MQ and uninstrumented proxy cases, and support...

Fix endpoint dependency bug in MQ and uninstrumented proxy cases, and support endpoint dependency(v2 of endpoint topology case). (#4995)

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyQueryService.java

@@ -30,6 +30,8 @@ import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.analysis.IDManager;
 import org.apache.skywalking.oap.server.core.config.IComponentLibraryCatalogService;
 import org.apache.skywalking.oap.server.core.query.type.Call;
+import org.apache.skywalking.oap.server.core.query.type.EndpointNode;
+import org.apache.skywalking.oap.server.core.query.type.EndpointTopology;
 import org.apache.skywalking.oap.server.core.query.type.Node;
 import org.apache.skywalking.oap.server.core.query.type.ServiceInstanceTopology;
 import org.apache.skywalking.oap.server.core.query.type.Topology;
@@ -132,6 +134,7 @@ public class TopologyQueryService implements Service {
         return builder.build(serviceInstanceRelationClientCalls, serviceInstanceRelationServerCalls);
     }
 
+    @Deprecated
     public Topology getEndpointTopology(final long startTB, final long endTB,
                                         final String endpointId) throws IOException {
         List<Call.CallDetail> serverSideCalls = getTopologyQueryDAO().loadEndpointRelation(
@@ -162,6 +165,37 @@ public class TopologyQueryService implements Service {
         return topology;
     }
 
+    public EndpointTopology getEndpointDependencies(final long startTB, final long endTB,
+                                                    final String endpointId) throws IOException {
+        List<Call.CallDetail> serverSideCalls = getTopologyQueryDAO().loadEndpointRelation(
+            startTB, endTB, endpointId);
+
+        EndpointTopology topology = new EndpointTopology();
+        serverSideCalls.forEach(callDetail -> {
+            Call call = new Call();
+            call.setId(callDetail.getId());
+            call.setSource(callDetail.getSource());
+            call.setTarget(callDetail.getTarget());
+            call.addDetectPoint(DetectPoint.SERVER);
+            topology.getCalls().add(call);
+        });
+
+        Set<String> nodeIds = new HashSet<>();
+        serverSideCalls.forEach(call -> {
+            if (!nodeIds.contains(call.getSource())) {
+                topology.getNodes().add(buildEndpointDependencyNode(call.getSource()));
+                nodeIds.add(call.getSource());
+            }
+            if (!nodeIds.contains(call.getTarget())) {
+                topology.getNodes().add(buildEndpointDependencyNode(call.getTarget()));
+                nodeIds.add(call.getTarget());
+            }
+        });
+
+        return topology;
+    }
+
+    @Deprecated
     private Node buildEndpointNode(String endpointId) {
         Node node = new Node();
         node.setId(endpointId);
@@ -172,4 +206,18 @@ public class TopologyQueryService implements Service {
         node.setReal(true);
         return node;
     }
+
+    private EndpointNode buildEndpointDependencyNode(String endpointId) {
+        final IDManager.EndpointID.EndpointIDDefinition endpointIDDefinition = IDManager.EndpointID.analysisId(
+            endpointId);
+        EndpointNode instanceNode = new EndpointNode();
+        instanceNode.setId(endpointId);
+        instanceNode.setName(endpointIDDefinition.getEndpointName());
+        instanceNode.setServiceId(endpointIDDefinition.getServiceId());
+        final IDManager.ServiceID.ServiceIDDefinition serviceIDDefinition = IDManager.ServiceID.analysisId(
+            endpointIDDefinition.getServiceId());
+        instanceNode.setServiceName(serviceIDDefinition.getName());
+        instanceNode.setReal(serviceIDDefinition.isReal());
+        return instanceNode;
+    }
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/EndpointNode.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.query.type;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Setter
+@Getter
+public class EndpointNode {
+    private String id;
+    private String name;
+    private String serviceId;
+    private String serviceName;
+    /**
+     * Not type for endpoint for now.
+     */
+    private String type = "";
+    private boolean isReal;
+}

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/type/EndpointTopology.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.core.query.type;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Getter;
+
+@Getter
+public class EndpointTopology {
+    private final List<EndpointNode> nodes;
+    private final List<Call> calls;
+
+    public EndpointTopology() {
+        this.nodes = new ArrayList<>();
+        this.calls = new ArrayList<>();
+    }
+}

oap-server/server-query-plugin/query-graphql-plugin/src/main/java/org/apache/skywalking/oap/query/graphql/resolver/TopologyQuery.java

@@ -25,6 +25,7 @@ import java.util.List;
 import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.query.TopologyQueryService;
 import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.query.type.EndpointTopology;
 import org.apache.skywalking.oap.server.core.query.type.ServiceInstanceTopology;
 import org.apache.skywalking.oap.server.core.query.type.Topology;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
@@ -69,8 +70,18 @@ public class TopologyQuery implements GraphQLQueryResolver {
         );
     }
 
+    /**
+     * Replaced by {@link #getEndpointDependencies(String, Duration)}
+     */
+    @Deprecated
     public Topology getEndpointTopology(final String endpointId, final Duration duration) throws IOException {
         return getQueryService().getEndpointTopology(
             duration.getStartTimeBucket(), duration.getEndTimeBucket(), endpointId);
     }
+
+    public EndpointTopology getEndpointDependencies(final String endpointId,
+                                                    final Duration duration) throws IOException {
+        return getQueryService().getEndpointDependencies(
+            duration.getStartTimeBucket(), duration.getEndTimeBucket(), endpointId);
+    }
 }

query-protocol @ bea847b9

-Subproject commit 4c1d1d996f6baece949fce90c676647b52e25620
+Subproject commit bea847b90e08c07a5407c4121fe4cec1eec77a78

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/MultiScopesAnalysisListener.java

@@ -107,6 +107,7 @@ public class MultiScopesAnalysisListener implements EntryAnalysisListener, ExitA
                 if (span.getSpanLayer().equals(SpanLayer.MQ) ||
                     config.getUninstrumentedGatewaysConfig().isAddressConfiguredAsGateway(networkAddressUsedAtPeer)) {
                     sourceBuilder.setSourceServiceName(networkAddressUsedAtPeer);
+                    sourceBuilder.setSourceEndpointOwnerServiceName(reference.getParentService());
                     sourceBuilder.setSourceServiceInstanceName(networkAddressUsedAtPeer);
                     sourceBuilder.setSourceNodeType(NodeType.fromSpanLayerValue(span.getSpanLayer()));
                 } else {

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/SourceBuilder.java

@@ -57,6 +57,17 @@ class SourceBuilder {
         this.sourceServiceInstanceName = namingControl.formatInstanceName(sourceServiceInstanceName);
     }
 
+    /**
+     * Source endpoint could be not owned by {@link #sourceServiceName}, such as in the MQ or un-instrumented proxy
+     * cases. This service always comes from the span.ref, so it is always a normal service.
+     */
+    @Getter
+    private String sourceEndpointOwnerServiceName;
+
+    public void setSourceEndpointOwnerServiceName(final String sourceServiceName) {
+        this.sourceEndpointOwnerServiceName = namingControl.formatServiceName(sourceServiceName);
+    }
+
     @Getter
     private String sourceEndpointName;
 
@@ -234,8 +245,13 @@ class SourceBuilder {
         }
         EndpointRelation endpointRelation = new EndpointRelation();
         endpointRelation.setEndpoint(sourceEndpointName);
-        endpointRelation.setServiceName(sourceServiceName);
-        endpointRelation.setServiceNodeType(sourceNodeType);
+        if (sourceEndpointOwnerServiceName == null) {
+            endpointRelation.setServiceName(sourceServiceName);
+            endpointRelation.setServiceNodeType(sourceNodeType);
+        } else {
+            endpointRelation.setServiceName(sourceEndpointOwnerServiceName);
+            endpointRelation.setServiceNodeType(NodeType.Normal);
+        }
         endpointRelation.setServiceInstanceName(sourceServiceInstanceName);
         endpointRelation.setChildEndpoint(destEndpointName);
         endpointRelation.setChildServiceName(destServiceName);

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/MultiScopesAnalysisListenerTest.java

@@ -282,6 +282,7 @@ public class MultiScopesAnalysisListenerTest {
         Assert.assertEquals("127.0.0.1", serviceInstanceRelation.getSourceServiceInstanceName());
         Assert.assertEquals(serviceInstance.getName(), serviceInstanceRelation.getDestServiceInstanceName());
         Assert.assertEquals("downstream-endpoint", endpointRelation.getEndpoint());
+        Assert.assertEquals("downstream-service", endpointRelation.getServiceName());
         Assert.assertEquals(endpoint.getName(), endpointRelation.getChildEndpoint());
     }