You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by GitBox <gi...@apache.org> on 2018/10/17 03:28:18 UTC
[GitHub] peng-yongsheng closed pull request #1778: Fixed the endpoint
topology bug.
peng-yongsheng closed pull request #1778: Fixed the endpoint topology bug.
URL: https://github.com/apache/incubator-skywalking/pull/1778
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointCallRelationDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointCallRelationDispatcher.java
index 8b011694c..6dc5ea0dd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointCallRelationDispatcher.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointCallRelationDispatcher.java
@@ -23,15 +23,12 @@
import org.apache.skywalking.oap.server.core.source.EndpointRelation;
/**
- * @author wusheng
+ * @author wusheng, peng-yongsheng
*/
public class EndpointCallRelationDispatcher implements SourceDispatcher<EndpointRelation> {
@Override
public void dispatch(EndpointRelation source) {
switch (source.getDetectPoint()) {
- case CLIENT:
- clientSide(source);
- break;
case SERVER:
serverSide(source);
break;
@@ -45,12 +42,4 @@ private void serverSide(EndpointRelation source) {
indicator.setDestEndpointId(source.getChildEndpointId());
IndicatorProcess.INSTANCE.in(indicator);
}
-
- private void clientSide(EndpointRelation source) {
- EndpointRelationClientSideIndicator indicator = new EndpointRelationClientSideIndicator();
- indicator.setTimeBucket(source.getTimeBucket());
- indicator.setSourceEndpointId(source.getEndpointId());
- indicator.setDestEndpointId(source.getChildEndpointId());
- IndicatorProcess.INSTANCE.in(indicator);
- }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationClientSideIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationClientSideIndicator.java
deleted file mode 100644
index 9bea74bff..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationClientSideIndicator.java
+++ /dev/null
@@ -1,151 +0,0 @@
-/*
- * 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.analysis.manual.endpointrelation;
-
-import java.util.*;
-import lombok.*;
-import org.apache.skywalking.oap.server.core.Const;
-import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
-import org.apache.skywalking.oap.server.core.analysis.indicator.annotation.IndicatorType;
-import org.apache.skywalking.oap.server.core.remote.annotation.StreamData;
-import org.apache.skywalking.oap.server.core.remote.grpc.proto.RemoteData;
-import org.apache.skywalking.oap.server.core.storage.StorageBuilder;
-import org.apache.skywalking.oap.server.core.storage.annotation.*;
-
-@IndicatorType
-@StreamData
-@StorageEntity(name = EndpointRelationClientSideIndicator.INDEX_NAME, builder = EndpointRelationClientSideIndicator.Builder.class)
-public class EndpointRelationClientSideIndicator extends Indicator {
-
- public static final String INDEX_NAME = "endpoint_relation_client_side";
- public static final String SOURCE_ENDPOINT_ID = "source_endpoint_id";
- public static final String DEST_ENDPOINT_ID = "dest_endpoint_id";
-
- @Setter @Getter @Column(columnName = SOURCE_ENDPOINT_ID) @IDColumn private int sourceEndpointId;
- @Setter @Getter @Column(columnName = DEST_ENDPOINT_ID) @IDColumn private int destEndpointId;
-
- @Override public String id() {
- String splitJointId = String.valueOf(getTimeBucket());
- splitJointId += Const.ID_SPLIT + String.valueOf(sourceEndpointId);
- splitJointId += Const.ID_SPLIT + String.valueOf(destEndpointId);
- return splitJointId;
- }
-
- @Override public void combine(Indicator indicator) {
-
- }
-
- @Override public void calculate() {
-
- }
-
- @Override public Indicator toHour() {
- EndpointRelationClientSideIndicator indicator = new EndpointRelationClientSideIndicator();
- indicator.setTimeBucket(toTimeBucketInHour());
- indicator.setSourceEndpointId(getSourceEndpointId());
- indicator.setDestEndpointId(getDestEndpointId());
- return indicator;
- }
-
- @Override public Indicator toDay() {
- EndpointRelationClientSideIndicator indicator = new EndpointRelationClientSideIndicator();
- indicator.setTimeBucket(toTimeBucketInDay());
- indicator.setSourceEndpointId(getSourceEndpointId());
- indicator.setDestEndpointId(getDestEndpointId());
- return indicator;
- }
-
- @Override public Indicator toMonth() {
- EndpointRelationClientSideIndicator indicator = new EndpointRelationClientSideIndicator();
- indicator.setTimeBucket(toTimeBucketInMonth());
- indicator.setSourceEndpointId(getSourceEndpointId());
- indicator.setDestEndpointId(getDestEndpointId());
- return indicator;
- }
-
- @Override public int remoteHashCode() {
- int result = 17;
- result = 31 * result + sourceEndpointId;
- result = 31 * result + destEndpointId;
- return result;
- }
-
- @Override public void deserialize(RemoteData remoteData) {
- setSourceEndpointId(remoteData.getDataIntegers(0));
- setDestEndpointId(remoteData.getDataIntegers(1));
- setTimeBucket(remoteData.getDataLongs(0));
- }
-
- @Override public RemoteData.Builder serialize() {
- RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
-
- remoteBuilder.setDataIntegers(1, getDestEndpointId());
- remoteBuilder.setDataIntegers(0, getSourceEndpointId());
- remoteBuilder.setDataLongs(0, getTimeBucket());
-
- return remoteBuilder;
- }
-
- @Override public int hashCode() {
- int result = 17;
- result = 31 * result + sourceEndpointId;
- result = 31 * result + destEndpointId;
- result = 31 * result + (int)getTimeBucket();
- return result;
- }
-
- @Override public boolean equals(Object obj) {
- if (this == obj)
- return true;
- if (obj == null)
- return false;
- if (getClass() != obj.getClass())
- return false;
-
- EndpointRelationClientSideIndicator indicator = (EndpointRelationClientSideIndicator)obj;
- if (sourceEndpointId != indicator.sourceEndpointId)
- return false;
- if (destEndpointId != indicator.destEndpointId)
- return false;
-
- if (getTimeBucket() != indicator.getTimeBucket())
- return false;
-
- return true;
- }
-
- public static class Builder implements StorageBuilder<EndpointRelationClientSideIndicator> {
-
- @Override public EndpointRelationClientSideIndicator map2Data(Map<String, Object> dbMap) {
- EndpointRelationClientSideIndicator indicator = new EndpointRelationClientSideIndicator();
- indicator.setSourceEndpointId(((Number)dbMap.get(SOURCE_ENDPOINT_ID)).intValue());
- indicator.setDestEndpointId(((Number)dbMap.get(DEST_ENDPOINT_ID)).intValue());
- indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
- return indicator;
- }
-
- @Override public Map<String, Object> data2Map(EndpointRelationClientSideIndicator storageData) {
- Map<String, Object> map = new HashMap<>();
- map.put(TIME_BUCKET, storageData.getTimeBucket());
- map.put(SOURCE_ENDPOINT_ID, storageData.getSourceEndpointId());
- map.put(DEST_ENDPOINT_ID, storageData.getDestEndpointId());
- return map;
- }
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyQueryService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyQueryService.java
index 74b80fda2..766fa7d79 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyQueryService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyQueryService.java
@@ -109,15 +109,35 @@ public Topology getEndpointTopology(final Step step, final long startTB, final l
Map<Integer, String> components = new HashMap<>();
serviceComponents.forEach(component -> components.put(component.getServiceId(), getComponentLibraryCatalogService().getComponentName(component.getComponentId())));
- List<Call> calls = getTopologyQueryDAO().loadSpecifiedDestOfServerSideEndpointRelations(step, startTB, endTB, endpointId);
- calls.addAll(getTopologyQueryDAO().loadSpecifiedSourceOfClientSideEndpointRelations(step, startTB, endTB, endpointId));
+ List<Call> serverSideCalls = getTopologyQueryDAO().loadSpecifiedDestOfServerSideEndpointRelations(step, startTB, endTB, endpointId);
+ serverSideCalls.forEach(call -> call.setDetectPoint(DetectPoint.SERVER));
- calls.forEach(call -> {
- call.setCallType(components.getOrDefault(getEndpointInventoryCache().get(call.getTarget()).getServiceId(), Const.UNKNOWN));
- });
+ serverSideCalls.forEach(call -> call.setCallType(components.getOrDefault(getEndpointInventoryCache().get(call.getTarget()).getServiceId(), Const.UNKNOWN)));
Topology topology = new Topology();
- topology.getCalls().addAll(calls);
+ topology.getCalls().addAll(serverSideCalls);
+
+ Set<Integer> nodeIds = new HashSet<>();
+ serverSideCalls.forEach(call -> {
+ if (!nodeIds.contains(call.getSource())) {
+ topology.getNodes().add(buildEndpointNode(components, call.getSource()));
+ nodeIds.add(call.getSource());
+ }
+ if (!nodeIds.contains(call.getTarget())) {
+ topology.getNodes().add(buildEndpointNode(components, call.getTarget()));
+ nodeIds.add(call.getTarget());
+ }
+ });
+
return topology;
}
+
+ private Node buildEndpointNode(Map<Integer, String> components, int endpointId) {
+ Node node = new Node();
+ node.setId(endpointId);
+ node.setName(getEndpointInventoryCache().get(endpointId).getName());
+ node.setType(components.getOrDefault(getEndpointInventoryCache().get(endpointId).getServiceId(), Const.UNKNOWN));
+ node.setReal(true);
+ return node;
+ }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ITopologyQueryDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ITopologyQueryDAO.java
index 514fb75b8..440b8b3a8 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ITopologyQueryDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/query/ITopologyQueryDAO.java
@@ -45,7 +45,4 @@
List<Call> loadSpecifiedDestOfServerSideEndpointRelations(Step step, long startTB, long endTB,
int destEndpointId) throws IOException;
-
- List<Call> loadSpecifiedSourceOfClientSideEndpointRelations(Step step, long startTB, long endTB,
- int sourceEndpointId) throws IOException;
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java
index dd7797c56..4e849fa83 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/endpoint/MultiScopesSpanListener.java
@@ -182,7 +182,6 @@ private void setPublicAttrs(SourceBuilder sourceBuilder, SpanDecorator spanDecor
exitSourceBuilder.setTimeBucket(minuteTimeBucket);
sourceReceiver.receive(exitSourceBuilder.toServiceRelation());
sourceReceiver.receive(exitSourceBuilder.toServiceInstanceRelation());
- sourceReceiver.receive(exitSourceBuilder.toEndpointRelation());
});
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TopologyQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TopologyQueryEsDAO.java
index b9d6093d3..05689b665 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TopologyQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/TopologyQueryEsDAO.java
@@ -21,7 +21,7 @@
import java.io.IOException;
import java.util.*;
import org.apache.skywalking.oap.server.core.UnexpectedException;
-import org.apache.skywalking.oap.server.core.analysis.manual.endpointrelation.*;
+import org.apache.skywalking.oap.server.core.analysis.manual.endpointrelation.EndpointRelationServerSideIndicator;
import org.apache.skywalking.oap.server.core.analysis.manual.service.*;
import org.apache.skywalking.oap.server.core.analysis.manual.servicerelation.*;
import org.apache.skywalking.oap.server.core.query.entity.*;
@@ -175,26 +175,15 @@ private void setQueryCondition(SearchSourceBuilder sourceBuilder, long startTB,
BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
boolQuery.must().add(QueryBuilders.rangeQuery(EndpointRelationServerSideIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
- boolQuery.must().add(QueryBuilders.termQuery(EndpointRelationServerSideIndicator.DEST_ENDPOINT_ID, destEndpointId));
- sourceBuilder.query(boolQuery);
-
- return load(sourceBuilder, indexName, EndpointRelationServerSideIndicator.SOURCE_ENDPOINT_ID, EndpointRelationServerSideIndicator.DEST_ENDPOINT_ID, Source.Endpoint);
- }
-
- @Override
- public List<Call> loadSpecifiedSourceOfClientSideEndpointRelations(Step step, long startTB, long endTB,
- int sourceEndpointId) throws IOException {
- String indexName = DownsampleingModelNameBuilder.build(step, EndpointRelationClientSideIndicator.INDEX_NAME);
- SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
- sourceBuilder.size(0);
+ BoolQueryBuilder serviceIdBoolQuery = QueryBuilders.boolQuery();
+ boolQuery.must().add(serviceIdBoolQuery);
+ serviceIdBoolQuery.should().add(QueryBuilders.termQuery(EndpointRelationServerSideIndicator.SOURCE_ENDPOINT_ID, destEndpointId));
+ serviceIdBoolQuery.should().add(QueryBuilders.termQuery(EndpointRelationServerSideIndicator.DEST_ENDPOINT_ID, destEndpointId));
- BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
- boolQuery.must().add(QueryBuilders.rangeQuery(EndpointRelationClientSideIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
- boolQuery.must().add(QueryBuilders.termQuery(EndpointRelationClientSideIndicator.SOURCE_ENDPOINT_ID, sourceEndpointId));
sourceBuilder.query(boolQuery);
- return load(sourceBuilder, indexName, EndpointRelationClientSideIndicator.SOURCE_ENDPOINT_ID, EndpointRelationClientSideIndicator.DEST_ENDPOINT_ID, Source.Endpoint);
+ return load(sourceBuilder, indexName, EndpointRelationServerSideIndicator.SOURCE_ENDPOINT_ID, EndpointRelationServerSideIndicator.DEST_ENDPOINT_ID, Source.Endpoint);
}
private List<Call> load(SearchSourceBuilder sourceBuilder, String indexName, String sourceCName,
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services