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/25 03:19:33 UTC
[GitHub] wu-sheng closed pull request #1809: A big change for topology
indicator and query.
wu-sheng closed pull request #1809: A big change for topology indicator and query.
URL: https://github.com/apache/incubator-skywalking/pull/1809
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/docs/en/concepts-and-designs/scope-definitions.md b/docs/en/concepts-and-designs/scope-definitions.md
index d788956eb..d039b18e0 100644
--- a/docs/en/concepts-and-designs/scope-definitions.md
+++ b/docs/en/concepts-and-designs/scope-definitions.md
@@ -119,6 +119,7 @@ Calculate the metric data from each request between one service and the other se
| destServiceName | Represent the name of the destination service. | | string |
| destServiceInstanceName | Represent the name of the destination service instance.| | string|
| endpoint | Represent the endpoint used in this call. | | string
+| componentId | Represent the id of component used in this call. | yes | string
| latency | Represent how much time of each request. | | int |
| status | Represent whether success or fail of the request.| | bool(true for success) |
| responseCode | Represent the response code of HTTP response, if this request is the HTTP call. | | int |
@@ -139,6 +140,7 @@ Calculate the metric data from each request between one service instance and the
| destServiceInstanceId | Represent the id of the destination service instance. | yes | int|
| destServiceInstanceName | Represent the name of the destination service instance. | | string |
| endpoint | Represent the endpoint used in this call. | | string
+| componentId | Represent the id of component used in this call. | yes | string
| latency | Represent how much time of each request. | | int |
| status | Represent whether success or fail of the request.| | bool(true for success) |
| responseCode | Represent the response code of HTTP response, if this request is the HTTP call. | | int |
@@ -159,6 +161,7 @@ including auto instrument agents(like Java, .NET), OpenCensus SkyWalking exporte
| childEndpointId | Represent the id of the endpoint being used by the parent endpoint in row(1) | yes | int|
| childEndpoint| Represent the endpoint being used by the parent endpoint in row(2) | | string |
| rpcLatency | Represent the latency of the RPC from some codes in the endpoint to the childEndpoint. Exclude the latency caused by the endpoint(1) itself.
+| componentId | Represent the id of component used in this call. | yes | string
| status | Represent whether success or fail of the request.| | bool(true for success) |
| responseCode | Represent the response code of HTTP response, if this request is the HTTP call. | | int |
| type | Represent the type of each request. Such as: Database, HTTP, RPC, gRPC. | | enum |
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
index 2ed633dfe..698ec7313 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
@@ -158,6 +158,8 @@ public CoreModuleProvider() {
DataTTLKeeperTimer.INSTANCE.setDataTTL(moduleConfig.getDataTTL());
DataTTLKeeperTimer.INSTANCE.start(getManager());
+
+ CacheUpdateTimer.INSTANCE.start(getManager());
}
@Override
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
index b216ff849..5e55a75b7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/DispatcherManager.java
@@ -32,7 +32,6 @@
import org.apache.skywalking.oap.server.core.analysis.generated.servicerelation.ServiceRelationDispatcher;
import org.apache.skywalking.oap.server.core.analysis.manual.endpointrelation.EndpointCallRelationDispatcher;
import org.apache.skywalking.oap.server.core.analysis.manual.segment.SegmentDispatcher;
-import org.apache.skywalking.oap.server.core.analysis.manual.service.*;
import org.apache.skywalking.oap.server.core.analysis.manual.servicerelation.ServiceCallRelationDispatcher;
import org.apache.skywalking.oap.server.core.source.*;
import org.slf4j.*;
@@ -57,9 +56,6 @@ public DispatcherManager() {
this.dispatcherMap.put(Scope.ServiceInstance, new SourceDispatcher[] {new ServiceInstanceDispatcher()});
this.dispatcherMap.put(Scope.Endpoint, new SourceDispatcher[] {new EndpointDispatcher()});
- this.dispatcherMap.put(Scope.ServiceComponent, new SourceDispatcher[] {new ServiceComponentDispatcher()});
- this.dispatcherMap.put(Scope.ServiceMapping, new SourceDispatcher[] {new ServiceMappingDispatcher()});
-
this.dispatcherMap.put(Scope.ServiceRelation, new SourceDispatcher[] {new ServiceRelationDispatcher(), new ServiceCallRelationDispatcher()});
this.dispatcherMap.put(Scope.ServiceInstanceRelation, new SourceDispatcher[] {new ServiceInstanceRelationDispatcher()});
this.dispatcherMap.put(Scope.EndpointRelation, new SourceDispatcher[] {new EndpointRelationDispatcher(), new EndpointCallRelationDispatcher()});
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 6dc5ea0dd..cd0749840 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
@@ -40,6 +40,8 @@ private void serverSide(EndpointRelation source) {
indicator.setTimeBucket(source.getTimeBucket());
indicator.setSourceEndpointId(source.getEndpointId());
indicator.setDestEndpointId(source.getChildEndpointId());
+ indicator.setComponentId(source.getComponentId());
+ indicator.setEntityId(source.getEntityId());
IndicatorProcess.INSTANCE.in(indicator);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationServerSideIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationServerSideIndicator.java
index 52c7ee518..82aaceba9 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationServerSideIndicator.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/endpointrelation/EndpointRelationServerSideIndicator.java
@@ -36,14 +36,18 @@
public static final String INDEX_NAME = "endpoint_relation_server_side";
public static final String SOURCE_ENDPOINT_ID = "source_endpoint_id";
public static final String DEST_ENDPOINT_ID = "dest_endpoint_id";
+ public static final String COMPONENT_ID = "component_id";
@Setter @Getter @Column(columnName = SOURCE_ENDPOINT_ID) @IDColumn private int sourceEndpointId;
@Setter @Getter @Column(columnName = DEST_ENDPOINT_ID) @IDColumn private int destEndpointId;
+ @Setter @Getter @Column(columnName = COMPONENT_ID) @IDColumn private int componentId;
+ @Setter @Getter @Column(columnName = ENTITY_ID) @IDColumn private String entityId;
@Override public String id() {
String splitJointId = String.valueOf(getTimeBucket());
splitJointId += Const.ID_SPLIT + String.valueOf(sourceEndpointId);
splitJointId += Const.ID_SPLIT + String.valueOf(destEndpointId);
+ splitJointId += Const.ID_SPLIT + String.valueOf(componentId);
return splitJointId;
}
@@ -60,6 +64,8 @@
indicator.setTimeBucket(toTimeBucketInHour());
indicator.setSourceEndpointId(getSourceEndpointId());
indicator.setDestEndpointId(getDestEndpointId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -68,6 +74,8 @@
indicator.setTimeBucket(toTimeBucketInDay());
indicator.setSourceEndpointId(getSourceEndpointId());
indicator.setDestEndpointId(getDestEndpointId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -76,6 +84,8 @@
indicator.setTimeBucket(toTimeBucketInMonth());
indicator.setSourceEndpointId(getSourceEndpointId());
indicator.setDestEndpointId(getDestEndpointId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -83,13 +93,16 @@
int result = 17;
result = 31 * result + sourceEndpointId;
result = 31 * result + destEndpointId;
+ result = 31 * result + componentId;
return result;
}
@Override public void deserialize(RemoteData remoteData) {
setSourceEndpointId(remoteData.getDataIntegers(0));
setDestEndpointId(remoteData.getDataIntegers(1));
+ setComponentId(remoteData.getDataIntegers(2));
setTimeBucket(remoteData.getDataLongs(0));
+ setEntityId(remoteData.getDataStrings(0));
}
@Override public RemoteData.Builder serialize() {
@@ -97,7 +110,9 @@
remoteBuilder.setDataIntegers(0, getSourceEndpointId());
remoteBuilder.setDataIntegers(1, getDestEndpointId());
+ remoteBuilder.setDataIntegers(2, getComponentId());
remoteBuilder.setDataLongs(0, getTimeBucket());
+ remoteBuilder.setDataStrings(0, getEntityId());
return remoteBuilder;
}
@@ -106,6 +121,7 @@
int result = 17;
result = 31 * result + sourceEndpointId;
result = 31 * result + destEndpointId;
+ result = 31 * result + componentId;
result = 31 * result + (int)getTimeBucket();
return result;
}
@@ -123,6 +139,8 @@
return false;
if (destEndpointId != indicator.destEndpointId)
return false;
+ if (componentId != indicator.componentId)
+ return false;
if (getTimeBucket() != indicator.getTimeBucket())
return false;
@@ -136,7 +154,9 @@
EndpointRelationServerSideIndicator indicator = new EndpointRelationServerSideIndicator();
indicator.setSourceEndpointId(((Number)dbMap.get(SOURCE_ENDPOINT_ID)).intValue());
indicator.setDestEndpointId(((Number)dbMap.get(DEST_ENDPOINT_ID)).intValue());
+ indicator.setComponentId(((Number)dbMap.get(COMPONENT_ID)).intValue());
indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
+ indicator.setEntityId((String)dbMap.get(ENTITY_ID));
return indicator;
}
@@ -144,7 +164,9 @@
Map<String, Object> map = new HashMap<>();
map.put(SOURCE_ENDPOINT_ID, storageData.getSourceEndpointId());
map.put(DEST_ENDPOINT_ID, storageData.getDestEndpointId());
+ map.put(COMPONENT_ID, storageData.getComponentId());
map.put(TIME_BUCKET, storageData.getTimeBucket());
+ map.put(ENTITY_ID, storageData.getEntityId());
return map;
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentDispatcher.java
deleted file mode 100644
index 9b2904270..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentDispatcher.java
+++ /dev/null
@@ -1,42 +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.service;
-
-import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
-import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess;
-import org.apache.skywalking.oap.server.core.source.ServiceComponent;
-
-/**
- * @author peng-yongsheng
- */
-public class ServiceComponentDispatcher implements SourceDispatcher<ServiceComponent> {
-
- @Override public void dispatch(ServiceComponent source) {
- doDispatch(source);
- }
-
- private void doDispatch(ServiceComponent source) {
- ServiceComponentIndicator indicator = new ServiceComponentIndicator();
-
- indicator.setTimeBucket(source.getTimeBucket());
- indicator.setServiceId(source.getServiceId());
- indicator.setComponentId(source.getComponentId());
- IndicatorProcess.INSTANCE.in(indicator);
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentIndicator.java
deleted file mode 100644
index d57ca5c99..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceComponentIndicator.java
+++ /dev/null
@@ -1,157 +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.service;
-
-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.*;
-
-/**
- * @author peng-yongsheng
- */
-@IndicatorType
-@StreamData
-@StorageEntity(name = ServiceComponentIndicator.INDEX_NAME, builder = ServiceComponentIndicator.Builder.class)
-public class ServiceComponentIndicator extends Indicator {
-
- public static final String INDEX_NAME = "service_component";
- public static final String SERVICE_ID = "service_id";
- public static final String COMPONENT_ID = "component_id";
-
- @Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId;
- @Setter @Getter @Column(columnName = COMPONENT_ID) private int componentId;
-
- @Override public String id() {
- String splitJointId = String.valueOf(getTimeBucket());
- splitJointId += Const.ID_SPLIT + String.valueOf(serviceId);
- splitJointId += Const.ID_SPLIT + String.valueOf(componentId);
- return splitJointId;
- }
-
- @Override public int hashCode() {
- int result = 17;
- result = 31 * result + serviceId;
- result = 31 * result + componentId;
- 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;
-
- ServiceComponentIndicator indicator = (ServiceComponentIndicator)obj;
- if (serviceId != indicator.serviceId)
- return false;
- if (componentId != indicator.componentId)
- return false;
-
- if (getTimeBucket() != indicator.getTimeBucket())
- return false;
-
- return true;
- }
-
- @Override public RemoteData.Builder serialize() {
- RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
-
- remoteBuilder.setDataLongs(0, getTimeBucket());
-
- remoteBuilder.setDataIntegers(0, getServiceId());
- remoteBuilder.setDataIntegers(1, getComponentId());
-
- return remoteBuilder;
- }
-
- @Override public void deserialize(RemoteData remoteData) {
- setTimeBucket(remoteData.getDataLongs(0));
-
- setServiceId(remoteData.getDataIntegers(0));
- setComponentId(remoteData.getDataIntegers(1));
- }
-
- @Override public void calculate() {
- }
-
- @Override public Indicator toHour() {
- ServiceComponentIndicator indicator = new ServiceComponentIndicator();
- indicator.setTimeBucket(toTimeBucketInHour());
- indicator.setServiceId(this.getServiceId());
- indicator.setComponentId(this.getComponentId());
-
- return indicator;
- }
-
- @Override public Indicator toDay() {
- ServiceComponentIndicator indicator = new ServiceComponentIndicator();
- indicator.setTimeBucket(toTimeBucketInDay());
- indicator.setServiceId(this.getServiceId());
- indicator.setComponentId(this.getComponentId());
-
- return indicator;
- }
-
- @Override public Indicator toMonth() {
- ServiceComponentIndicator indicator = new ServiceComponentIndicator();
- indicator.setTimeBucket(toTimeBucketInMonth());
- indicator.setServiceId(this.getServiceId());
- indicator.setComponentId(this.getComponentId());
-
- return indicator;
- }
-
- @Override public final void combine(Indicator indicator) {
- }
-
- @Override public int remoteHashCode() {
- int result = 17;
- result = 31 * result + serviceId;
- result = 31 * result + componentId;
- return result;
- }
-
- public static class Builder implements StorageBuilder<ServiceComponentIndicator> {
-
- @Override public Map<String, Object> data2Map(ServiceComponentIndicator storageData) {
- Map<String, Object> map = new HashMap<>();
- map.put(SERVICE_ID, storageData.getServiceId());
- map.put(COMPONENT_ID, storageData.getComponentId());
- map.put(TIME_BUCKET, storageData.getTimeBucket());
- return map;
- }
-
- @Override public ServiceComponentIndicator map2Data(Map<String, Object> dbMap) {
- ServiceComponentIndicator indicator = new ServiceComponentIndicator();
- indicator.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue());
- indicator.setComponentId(((Number)dbMap.get(COMPONENT_ID)).intValue());
- indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
- return indicator;
- }
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingDispatcher.java
deleted file mode 100644
index 7efb503c4..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingDispatcher.java
+++ /dev/null
@@ -1,42 +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.service;
-
-import org.apache.skywalking.oap.server.core.analysis.SourceDispatcher;
-import org.apache.skywalking.oap.server.core.analysis.worker.IndicatorProcess;
-import org.apache.skywalking.oap.server.core.source.ServiceMapping;
-
-/**
- * @author peng-yongsheng
- */
-public class ServiceMappingDispatcher implements SourceDispatcher<ServiceMapping> {
-
- @Override public void dispatch(ServiceMapping source) {
- doDispatch(source);
- }
-
- private void doDispatch(ServiceMapping source) {
- ServiceMappingIndicator indicator = new ServiceMappingIndicator();
-
- indicator.setTimeBucket(source.getTimeBucket());
- indicator.setServiceId(source.getServiceId());
- indicator.setMappingServiceId(source.getMappingServiceId());
- IndicatorProcess.INSTANCE.in(indicator);
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingIndicator.java
deleted file mode 100644
index 46f3c1924..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/service/ServiceMappingIndicator.java
+++ /dev/null
@@ -1,160 +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.service;
-
-import java.util.HashMap;
-import java.util.Map;
-import lombok.Getter;
-import lombok.Setter;
-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.Column;
-import org.apache.skywalking.oap.server.core.storage.annotation.StorageEntity;
-
-/**
- * @author peng-yongsheng
- */
-@IndicatorType
-@StreamData
-@StorageEntity(name = ServiceMappingIndicator.INDEX_NAME, builder = ServiceMappingIndicator.Builder.class)
-public class ServiceMappingIndicator extends Indicator {
-
- public static final String INDEX_NAME = "service_mapping";
- public static final String SERVICE_ID = "service_id";
- public static final String MAPPING_SERVICE_ID = "mapping_service_id";
-
- @Setter @Getter @Column(columnName = SERVICE_ID) private int serviceId;
- @Setter @Getter @Column(columnName = MAPPING_SERVICE_ID) private int mappingServiceId;
-
- @Override public String id() {
- String splitJointId = String.valueOf(getTimeBucket());
- splitJointId += Const.ID_SPLIT + String.valueOf(serviceId);
- splitJointId += Const.ID_SPLIT + String.valueOf(mappingServiceId);
- return splitJointId;
- }
-
- @Override public int hashCode() {
- int result = 17;
- result = 31 * result + serviceId;
- result = 31 * result + mappingServiceId;
- 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;
-
- ServiceMappingIndicator indicator = (ServiceMappingIndicator)obj;
- if (serviceId != indicator.serviceId)
- return false;
- if (mappingServiceId != indicator.mappingServiceId)
- return false;
-
- if (getTimeBucket() != indicator.getTimeBucket())
- return false;
-
- return true;
- }
-
- @Override public RemoteData.Builder serialize() {
- RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
-
- remoteBuilder.setDataLongs(0, getTimeBucket());
-
- remoteBuilder.setDataIntegers(0, getServiceId());
- remoteBuilder.setDataIntegers(1, getMappingServiceId());
-
- return remoteBuilder;
- }
-
- @Override public void deserialize(RemoteData remoteData) {
- setTimeBucket(remoteData.getDataLongs(0));
-
- setServiceId(remoteData.getDataIntegers(0));
- setMappingServiceId(remoteData.getDataIntegers(1));
- }
-
- @Override public void calculate() {
- }
-
- @Override public Indicator toHour() {
- ServiceMappingIndicator indicator = new ServiceMappingIndicator();
- indicator.setTimeBucket(toTimeBucketInHour());
- indicator.setServiceId(this.getServiceId());
- indicator.setMappingServiceId(this.getMappingServiceId());
-
- return indicator;
- }
-
- @Override public Indicator toDay() {
- ServiceMappingIndicator indicator = new ServiceMappingIndicator();
- indicator.setTimeBucket(toTimeBucketInDay());
- indicator.setServiceId(this.getServiceId());
- indicator.setMappingServiceId(this.getMappingServiceId());
-
- return indicator;
- }
-
- @Override public Indicator toMonth() {
- ServiceMappingIndicator indicator = new ServiceMappingIndicator();
- indicator.setTimeBucket(toTimeBucketInMonth());
- indicator.setServiceId(this.getServiceId());
- indicator.setMappingServiceId(this.getMappingServiceId());
-
- return indicator;
- }
-
- @Override public final void combine(Indicator indicator) {
- }
-
- @Override public int remoteHashCode() {
- int result = 17;
- result = 31 * result + serviceId;
- result = 31 * result + mappingServiceId;
- return result;
- }
-
- public static class Builder implements StorageBuilder<ServiceMappingIndicator> {
-
- @Override public Map<String, Object> data2Map(ServiceMappingIndicator storageData) {
- Map<String, Object> map = new HashMap<>();
- map.put(SERVICE_ID, storageData.getServiceId());
- map.put(MAPPING_SERVICE_ID, storageData.getMappingServiceId());
- map.put(TIME_BUCKET, storageData.getTimeBucket());
- return map;
- }
-
- @Override public ServiceMappingIndicator map2Data(Map<String, Object> dbMap) {
- ServiceMappingIndicator indicator = new ServiceMappingIndicator();
- indicator.setServiceId(((Number)dbMap.get(SERVICE_ID)).intValue());
- indicator.setMappingServiceId(((Number)dbMap.get(MAPPING_SERVICE_ID)).intValue());
- indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
- return indicator;
- }
- }
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceCallRelationDispatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceCallRelationDispatcher.java
index c1e3cb426..596f1310c 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceCallRelationDispatcher.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceCallRelationDispatcher.java
@@ -43,6 +43,8 @@ private void serverSide(ServiceRelation source) {
indicator.setTimeBucket(source.getTimeBucket());
indicator.setSourceServiceId(source.getSourceServiceId());
indicator.setDestServiceId(source.getDestServiceId());
+ indicator.setComponentId(source.getComponentId());
+ indicator.setEntityId(source.getEntityId());
IndicatorProcess.INSTANCE.in(indicator);
}
@@ -51,6 +53,8 @@ private void clientSide(ServiceRelation source) {
indicator.setTimeBucket(source.getTimeBucket());
indicator.setSourceServiceId(source.getSourceServiceId());
indicator.setDestServiceId(source.getDestServiceId());
+ indicator.setComponentId(source.getComponentId());
+ indicator.setEntityId(source.getEntityId());
IndicatorProcess.INSTANCE.in(indicator);
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationClientSideIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationClientSideIndicator.java
index cd38b00b0..2e91fc826 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationClientSideIndicator.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationClientSideIndicator.java
@@ -36,14 +36,18 @@
public static final String INDEX_NAME = "service_relation_client_side";
public static final String SOURCE_SERVICE_ID = "source_service_id";
public static final String DEST_SERVICE_ID = "dest_service_id";
+ public static final String COMPONENT_ID = "component_id";
@Setter @Getter @Column(columnName = SOURCE_SERVICE_ID) @IDColumn private int sourceServiceId;
@Setter @Getter @Column(columnName = DEST_SERVICE_ID) @IDColumn private int destServiceId;
+ @Setter @Getter @Column(columnName = COMPONENT_ID) @IDColumn private int componentId;
+ @Setter @Getter @Column(columnName = ENTITY_ID) @IDColumn private String entityId;
@Override public String id() {
String splitJointId = String.valueOf(getTimeBucket());
splitJointId += Const.ID_SPLIT + String.valueOf(sourceServiceId);
splitJointId += Const.ID_SPLIT + String.valueOf(destServiceId);
+ splitJointId += Const.ID_SPLIT + String.valueOf(componentId);
return splitJointId;
}
@@ -57,25 +61,31 @@
@Override public Indicator toHour() {
ServiceRelationClientSideIndicator indicator = new ServiceRelationClientSideIndicator();
+ indicator.setEntityId(getEntityId());
indicator.setTimeBucket(toTimeBucketInHour());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
return indicator;
}
@Override public Indicator toDay() {
ServiceRelationClientSideIndicator indicator = new ServiceRelationClientSideIndicator();
+ indicator.setEntityId(getEntityId());
indicator.setTimeBucket(toTimeBucketInDay());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
return indicator;
}
@Override public Indicator toMonth() {
ServiceRelationClientSideIndicator indicator = new ServiceRelationClientSideIndicator();
+ indicator.setEntityId(getEntityId());
indicator.setTimeBucket(toTimeBucketInMonth());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
return indicator;
}
@@ -83,13 +93,16 @@
int result = 17;
result = 31 * result + sourceServiceId;
result = 31 * result + destServiceId;
+ result = 31 * result + componentId;
return result;
}
@Override public void deserialize(RemoteData remoteData) {
setSourceServiceId(remoteData.getDataIntegers(0));
setDestServiceId(remoteData.getDataIntegers(1));
+ setComponentId(remoteData.getDataIntegers(2));
setTimeBucket(remoteData.getDataLongs(0));
+ setEntityId(remoteData.getDataStrings(0));
}
@Override public RemoteData.Builder serialize() {
@@ -97,7 +110,9 @@
remoteBuilder.setDataIntegers(1, getDestServiceId());
remoteBuilder.setDataIntegers(0, getSourceServiceId());
+ remoteBuilder.setDataIntegers(2, getComponentId());
remoteBuilder.setDataLongs(0, getTimeBucket());
+ remoteBuilder.setDataStrings(0, getEntityId());
return remoteBuilder;
}
@@ -106,6 +121,7 @@
int result = 17;
result = 31 * result + sourceServiceId;
result = 31 * result + destServiceId;
+ result = 31 * result + componentId;
result = 31 * result + (int)getTimeBucket();
return result;
}
@@ -123,6 +139,8 @@
return false;
if (destServiceId != indicator.destServiceId)
return false;
+ if (componentId != indicator.componentId)
+ return false;
if (getTimeBucket() != indicator.getTimeBucket())
return false;
@@ -136,7 +154,9 @@
ServiceRelationClientSideIndicator indicator = new ServiceRelationClientSideIndicator();
indicator.setSourceServiceId(((Number)dbMap.get(SOURCE_SERVICE_ID)).intValue());
indicator.setDestServiceId(((Number)dbMap.get(DEST_SERVICE_ID)).intValue());
+ indicator.setComponentId(((Number)dbMap.get(COMPONENT_ID)).intValue());
indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
+ indicator.setEntityId((String)dbMap.get(ENTITY_ID));
return indicator;
}
@@ -145,6 +165,8 @@
map.put(TIME_BUCKET, storageData.getTimeBucket());
map.put(SOURCE_SERVICE_ID, storageData.getSourceServiceId());
map.put(DEST_SERVICE_ID, storageData.getDestServiceId());
+ map.put(COMPONENT_ID, storageData.getComponentId());
+ map.put(ENTITY_ID, storageData.getEntityId());
return map;
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationServerSideIndicator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationServerSideIndicator.java
index c438c6852..7de0e4f85 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationServerSideIndicator.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/servicerelation/ServiceRelationServerSideIndicator.java
@@ -36,14 +36,18 @@
public static final String INDEX_NAME = "service_relation_server_side";
public static final String SOURCE_SERVICE_ID = "source_service_id";
public static final String DEST_SERVICE_ID = "dest_service_id";
+ public static final String COMPONENT_ID = "component_id";
@Setter @Getter @Column(columnName = SOURCE_SERVICE_ID) @IDColumn private int sourceServiceId;
@Setter @Getter @Column(columnName = DEST_SERVICE_ID) @IDColumn private int destServiceId;
+ @Setter @Getter @Column(columnName = COMPONENT_ID) @IDColumn private int componentId;
+ @Setter @Getter @Column(columnName = ENTITY_ID) @IDColumn private String entityId;
@Override public String id() {
String splitJointId = String.valueOf(getTimeBucket());
splitJointId += Const.ID_SPLIT + String.valueOf(sourceServiceId);
splitJointId += Const.ID_SPLIT + String.valueOf(destServiceId);
+ splitJointId += Const.ID_SPLIT + String.valueOf(componentId);
return splitJointId;
}
@@ -60,6 +64,8 @@
indicator.setTimeBucket(toTimeBucketInHour());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -68,6 +74,8 @@
indicator.setTimeBucket(toTimeBucketInDay());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -76,6 +84,8 @@
indicator.setTimeBucket(toTimeBucketInMonth());
indicator.setSourceServiceId(getSourceServiceId());
indicator.setDestServiceId(getDestServiceId());
+ indicator.setComponentId(getComponentId());
+ indicator.setEntityId(getEntityId());
return indicator;
}
@@ -83,20 +93,25 @@
int result = 17;
result = 31 * result + sourceServiceId;
result = 31 * result + destServiceId;
+ result = 31 * result + componentId;
return result;
}
@Override public void deserialize(RemoteData remoteData) {
+ setEntityId(remoteData.getDataStrings(0));
setSourceServiceId(remoteData.getDataIntegers(0));
setDestServiceId(remoteData.getDataIntegers(1));
+ setComponentId(remoteData.getDataIntegers(2));
setTimeBucket(remoteData.getDataLongs(0));
}
@Override public RemoteData.Builder serialize() {
RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
+ remoteBuilder.setDataStrings(0, getEntityId());
remoteBuilder.setDataIntegers(0, getSourceServiceId());
remoteBuilder.setDataIntegers(1, getDestServiceId());
+ remoteBuilder.setDataIntegers(2, getComponentId());
remoteBuilder.setDataLongs(0, getTimeBucket());
return remoteBuilder;
@@ -106,6 +121,7 @@
int result = 17;
result = 31 * result + sourceServiceId;
result = 31 * result + destServiceId;
+ result = 31 * result + componentId;
result = 31 * result + (int)getTimeBucket();
return result;
}
@@ -123,6 +139,8 @@
return false;
if (destServiceId != indicator.destServiceId)
return false;
+ if (componentId != indicator.componentId)
+ return false;
if (getTimeBucket() != indicator.getTimeBucket())
return false;
@@ -134,16 +152,20 @@
@Override public ServiceRelationServerSideIndicator map2Data(Map<String, Object> dbMap) {
ServiceRelationServerSideIndicator indicator = new ServiceRelationServerSideIndicator();
+ indicator.setEntityId((String)dbMap.get(ENTITY_ID));
indicator.setSourceServiceId(((Number)dbMap.get(SOURCE_SERVICE_ID)).intValue());
indicator.setDestServiceId(((Number)dbMap.get(DEST_SERVICE_ID)).intValue());
+ indicator.setComponentId(((Number)dbMap.get(COMPONENT_ID)).intValue());
indicator.setTimeBucket(((Number)dbMap.get(TIME_BUCKET)).longValue());
return indicator;
}
@Override public Map<String, Object> data2Map(ServiceRelationServerSideIndicator storageData) {
Map<String, Object> map = new HashMap<>();
+ map.put(ENTITY_ID, storageData.getEntityId());
map.put(SOURCE_SERVICE_ID, storageData.getSourceServiceId());
map.put(DEST_SERVICE_ID, storageData.getDestServiceId());
+ map.put(COMPONENT_ID, storageData.getComponentId());
map.put(TIME_BUCKET, storageData.getTimeBucket());
return map;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/CacheUpdateTimer.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/CacheUpdateTimer.java
new file mode 100644
index 000000000..3d3cbdfff
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cache/CacheUpdateTimer.java
@@ -0,0 +1,71 @@
+/*
+ * 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.cache;
+
+import java.util.*;
+import java.util.concurrent.*;
+import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.register.ServiceInventory;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.slf4j.*;
+
+/**
+ * @author peng-yongsheng
+ */
+public enum CacheUpdateTimer {
+ INSTANCE;
+
+ private static final Logger logger = LoggerFactory.getLogger(CacheUpdateTimer.class);
+
+ private Boolean isStarted = false;
+
+ public void start(ModuleManager moduleManager) {
+ logger.info("Cache update timer start");
+
+ final long timeInterval = 3;
+
+ if (!isStarted) {
+ Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
+ new RunnableWithExceptionProtection(() -> update(moduleManager),
+ t -> logger.error("Cache update failure.", t)), 1, timeInterval, TimeUnit.SECONDS);
+
+ this.isStarted = true;
+ }
+ }
+
+ private void update(ModuleManager moduleManager) {
+ IServiceInventoryCacheDAO serviceInventoryCacheDAO = moduleManager.find(StorageModule.NAME).getService(IServiceInventoryCacheDAO.class);
+ ServiceInventoryCache serviceInventoryCache = moduleManager.find(CoreModule.NAME).getService(ServiceInventoryCache.class);
+ List<ServiceInventory> serviceInventories = serviceInventoryCacheDAO.loadLastMappingUpdate();
+
+ serviceInventories.forEach(serviceInventory -> {
+ logger.info("Update mapping service id in the cache of service inventory, service id: {}, mapping service id: {}", serviceInventory.getSequence(), serviceInventory.getMappingServiceId());
+ ServiceInventory cache = serviceInventoryCache.get(serviceInventory.getSequence());
+ if (Objects.nonNull(cache)) {
+ cache.setMappingServiceId(serviceInventory.getMappingServiceId());
+ cache.setMappingLastUpdateTime(serviceInventory.getMappingLastUpdateTime());
+ } else {
+ logger.warn("Unable to found the id of {} in service inventory cache.");
+ }
+ });
+ }
+}
\ No newline at end of file
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogService.java
index 535cf9349..9e0b383bd 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogService.java
@@ -20,6 +20,7 @@
import java.io.*;
import java.util.*;
+import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.library.util.ResourceUtils;
import org.slf4j.*;
import org.yaml.snakeyaml.Yaml;
@@ -61,8 +62,9 @@ public String getComponentName(int componentId) {
}
@Override
- public String getServerName(int componentServerId) {
- return getComponentName(componentServerId);
+ public String getServerNameBasedOnComponent(int componentId) {
+ Integer serverComponentId = componentId2ServerId.get(componentId);
+ return serverComponentId == null ? Const.UNKNOWN : getComponentName(serverComponentId);
}
private void init() throws InitialComponentCatalogException {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/IComponentLibraryCatalogService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/IComponentLibraryCatalogService.java
index 486be1e67..b848b1519 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/IComponentLibraryCatalogService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/config/IComponentLibraryCatalogService.java
@@ -30,5 +30,5 @@
String getComponentName(int componentId);
- String getServerName(int componentServerId);
+ String getServerNameBasedOnComponent(int componentId);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyBuilder.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyBuilder.java
index b10f9b3d1..c518c107d 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyBuilder.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/TopologyBuilder.java
@@ -24,7 +24,7 @@
import org.apache.skywalking.oap.server.core.config.IComponentLibraryCatalogService;
import org.apache.skywalking.oap.server.core.query.entity.*;
import org.apache.skywalking.oap.server.core.register.ServiceInventory;
-import org.apache.skywalking.oap.server.core.source.*;
+import org.apache.skywalking.oap.server.core.source.DetectPoint;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.library.util.BooleanUtils;
import org.slf4j.*;
@@ -44,45 +44,31 @@
this.componentLibraryCatalogService = moduleManager.find(CoreModule.NAME).getService(IComponentLibraryCatalogService.class);
}
- Topology build(List<ServiceComponent> serviceComponents, List<ServiceMapping> serviceMappings,
- List<Call> serviceRelationClientCalls, List<Call> serviceRelationServerCalls) {
- Map<Integer, String> nodeCompMap = buildNodeCompMap(serviceComponents);
- Map<Integer, String> conjecturalNodeCompMap = buildConjecturalNodeCompMap(serviceComponents);
- Map<Integer, Integer> mappings = changeMapping2Map(serviceMappings);
+ Topology build(List<Call> serviceRelationClientCalls, List<Call> serviceRelationServerCalls) {
filterZeroSourceOrTargetReference(serviceRelationClientCalls);
filterZeroSourceOrTargetReference(serviceRelationServerCalls);
- mappingIdExchange(mappings, serviceRelationClientCalls);
- mappingIdExchange(mappings, serviceRelationServerCalls);
- List<Node> nodes = new LinkedList<>();
+ Map<Integer, Node> nodes = new HashMap<>();
List<Call> calls = new LinkedList<>();
- Set<Integer> nodeIds = new HashSet<>();
Set<String> callIds = new HashSet<>();
- serviceRelationClientCalls.forEach(clientCall -> {
+ for (Call clientCall : serviceRelationClientCalls) {
ServiceInventory source = serviceInventoryCache.get(clientCall.getSource());
ServiceInventory target = serviceInventoryCache.get(clientCall.getTarget());
- if (BooleanUtils.valueToBoolean(target.getIsAddress()) && !mappings.containsKey(target.getSequence())) {
- if (!nodeIds.contains(target.getSequence())) {
- Node conjecturalNode = new Node();
- conjecturalNode.setId(target.getSequence());
- conjecturalNode.setName(target.getName());
- conjecturalNode.setType(conjecturalNodeCompMap.getOrDefault(target.getSequence(), Const.UNKNOWN));
- conjecturalNode.setReal(false);
- nodes.add(conjecturalNode);
- nodeIds.add(target.getSequence());
- }
+ if (target.getMappingServiceId() != Const.NONE) {
+ continue;
}
- if (!nodeIds.contains(source.getSequence())) {
- nodes.add(buildNode(nodeCompMap, source));
- nodeIds.add(source.getSequence());
+ if (!nodes.containsKey(source.getSequence())) {
+ nodes.put(source.getSequence(), buildNode(source));
}
- if (!nodeIds.contains(target.getSequence())) {
- nodes.add(buildNode(nodeCompMap, target));
- nodeIds.add(target.getSequence());
+ if (!nodes.containsKey(target.getSequence())) {
+ nodes.put(target.getSequence(), buildNode(target));
+ if (BooleanUtils.valueToBoolean(target.getIsAddress())) {
+ nodes.get(target.getSequence()).setType(componentLibraryCatalogService.getServerNameBasedOnComponent(clientCall.getComponentId()));
+ }
}
String callId = source.getSequence() + Const.ID_SPLIT + target.getSequence();
@@ -90,115 +76,82 @@ Topology build(List<ServiceComponent> serviceComponents, List<ServiceMapping> se
callIds.add(callId);
Call call = new Call();
- call.setSource(source.getSequence());
- call.setTarget(target.getSequence());
- call.setCallType(nodeCompMap.get(clientCall.getTarget()));
+ call.setSource(clientCall.getSource());
+ call.setTarget(clientCall.getTarget());
call.setId(clientCall.getId());
call.setDetectPoint(DetectPoint.CLIENT);
+ call.setCallType(componentLibraryCatalogService.getComponentName(clientCall.getComponentId()));
calls.add(call);
}
- });
+ }
- serviceRelationServerCalls.forEach(serverCall -> {
+ for (Call serverCall : serviceRelationServerCalls) {
ServiceInventory source = serviceInventoryCache.get(serverCall.getSource());
ServiceInventory target = serviceInventoryCache.get(serverCall.getTarget());
if (source.getSequence() == Const.USER_SERVICE_ID) {
- if (!nodeIds.contains(source.getSequence())) {
+ if (!nodes.containsKey(source.getSequence())) {
Node visualUserNode = new Node();
visualUserNode.setId(source.getSequence());
visualUserNode.setName(Const.USER_CODE);
visualUserNode.setType(Const.USER_CODE.toUpperCase());
visualUserNode.setReal(false);
- nodes.add(visualUserNode);
- nodeIds.add(source.getSequence());
+ nodes.put(source.getSequence(), visualUserNode);
}
}
if (BooleanUtils.valueToBoolean(source.getIsAddress())) {
- if (!nodeIds.contains(source.getSequence())) {
+ if (!nodes.containsKey(source.getSequence())) {
Node conjecturalNode = new Node();
conjecturalNode.setId(source.getSequence());
conjecturalNode.setName(source.getName());
- conjecturalNode.setType(conjecturalNodeCompMap.getOrDefault(target.getSequence(), Const.UNKNOWN));
+ conjecturalNode.setType(componentLibraryCatalogService.getServerNameBasedOnComponent(serverCall.getComponentId()));
conjecturalNode.setReal(true);
- nodeIds.add(source.getSequence());
- nodes.add(conjecturalNode);
+ nodes.put(source.getSequence(), conjecturalNode);
}
}
- if (!nodeIds.contains(source.getSequence())) {
- nodes.add(buildNode(nodeCompMap, source));
- nodeIds.add(source.getSequence());
- }
-
- if (!nodeIds.contains(target.getSequence())) {
- nodes.add(buildNode(nodeCompMap, target));
- nodeIds.add(target.getSequence());
- }
-
String callId = source.getSequence() + Const.ID_SPLIT + target.getSequence();
if (!callIds.contains(callId)) {
callIds.add(callId);
Call call = new Call();
- call.setSource(source.getSequence());
- call.setTarget(target.getSequence());
+ call.setSource(serverCall.getSource());
+ call.setTarget(serverCall.getTarget());
call.setId(serverCall.getId());
call.setDetectPoint(DetectPoint.SERVER);
+ calls.add(call);
if (source.getSequence() == Const.USER_SERVICE_ID) {
call.setCallType(Const.EMPTY_STRING);
} else {
- call.setCallType(nodeCompMap.get(serverCall.getTarget()));
+ call.setCallType(componentLibraryCatalogService.getComponentName(serverCall.getComponentId()));
}
- calls.add(call);
}
- });
- Topology topology = new Topology();
- topology.getCalls().addAll(calls);
- topology.getNodes().addAll(nodes);
- return topology;
- }
-
- private void mappingIdExchange(Map<Integer, Integer> mappings, List<Call> serviceRelationCalls) {
- serviceRelationCalls.forEach(relationCall -> {
- relationCall.setSource(mappings.getOrDefault(relationCall.getSource(), relationCall.getSource()));
- relationCall.setTarget(mappings.getOrDefault(relationCall.getTarget(), relationCall.getTarget()));
- });
- }
+ if (!nodes.containsKey(source.getSequence())) {
+ nodes.put(source.getSequence(), buildNode(source));
+ }
- private Map<Integer, Integer> changeMapping2Map(List<ServiceMapping> serviceMappings) {
- Map<Integer, Integer> mappings = new HashMap<>();
- serviceMappings.forEach(serviceMapping -> mappings.put(serviceMapping.getMappingServiceId(), serviceMapping.getServiceId()));
- return mappings;
- }
+ if (!nodes.containsKey(target.getSequence())) {
+ nodes.put(target.getSequence(), buildNode(target));
+ }
- private Map<Integer, String> buildConjecturalNodeCompMap(List<ServiceComponent> serviceComponents) {
- Map<Integer, String> components = new HashMap<>();
- serviceComponents.forEach(serviceComponent -> {
- int componentServerId = this.componentLibraryCatalogService.getServerIdBasedOnComponent(serviceComponent.getComponentId());
- String componentName = this.componentLibraryCatalogService.getServerName(componentServerId);
- components.put(serviceComponent.getServiceId(), componentName);
- });
- return components;
- }
+ if (nodes.containsKey(target.getSequence())) {
+ nodes.get(target.getSequence()).setType(componentLibraryCatalogService.getComponentName(serverCall.getComponentId()));
+ }
+ }
- private Map<Integer, String> buildNodeCompMap(List<ServiceComponent> serviceComponents) {
- Map<Integer, String> components = new HashMap<>();
- serviceComponents.forEach(serviceComponent -> {
- String componentName = this.componentLibraryCatalogService.getComponentName(serviceComponent.getComponentId());
- components.put(serviceComponent.getServiceId(), componentName);
- });
- return components;
+ Topology topology = new Topology();
+ topology.getCalls().addAll(calls);
+ topology.getNodes().addAll(nodes.values());
+ return topology;
}
- private Node buildNode(Map<Integer, String> nodeCompMap, ServiceInventory serviceInventory) {
+ private Node buildNode(ServiceInventory serviceInventory) {
Node serviceNode = new Node();
serviceNode.setId(serviceInventory.getSequence());
serviceNode.setName(serviceInventory.getName());
- serviceNode.setType(nodeCompMap.getOrDefault(serviceInventory.getSequence(), Const.UNKNOWN));
if (BooleanUtils.valueToBoolean(serviceInventory.getIsAddress())) {
serviceNode.setReal(false);
} else {
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 077ba8b68..6e5dc1ddd 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
@@ -24,11 +24,12 @@
import org.apache.skywalking.oap.server.core.cache.EndpointInventoryCache;
import org.apache.skywalking.oap.server.core.config.IComponentLibraryCatalogService;
import org.apache.skywalking.oap.server.core.query.entity.*;
-import org.apache.skywalking.oap.server.core.source.*;
+import org.apache.skywalking.oap.server.core.source.DetectPoint;
import org.apache.skywalking.oap.server.core.storage.StorageModule;
import org.apache.skywalking.oap.server.core.storage.query.*;
import org.apache.skywalking.oap.server.library.module.*;
import org.apache.skywalking.oap.server.library.module.Service;
+import org.apache.skywalking.oap.server.library.util.*;
import org.slf4j.*;
/**
@@ -79,16 +80,13 @@ private EndpointInventoryCache getEndpointInventoryCache() {
public Topology getGlobalTopology(final Step step, final long startTB, final long endTB, final long startTimestamp,
final long endTimestamp) throws IOException {
logger.debug("step: {}, startTimeBucket: {}, endTimeBucket: {}", step, startTB, endTB);
- List<ServiceComponent> serviceComponents = getTopologyQueryDAO().loadServiceComponents(step, startTB, endTB);
- List<ServiceMapping> serviceMappings = getTopologyQueryDAO().loadServiceMappings(step, startTB, endTB);
-
- List<Call> serviceRelationClientCalls = getTopologyQueryDAO().loadClientSideServiceRelations(step, startTB, endTB);
List<Call> serviceRelationServerCalls = getTopologyQueryDAO().loadServerSideServiceRelations(step, startTB, endTB);
+ List<Call> serviceRelationClientCalls = getTopologyQueryDAO().loadClientSideServiceRelations(step, startTB, endTB);
- List<org.apache.skywalking.oap.server.core.query.entity.Service> serviceList = getMetadataQueryDAO().searchServices(startTimestamp, endTimestamp, null);
+ List<org.apache.skywalking.oap.server.core.query.entity.Service> serviceList = getMetadataQueryDAO().searchServices(startTimestamp, endTimestamp, Const.EMPTY_STRING);
TopologyBuilder builder = new TopologyBuilder(moduleManager);
- Topology topology = builder.build(serviceComponents, serviceMappings, serviceRelationClientCalls, serviceRelationServerCalls);
+ Topology topology = builder.build(serviceRelationClientCalls, serviceRelationServerCalls);
serviceList.forEach(service -> {
boolean contains = false;
@@ -114,37 +112,40 @@ public Topology getGlobalTopology(final Step step, final long startTB, final lon
public Topology getServiceTopology(final Step step, final long startTB, final long endTB,
final int serviceId) throws IOException {
- List<ServiceComponent> serviceComponents = getTopologyQueryDAO().loadServiceComponents(step, startTB, endTB);
- List<ServiceMapping> serviceMappings = getTopologyQueryDAO().loadServiceMappings(step, startTB, endTB);
-
- Set<Integer> serviceIds = new HashSet<>();
+ List<Integer> serviceIds = new ArrayList<>();
serviceIds.add(serviceId);
- serviceMappings.forEach(mapping -> {
- if (mapping.getServiceId() == serviceId) {
- serviceIds.add(mapping.getMappingServiceId());
- }
- });
-
- List<Integer> serviceIdList = new ArrayList<>(serviceIds);
- List<Call> serviceRelationClientCalls = getTopologyQueryDAO().loadSpecifiedClientSideServiceRelations(step, startTB, endTB, serviceIdList);
- List<Call> serviceRelationServerCalls = getTopologyQueryDAO().loadSpecifiedServerSideServiceRelations(step, startTB, endTB, serviceIdList);
+ List<Call> serviceRelationClientCalls = getTopologyQueryDAO().loadSpecifiedClientSideServiceRelations(step, startTB, endTB, serviceIds);
+ List<Call> serviceRelationServerCalls = getTopologyQueryDAO().loadSpecifiedServerSideServiceRelations(step, startTB, endTB, serviceIds);
TopologyBuilder builder = new TopologyBuilder(moduleManager);
- return builder.build(serviceComponents, serviceMappings, serviceRelationClientCalls, serviceRelationServerCalls);
+ Topology topology = builder.build(serviceRelationClientCalls, serviceRelationServerCalls);
+
+ List<Integer> sourceServiceIds = new ArrayList<>();
+ serviceRelationClientCalls.forEach(call -> sourceServiceIds.add(call.getSource()));
+ if (CollectionUtils.isNotEmpty(sourceServiceIds)) {
+ List<Call> sourceCalls = getTopologyQueryDAO().loadSpecifiedServerSideServiceRelations(step, startTB, endTB, sourceServiceIds);
+ topology.getNodes().forEach(node -> {
+ if (StringUtils.isEmpty(node.getType())) {
+ for (Call call : sourceCalls) {
+ if (node.getId() == call.getTarget()) {
+ node.setType(getComponentLibraryCatalogService().getComponentName(call.getComponentId()));
+ break;
+ }
+ }
+ }
+ });
+ }
+
+ return topology;
}
public Topology getEndpointTopology(final Step step, final long startTB, final long endTB,
final int endpointId) throws IOException {
- List<ServiceComponent> serviceComponents = getTopologyQueryDAO().loadServiceComponents(step, startTB, endTB);
-
- Map<Integer, String> components = new HashMap<>();
- serviceComponents.forEach(component -> components.put(component.getServiceId(), getComponentLibraryCatalogService().getComponentName(component.getComponentId())));
-
List<Call> serverSideCalls = getTopologyQueryDAO().loadSpecifiedDestOfServerSideEndpointRelations(step, startTB, endTB, endpointId);
serverSideCalls.forEach(call -> call.setDetectPoint(DetectPoint.SERVER));
- serverSideCalls.forEach(call -> call.setCallType(components.getOrDefault(getEndpointInventoryCache().get(call.getTarget()).getServiceId(), Const.UNKNOWN)));
+ serverSideCalls.forEach(call -> call.setCallType(Const.EMPTY_STRING));
Topology topology = new Topology();
topology.getCalls().addAll(serverSideCalls);
@@ -152,11 +153,11 @@ public Topology getEndpointTopology(final Step step, final long startTB, final l
Set<Integer> nodeIds = new HashSet<>();
serverSideCalls.forEach(call -> {
if (!nodeIds.contains(call.getSource())) {
- topology.getNodes().add(buildEndpointNode(components, call.getSource()));
+ topology.getNodes().add(buildEndpointNode(call.getSource()));
nodeIds.add(call.getSource());
}
if (!nodeIds.contains(call.getTarget())) {
- topology.getNodes().add(buildEndpointNode(components, call.getTarget()));
+ topology.getNodes().add(buildEndpointNode(call.getTarget()));
nodeIds.add(call.getTarget());
}
});
@@ -164,11 +165,11 @@ public Topology getEndpointTopology(final Step step, final long startTB, final l
return topology;
}
- private Node buildEndpointNode(Map<Integer, String> components, int endpointId) {
+ private Node buildEndpointNode(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.setType(Const.EMPTY_STRING);
node.setReal(true);
return node;
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Call.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Call.java
index d1d32135e..650ef1830 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Call.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/query/entity/Call.java
@@ -29,6 +29,7 @@
public class Call {
private int source;
private int target;
+ private int componentId;
private String callType;
private String id;
private DetectPoint detectPoint;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/NetworkAddressInventory.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/NetworkAddressInventory.java
index 8afe1f9b6..899779bd3 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/NetworkAddressInventory.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/NetworkAddressInventory.java
@@ -39,12 +39,8 @@
public static final String MODEL_NAME = "network_address_inventory";
private static final String NAME = "name";
- public static final String SRC_LAYER = "src_layer";
- private static final String SERVER_TYPE = "server_type";
@Setter @Getter @Column(columnName = NAME, matchQuery = true) private String name = Const.EMPTY_STRING;
- @Setter @Getter @Column(columnName = SRC_LAYER) private int srcLayer;
- @Setter @Getter @Column(columnName = SERVER_TYPE) private int serverType;
public static String buildId(String networkAddress) {
return networkAddress;
@@ -78,8 +74,6 @@ public static String buildId(String networkAddress) {
@Override public RemoteData.Builder serialize() {
RemoteData.Builder remoteBuilder = RemoteData.newBuilder();
remoteBuilder.setDataIntegers(0, getSequence());
- remoteBuilder.setDataIntegers(1, getSrcLayer());
- remoteBuilder.setDataIntegers(2, getServerType());
remoteBuilder.setDataLongs(0, getRegisterTime());
remoteBuilder.setDataLongs(1, getHeartbeatTime());
@@ -90,8 +84,6 @@ public static String buildId(String networkAddress) {
@Override public void deserialize(RemoteData remoteData) {
setSequence(remoteData.getDataIntegers(0));
- setSrcLayer(remoteData.getDataIntegers(1));
- setServerType(remoteData.getDataIntegers(2));
setRegisterTime(remoteData.getDataLongs(0));
setHeartbeatTime(remoteData.getDataLongs(1));
@@ -109,8 +101,6 @@ public static String buildId(String networkAddress) {
NetworkAddressInventory inventory = new NetworkAddressInventory();
inventory.setSequence((Integer)dbMap.get(SEQUENCE));
inventory.setName((String)dbMap.get(NAME));
- inventory.setSrcLayer((Integer)dbMap.get(SRC_LAYER));
- inventory.setServerType((Integer)dbMap.get(SERVER_TYPE));
inventory.setRegisterTime((Long)dbMap.get(REGISTER_TIME));
inventory.setHeartbeatTime((Long)dbMap.get(HEARTBEAT_TIME));
return inventory;
@@ -120,8 +110,6 @@ public static String buildId(String networkAddress) {
Map<String, Object> map = new HashMap<>();
map.put(SEQUENCE, storageData.getSequence());
map.put(NAME, storageData.getName());
- map.put(SRC_LAYER, storageData.getSrcLayer());
- map.put(SERVER_TYPE, storageData.getServerType());
map.put(REGISTER_TIME, storageData.getRegisterTime());
map.put(HEARTBEAT_TIME, storageData.getHeartbeatTime());
return map;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java
index c7b2166b5..7a497c1c7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/RegisterSource.java
@@ -36,7 +36,7 @@
@Getter @Setter @Column(columnName = REGISTER_TIME) private long registerTime;
@Getter @Setter @Column(columnName = HEARTBEAT_TIME) private long heartbeatTime;
- public final void combine(RegisterSource registerSource) {
+ public void combine(RegisterSource registerSource) {
if (heartbeatTime < registerSource.getHeartbeatTime()) {
heartbeatTime = registerSource.getHeartbeatTime();
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/ServiceInventory.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/ServiceInventory.java
index 7e195ba58..c6b456259 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/ServiceInventory.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/ServiceInventory.java
@@ -42,10 +42,14 @@
public static final String NAME = "name";
public static final String IS_ADDRESS = "is_address";
private static final String ADDRESS_ID = "address_id";
+ public static final String MAPPING_SERVICE_ID = "mapping_service_id";
+ public static final String MAPPING_LAST_UPDATE_TIME = "mapping_last_update_time";
@Setter @Getter @Column(columnName = NAME, matchQuery = true) private String name = Const.EMPTY_STRING;
@Setter @Getter @Column(columnName = IS_ADDRESS) private int isAddress;
@Setter @Getter @Column(columnName = ADDRESS_ID) private int addressId;
+ @Setter @Getter @Column(columnName = MAPPING_SERVICE_ID) private int mappingServiceId;
+ @Setter @Getter @Column(columnName = MAPPING_LAST_UPDATE_TIME) private long mappingLastUpdateTime;
public static String buildId(String serviceName) {
return serviceName + Const.ID_SPLIT + BooleanUtils.FALSE + Const.ID_SPLIT + Const.NONE;
@@ -95,9 +99,11 @@ public static String buildId(int addressId) {
remoteBuilder.setDataIntegers(0, getSequence());
remoteBuilder.setDataIntegers(1, isAddress);
remoteBuilder.setDataIntegers(2, addressId);
+ remoteBuilder.setDataIntegers(3, mappingServiceId);
remoteBuilder.setDataLongs(0, getRegisterTime());
remoteBuilder.setDataLongs(1, getHeartbeatTime());
+ remoteBuilder.setDataLongs(2, getMappingLastUpdateTime());
remoteBuilder.setDataStrings(0, name);
return remoteBuilder;
@@ -107,9 +113,11 @@ public static String buildId(int addressId) {
setSequence(remoteData.getDataIntegers(0));
setIsAddress(remoteData.getDataIntegers(1));
setAddressId(remoteData.getDataIntegers(2));
+ setMappingServiceId(remoteData.getDataIntegers(3));
setRegisterTime(remoteData.getDataLongs(0));
setHeartbeatTime(remoteData.getDataLongs(1));
+ setMappingLastUpdateTime(remoteData.getDataLongs(2));
setName(remoteData.getDataStrings(0));
}
@@ -118,16 +126,27 @@ public static String buildId(int addressId) {
return 0;
}
+ @Override public void combine(RegisterSource registerSource) {
+ super.combine(registerSource);
+ ServiceInventory serviceInventory = (ServiceInventory)registerSource;
+ if (Const.NONE != serviceInventory.getMappingServiceId() && serviceInventory.getMappingLastUpdateTime() >= this.getMappingLastUpdateTime()) {
+ this.mappingServiceId = serviceInventory.getMappingServiceId();
+ this.mappingLastUpdateTime = serviceInventory.getMappingLastUpdateTime();
+ }
+ }
+
public static class Builder implements StorageBuilder<ServiceInventory> {
@Override public ServiceInventory map2Data(Map<String, Object> dbMap) {
ServiceInventory inventory = new ServiceInventory();
inventory.setSequence((Integer)dbMap.get(SEQUENCE));
inventory.setIsAddress((Integer)dbMap.get(IS_ADDRESS));
+ inventory.setMappingServiceId((Integer)dbMap.get(MAPPING_SERVICE_ID));
inventory.setName((String)dbMap.get(NAME));
inventory.setAddressId((Integer)dbMap.get(ADDRESS_ID));
inventory.setRegisterTime((Long)dbMap.get(REGISTER_TIME));
inventory.setHeartbeatTime((Long)dbMap.get(HEARTBEAT_TIME));
+ inventory.setMappingLastUpdateTime((Long)dbMap.get(MAPPING_LAST_UPDATE_TIME));
return inventory;
}
@@ -135,10 +154,12 @@ public static String buildId(int addressId) {
Map<String, Object> map = new HashMap<>();
map.put(SEQUENCE, storageData.getSequence());
map.put(IS_ADDRESS, storageData.getIsAddress());
+ map.put(MAPPING_SERVICE_ID, storageData.getMappingServiceId());
map.put(NAME, storageData.getName());
map.put(ADDRESS_ID, storageData.getAddressId());
map.put(REGISTER_TIME, storageData.getRegisterTime());
map.put(HEARTBEAT_TIME, storageData.getHeartbeatTime());
+ map.put(MAPPING_LAST_UPDATE_TIME, storageData.getMappingLastUpdateTime());
return map;
}
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java
index f812f580b..49c88483b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/INetworkAddressInventoryRegister.java
@@ -28,7 +28,5 @@
int get(String networkAddress);
- void update(int addressId, int srcLayer, int serverType);
-
void heartbeat(int addressId, long heartBeatTime);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java
index a31b270d8..398149c1e 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/IServiceInventoryRegister.java
@@ -27,7 +27,9 @@
int getOrCreate(String serviceName);
- int getOrCreate(int addressId);
+ int getOrCreate(int addressId, String serviceName);
void heartbeat(int serviceId, long heartBeatTime);
+
+ void updateMapping(int serviceId, int mappingServiceId);
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java
index e702d673c..3cf548198 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/NetworkAddressInventoryRegister.java
@@ -26,7 +26,7 @@
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.slf4j.*;
-import static java.util.Objects.*;
+import static java.util.Objects.isNull;
/**
* @author peng-yongsheng
@@ -69,7 +69,7 @@ private IServiceInstanceInventoryRegister getServiceInstanceInventoryRegister()
int addressId = getNetworkAddressInventoryCache().getAddressId(networkAddress);
if (addressId != Const.NONE) {
- int serviceId = getServiceInventoryRegister().getOrCreate(addressId);
+ int serviceId = getServiceInventoryRegister().getOrCreate(addressId, networkAddress);
if (serviceId != Const.NONE) {
int serviceInstanceId = getServiceInstanceInventoryRegister().getOrCreate(serviceId, addressId, System.currentTimeMillis());
@@ -96,26 +96,6 @@ private IServiceInstanceInventoryRegister getServiceInstanceInventoryRegister()
return getNetworkAddressInventoryCache().getAddressId(networkAddress);
}
- @Override public void update(int addressId, int srcLayer, int serverType) {
- if (!this.compare(addressId, srcLayer, serverType)) {
- NetworkAddressInventory newNetworkAddress = getNetworkAddressInventoryCache().get(addressId);
- newNetworkAddress.setSrcLayer(srcLayer);
- newNetworkAddress.setServerType(serverType);
- newNetworkAddress.setHeartbeatTime(System.currentTimeMillis());
-
- InventoryProcess.INSTANCE.in(newNetworkAddress);
- }
- }
-
- private boolean compare(int addressId, int srcLayer, int serverType) {
- NetworkAddressInventory networkAddress = getNetworkAddressInventoryCache().get(addressId);
-
- if (nonNull(networkAddress)) {
- return srcLayer == networkAddress.getSrcLayer() && serverType == networkAddress.getServerType();
- }
- return true;
- }
-
@Override public void heartbeat(int addressId, long heartBeatTime) {
NetworkAddressInventory networkAddress = getNetworkAddressInventoryCache().get(addressId);
if (Objects.nonNull(networkAddress)) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
index 62e8ab6a4..f9a0b78b7 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/register/service/ServiceInventoryRegister.java
@@ -62,24 +62,26 @@ private ServiceInventoryCache getServiceInventoryCache() {
long now = System.currentTimeMillis();
serviceInventory.setRegisterTime(now);
serviceInventory.setHeartbeatTime(now);
+ serviceInventory.setMappingLastUpdateTime(now);
InventoryProcess.INSTANCE.in(serviceInventory);
}
return serviceId;
}
- @Override public int getOrCreate(int addressId) {
+ @Override public int getOrCreate(int addressId, String serviceName) {
int serviceId = getServiceInventoryCache().getServiceId(addressId);
if (serviceId == Const.NONE) {
ServiceInventory serviceInventory = new ServiceInventory();
- serviceInventory.setName(Const.EMPTY_STRING);
+ serviceInventory.setName(serviceName);
serviceInventory.setAddressId(addressId);
serviceInventory.setIsAddress(BooleanUtils.TRUE);
long now = System.currentTimeMillis();
serviceInventory.setRegisterTime(now);
serviceInventory.setHeartbeatTime(now);
+ serviceInventory.setMappingLastUpdateTime(now);
InventoryProcess.INSTANCE.in(serviceInventory);
}
@@ -96,4 +98,16 @@ private ServiceInventoryCache getServiceInventoryCache() {
logger.warn("Service {} heartbeat, but not found in storage.");
}
}
+
+ @Override public void updateMapping(int serviceId, int mappingServiceId) {
+ ServiceInventory serviceInventory = getServiceInventoryCache().get(serviceId);
+ if (Objects.nonNull(serviceInventory)) {
+ serviceInventory.setMappingServiceId(mappingServiceId);
+ serviceInventory.setMappingLastUpdateTime(System.currentTimeMillis());
+
+ InventoryProcess.INSTANCE.in(serviceInventory);
+ } else {
+ logger.warn("Service {} mapping update, but not found in storage.");
+ }
+ }
}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/EndpointRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/EndpointRelation.java
index d1a025e76..534309f07 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/EndpointRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/EndpointRelation.java
@@ -31,11 +31,11 @@
}
@Override public String getEntityId() {
- return String.valueOf(endpointId) + Const.ID_SPLIT + String.valueOf(childEndpointId);
+ return String.valueOf(endpointId) + Const.ID_SPLIT + String.valueOf(childEndpointId) + Const.ID_SPLIT + String.valueOf(componentId);
}
- public static String buildEntityId(int endpointId, int childEndpointId) {
- return String.valueOf(endpointId) + Const.ID_SPLIT + String.valueOf(childEndpointId);
+ public static String buildEntityId(int endpointId, int childEndpointId, int componentId) {
+ return String.valueOf(endpointId) + Const.ID_SPLIT + String.valueOf(childEndpointId) + Const.ID_SPLIT + String.valueOf(componentId);
}
@Getter @Setter private int endpointId;
@@ -52,6 +52,7 @@ public static String buildEntityId(int endpointId, int childEndpointId) {
@Getter @Setter private int childServiceInstanceId;
@Getter @Setter private String childServiceInstanceName;
+ @Getter @Setter private int componentId;
@Getter @Setter private int rpcLatency;
@Getter @Setter private boolean status;
@Getter @Setter private int responseCode;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java
index 880395cbf..7374a334b 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/Scope.java
@@ -24,7 +24,6 @@
public enum Scope {
All, Service, ServiceInstance, Endpoint, ServiceRelation, ServiceInstanceRelation, EndpointRelation, NetworkAddress,
ServiceInstanceJVMCPU, ServiceInstanceJVMMemory, ServiceInstanceJVMMemoryPool, ServiceInstanceJVMGC,
- ServiceComponent, ServiceMapping,
Segment;
public static Scope valueOf(int ordinal) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceComponent.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceComponent.java
deleted file mode 100644
index d2f0dff22..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceComponent.java
+++ /dev/null
@@ -1,43 +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.source;
-
-import lombok.*;
-import org.apache.skywalking.oap.server.core.UnexpectedException;
-import org.apache.skywalking.oap.server.core.source.annotation.SourceType;
-
-/**
- * @author peng-yongsheng
- */
-@SourceType
-public class ServiceComponent extends Source {
-
- @Override public Scope scope() {
- return Scope.ServiceComponent;
- }
-
-
- @Override public String getEntityId() {
- throw new UnexpectedException("ServiceMapping doesn't support getEntityId");
- }
-
-
- @Getter @Setter private int serviceId;
- @Getter @Setter private int componentId;
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
index a583a48e0..07f34cb92 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java
@@ -31,7 +31,7 @@
}
@Override public String getEntityId() {
- return String.valueOf(sourceServiceInstanceId) + Const.ID_SPLIT + String.valueOf(destServiceInstanceId);
+ return String.valueOf(sourceServiceInstanceId) + Const.ID_SPLIT + String.valueOf(destServiceInstanceId) + Const.ID_SPLIT + String.valueOf(componentId);
}
@Getter @Setter private int sourceServiceInstanceId;
@@ -43,6 +43,7 @@
@Getter @Setter private String destServiceName;
@Getter @Setter private String destServiceInstanceName;
@Getter @Setter private String endpoint;
+ @Getter @Setter private int componentId;
@Getter @Setter private int latency;
@Getter @Setter private boolean status;
@Getter @Setter private int responseCode;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceMapping.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceMapping.java
deleted file mode 100644
index 83bd63a84..000000000
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceMapping.java
+++ /dev/null
@@ -1,42 +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.source;
-
-import lombok.*;
-import org.apache.skywalking.oap.server.core.UnexpectedException;
-import org.apache.skywalking.oap.server.core.source.annotation.SourceType;
-
-/**
- * @author peng-yongsheng
- */
-@SourceType
-public class ServiceMapping extends Source {
-
- @Override public Scope scope() {
- return Scope.ServiceMapping;
- }
-
-
- @Override public String getEntityId() {
- throw new UnexpectedException("ServiceMapping doesn't support getEntityId");
- }
-
- @Getter @Setter private int serviceId;
- @Getter @Setter private int mappingServiceId;
-}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
index bed3929eb..9ac904b2e 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java
@@ -31,11 +31,11 @@
}
@Override public String getEntityId() {
- return String.valueOf(sourceServiceId) + Const.ID_SPLIT + String.valueOf(destServiceId);
+ return String.valueOf(sourceServiceId) + Const.ID_SPLIT + String.valueOf(destServiceId) + Const.ID_SPLIT + String.valueOf(componentId);
}
- public static String buildEntityId(int sourceServiceId, int destServiceId) {
- return String.valueOf(sourceServiceId) + Const.ID_SPLIT + String.valueOf(destServiceId);
+ public static String buildEntityId(int sourceServiceId, int destServiceId, int componentId) {
+ return String.valueOf(sourceServiceId) + Const.ID_SPLIT + String.valueOf(destServiceId) + Const.ID_SPLIT + String.valueOf(componentId);
}
@Getter @Setter private int sourceServiceId;
@@ -45,6 +45,7 @@ public static String buildEntityId(int sourceServiceId, int destServiceId) {
@Getter @Setter private String destServiceName;
@Getter @Setter private String destServiceInstanceName;
@Getter @Setter private String endpoint;
+ @Getter @Setter private int componentId;
@Getter @Setter private int latency;
@Getter @Setter private boolean status;
@Getter @Setter private int responseCode;
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/cache/IServiceInventoryCacheDAO.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/cache/IServiceInventoryCacheDAO.java
index 63e717f60..a1ea36756 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/cache/IServiceInventoryCacheDAO.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/storage/cache/IServiceInventoryCacheDAO.java
@@ -18,6 +18,7 @@
package org.apache.skywalking.oap.server.core.storage.cache;
+import java.util.List;
import org.apache.skywalking.oap.server.core.register.ServiceInventory;
import org.apache.skywalking.oap.server.core.storage.DAO;
@@ -31,4 +32,6 @@
int getServiceId(int addressId);
ServiceInventory get(int serviceId);
+
+ List<ServiceInventory> loadLastMappingUpdate();
}
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 440b8b3a8..b17a7eb87 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
@@ -21,7 +21,6 @@
import java.io.IOException;
import java.util.List;
import org.apache.skywalking.oap.server.core.query.entity.*;
-import org.apache.skywalking.oap.server.core.source.*;
import org.apache.skywalking.oap.server.library.module.Service;
/**
@@ -39,10 +38,6 @@
List<Call> loadClientSideServiceRelations(Step step, long startTB, long endTB) throws IOException;
- List<ServiceMapping> loadServiceMappings(Step step, long startTB, long endTB) throws IOException;
-
- List<ServiceComponent> loadServiceComponents(Step step, long startTB, long endTB) throws IOException;
-
List<Call> loadSpecifiedDestOfServerSideEndpointRelations(Step step, long startTB, long endTB,
int destEndpointId) throws IOException;
}
diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogFileTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogFileTest.java
index 0f73384d2..0c73715ae 100644
--- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogFileTest.java
+++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/config/ComponentLibraryCatalogFileTest.java
@@ -31,6 +31,6 @@ public void testInitAndSettings() {
Assert.assertEquals(1, service.getComponentId("Tomcat"));
Assert.assertEquals(7, service.getServerIdBasedOnComponent(30));
Assert.assertEquals(21, service.getServerIdBasedOnComponent(21));
- Assert.assertEquals("Jedis", service.getServerName(30));
+ Assert.assertEquals("Redis", service.getServerNameBasedOnComponent(30));
}
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
index c5b831aa8..0a2a739e1 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
@@ -28,7 +28,7 @@
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.*;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.endpoint.MultiScopesSpanListener;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.segment.SegmentSpanListener;
-import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.service.*;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.service.ServiceMappingSpanListener;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker;
/**
@@ -60,7 +60,6 @@ public TraceModuleProvider() {
@Override public void start() throws ModuleStartException {
SegmentParserListenerManager listenerManager = new SegmentParserListenerManager();
listenerManager.add(new MultiScopesSpanListener.Factory());
- listenerManager.add(new ServiceComponentSpanListener.Factory());
listenerManager.add(new ServiceMappingSpanListener.Factory());
listenerManager.add(new SegmentSpanListener.Factory());
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 4e849fa83..b6f2e85a4 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
@@ -31,10 +31,9 @@
import static java.util.Objects.nonNull;
/**
- *
* Notice, in here, there are following concepts match
*
- * v5 | v6
+ * v5 | v6
*
* 1. Application == Service
* 2. Server == Service Instance
@@ -92,6 +91,7 @@ public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreI
sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getApplicationInstanceId());
sourceBuilder.setDestServiceId(segmentCoreInfo.getApplicationId());
sourceBuilder.setDetectPoint(DetectPoint.SERVER);
+ sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
entrySourceBuilders.add(sourceBuilder);
}
@@ -104,6 +104,7 @@ public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreI
sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getApplicationInstanceId());
sourceBuilder.setDestServiceId(segmentCoreInfo.getApplicationId());
sourceBuilder.setDetectPoint(DetectPoint.SERVER);
+ sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
entrySourceBuilders.add(sourceBuilder);
@@ -120,6 +121,7 @@ public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreI
int peerId = spanDecorator.getPeerId();
int destServiceId = serviceInventoryCache.getServiceId(peerId);
+ int mappingServiceId = serviceInventoryCache.get(destServiceId).getMappingServiceId();
int destInstanceId = instanceInventoryCache.getServiceInstanceId(destServiceId, peerId);
sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID);
@@ -127,8 +129,13 @@ public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreI
sourceBuilder.setSourceServiceId(segmentCoreInfo.getApplicationId());
sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId());
sourceBuilder.setDestServiceInstanceId(destInstanceId);
- sourceBuilder.setDestServiceId(destServiceId);
+ if (Const.NONE == mappingServiceId) {
+ sourceBuilder.setDestServiceId(destServiceId);
+ } else {
+ sourceBuilder.setDestServiceId(mappingServiceId);
+ }
sourceBuilder.setDetectPoint(DetectPoint.CLIENT);
+ sourceBuilder.setComponentId(spanDecorator.getComponentId());
setPublicAttrs(sourceBuilder, spanDecorator);
exitSourceBuilders.add(sourceBuilder);
}
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/SourceBuilder.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/SourceBuilder.java
index 3b3200895..613245821 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/SourceBuilder.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/SourceBuilder.java
@@ -38,6 +38,7 @@
@Getter @Setter private String destServiceInstanceName;
@Getter @Setter private int destEndpointId;
@Getter @Setter private String destEndpointName;
+ @Getter @Setter private int componentId;
@Getter @Setter private int latency;
@Getter @Setter private boolean status;
@Getter @Setter private int responseCode;
@@ -81,6 +82,7 @@ ServiceRelation toServiceRelation() {
serviceRelation.setDestServiceName(destServiceName);
serviceRelation.setDestServiceInstanceName(destServiceInstanceName);
serviceRelation.setEndpoint(destEndpointName);
+ serviceRelation.setComponentId(componentId);
serviceRelation.setLatency(latency);
serviceRelation.setStatus(status);
serviceRelation.setResponseCode(responseCode);
@@ -116,6 +118,7 @@ ServiceInstanceRelation toServiceInstanceRelation() {
serviceInstanceRelation.setDestServiceName(destServiceName);
serviceInstanceRelation.setDestServiceInstanceName(destServiceInstanceName);
serviceInstanceRelation.setEndpoint(destEndpointName);
+ serviceInstanceRelation.setComponentId(componentId);
serviceInstanceRelation.setLatency(latency);
serviceInstanceRelation.setStatus(status);
serviceInstanceRelation.setResponseCode(responseCode);
@@ -155,6 +158,7 @@ EndpointRelation toEndpointRelation() {
endpointRelation.setChildServiceName(destServiceName);
endpointRelation.setChildServiceInstanceId(destServiceInstanceId);
endpointRelation.setChildServiceInstanceName(destServiceInstanceName);
+ endpointRelation.setComponentId(componentId);
endpointRelation.setRpcLatency(latency);
endpointRelation.setStatus(status);
endpointRelation.setResponseCode(responseCode);
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/service/ServiceComponentSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/service/ServiceComponentSpanListener.java
deleted file mode 100644
index cf296e5fd..000000000
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/service/ServiceComponentSpanListener.java
+++ /dev/null
@@ -1,77 +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.receiver.trace.provider.parser.listener.service;
-
-import java.util.*;
-import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache;
-import org.apache.skywalking.oap.server.core.source.*;
-import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*;
-import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*;
-
-/**
- * @author peng-yongsheng
- */
-public class ServiceComponentSpanListener implements EntrySpanListener, ExitSpanListener {
-
- private final SourceReceiver sourceReceiver;
- private final ServiceInventoryCache serviceInventoryCache;
- private final List<ServiceComponent> serviceComponents = new LinkedList<>();
-
- private ServiceComponentSpanListener(ModuleManager moduleManager) {
- this.sourceReceiver = moduleManager.find(CoreModule.NAME).getService(SourceReceiver.class);
- this.serviceInventoryCache = moduleManager.find(CoreModule.NAME).getService(ServiceInventoryCache.class);
- }
-
- @Override public boolean containsPoint(Point point) {
- return Point.Entry.equals(point) || Point.Exit.equals(point);
- }
-
- @Override
- public void parseExit(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
- int serviceIdByPeerId = serviceInventoryCache.getServiceId(spanDecorator.getPeerId());
-
- ServiceComponent serviceComponent = new ServiceComponent();
- serviceComponent.setServiceId(serviceIdByPeerId);
- serviceComponent.setComponentId(spanDecorator.getComponentId());
- serviceComponent.setTimeBucket(segmentCoreInfo.getMinuteTimeBucket());
- serviceComponents.add(serviceComponent);
- }
-
- @Override
- public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreInfo) {
- ServiceComponent serviceComponent = new ServiceComponent();
- serviceComponent.setServiceId(segmentCoreInfo.getApplicationId());
- serviceComponent.setComponentId(spanDecorator.getComponentId());
- serviceComponent.setTimeBucket(segmentCoreInfo.getMinuteTimeBucket());
- serviceComponents.add(serviceComponent);
- }
-
- @Override public void build() {
- serviceComponents.forEach(sourceReceiver::receive);
- }
-
- public static class Factory implements SpanListenerFactory {
-
- @Override public SpanListener create(ModuleManager moduleManager) {
- return new ServiceComponentSpanListener(moduleManager);
- }
- }
-}
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/service/ServiceMappingSpanListener.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/service/ServiceMappingSpanListener.java
index a9bc5a5dd..a6ca9eed4 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/service/ServiceMappingSpanListener.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/listener/service/ServiceMappingSpanListener.java
@@ -19,10 +19,11 @@
package org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.service;
import java.util.*;
+import lombok.*;
import org.apache.skywalking.apm.network.language.agent.SpanLayer;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache;
-import org.apache.skywalking.oap.server.core.source.*;
+import org.apache.skywalking.oap.server.core.register.service.IServiceInventoryRegister;
import org.apache.skywalking.oap.server.library.module.ModuleManager;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.decorator.*;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.*;
@@ -35,13 +36,13 @@
private static final Logger logger = LoggerFactory.getLogger(ServiceMappingSpanListener.class);
- private final SourceReceiver sourceReceiver;
+ private final IServiceInventoryRegister serviceInventoryRegister;
private final ServiceInventoryCache serviceInventoryCache;
private List<ServiceMapping> serviceMappings = new LinkedList<>();
private ServiceMappingSpanListener(ModuleManager moduleManager) {
- this.sourceReceiver = moduleManager.find(CoreModule.NAME).getService(SourceReceiver.class);
this.serviceInventoryCache = moduleManager.find(CoreModule.NAME).getService(ServiceInventoryCache.class);
+ this.serviceInventoryRegister = moduleManager.find(CoreModule.NAME).getService(IServiceInventoryRegister.class);
}
@Override public boolean containsPoint(Point point) {
@@ -56,25 +57,26 @@ private ServiceMappingSpanListener(ModuleManager moduleManager) {
if (!spanDecorator.getSpanLayer().equals(SpanLayer.MQ)) {
if (spanDecorator.getRefsCount() > 0) {
for (int i = 0; i < spanDecorator.getRefsCount(); i++) {
- ServiceMapping serviceMapping = new ServiceMapping();
- serviceMapping.setServiceId(segmentCoreInfo.getApplicationId());
-
- int addressId = spanDecorator.getRefs(i).getNetworkAddressId();
- int mappingServiceId = serviceInventoryCache.getServiceId(addressId);
- serviceMapping.setMappingServiceId(mappingServiceId);
- serviceMapping.setTimeBucket(segmentCoreInfo.getMinuteTimeBucket());
- serviceMappings.add(serviceMapping);
+ int serviceId = serviceInventoryCache.getServiceId(spanDecorator.getRefs(i).getNetworkAddressId());
+ int mappingServiceId = serviceInventoryCache.get(serviceId).getMappingServiceId();
+ if (mappingServiceId != segmentCoreInfo.getApplicationId()) {
+ ServiceMapping serviceMapping = new ServiceMapping();
+ serviceMapping.setServiceId(serviceId);
+ serviceMapping.setMappingServiceId(segmentCoreInfo.getApplicationId());
+ serviceMappings.add(serviceMapping);
+ }
}
}
}
}
@Override public void build() {
- if (logger.isDebugEnabled()) {
- logger.debug("service mapping listener build");
- }
-
- serviceMappings.forEach(sourceReceiver::receive);
+ serviceMappings.forEach(serviceMapping -> {
+ if (logger.isDebugEnabled()) {
+ logger.debug("service mapping listener build, service id: {}, mapping service id: {}", serviceMapping.getServiceId(), serviceMapping.getMappingServiceId());
+ }
+ serviceInventoryRegister.updateMapping(serviceMapping.getServiceId(), serviceMapping.getMappingServiceId());
+ });
}
public static class Factory implements SpanListenerFactory {
@@ -83,4 +85,11 @@ private ServiceMappingSpanListener(ModuleManager moduleManager) {
return new ServiceMappingSpanListener(moduleManager);
}
}
+
+ @Setter
+ @Getter
+ private class ServiceMapping {
+ private int serviceId;
+ private int mappingServiceId;
+ }
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SpanIdExchanger.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SpanIdExchanger.java
index 36d399154..98e9c9eb1 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SpanIdExchanger.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/parser/standardization/SpanIdExchanger.java
@@ -80,10 +80,6 @@ private SpanIdExchanger(ModuleManager moduleManager) {
standardBuilder.toBuilder();
standardBuilder.setPeerId(peerId);
standardBuilder.setPeer(Const.EMPTY_STRING);
-
- int spanLayer = standardBuilder.getSpanLayerValue();
- int serverType = componentLibraryCatalogService.getServerIdBasedOnComponent(standardBuilder.getComponentId());
- networkAddressInventoryRegister.update(peerId, spanLayer, serverType);
}
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java
index a96e56b50..c112452ff 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/AgentDataMock.java
@@ -22,43 +22,54 @@
import io.grpc.stub.StreamObserver;
import java.util.concurrent.TimeUnit;
import org.apache.skywalking.apm.network.language.agent.*;
-import org.slf4j.*;
/**
* @author peng-yongsheng
*/
public class AgentDataMock {
- private static final Logger logger = LoggerFactory.getLogger(AgentDataMock.class);
-
private static boolean IS_COMPLETED = false;
public static void main(String[] args) throws InterruptedException {
ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).usePlaintext(true).build();
- RegisterMock registerMock = new RegisterMock();
- registerMock.mock(channel);
+ RegisterMock registerMock = new RegisterMock(channel);
StreamObserver<UpstreamSegment> streamObserver = createStreamObserver();
UniqueId.Builder globalTraceId = UniqueIdBuilder.INSTANCE.create();
long startTimestamp = System.currentTimeMillis();
- ConsumerMock consumerMock = new ConsumerMock();
- UniqueId.Builder consumerSegmentId = UniqueIdBuilder.INSTANCE.create();
- consumerMock.mock(streamObserver, globalTraceId, consumerSegmentId, startTimestamp, true);
+ // ServiceAMock
+ ServiceAMock serviceAMock = new ServiceAMock(registerMock);
+ serviceAMock.register();
+
+ UniqueId.Builder serviceASegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, true);
+
+ // ServiceBMock
+ ServiceBMock serviceBMock = new ServiceBMock(registerMock);
+ serviceBMock.register();
+
+ UniqueId.Builder serviceBSegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, true);
+
+ // ServiceCMock
+ ServiceCMock serviceCMock = new ServiceCMock(registerMock);
+ serviceCMock.register();
- ProviderMock providerMock = new ProviderMock();
- UniqueId.Builder providerSegmentId = UniqueIdBuilder.INSTANCE.create();
- providerMock.mock(streamObserver, globalTraceId, providerSegmentId, consumerSegmentId, startTimestamp, true);
+ UniqueId.Builder serviceCSegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, true);
TimeUnit.SECONDS.sleep(10);
globalTraceId = UniqueIdBuilder.INSTANCE.create();
- consumerSegmentId = UniqueIdBuilder.INSTANCE.create();
- providerSegmentId = UniqueIdBuilder.INSTANCE.create();
- consumerMock.mock(streamObserver, globalTraceId, consumerSegmentId, startTimestamp, false);
- providerMock.mock(streamObserver, globalTraceId, providerSegmentId, consumerSegmentId, startTimestamp, false);
+ serviceASegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceBSegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceCSegmentId = UniqueIdBuilder.INSTANCE.create();
+ serviceAMock.mock(streamObserver, globalTraceId, serviceASegmentId, startTimestamp, false);
+ serviceBMock.mock(streamObserver, globalTraceId, serviceBSegmentId, serviceASegmentId, startTimestamp, false);
+ serviceCMock.mock(streamObserver, globalTraceId, serviceCSegmentId, serviceBSegmentId, startTimestamp, false);
streamObserver.onCompleted();
while (!IS_COMPLETED) {
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/RegisterMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/RegisterMock.java
index b9514c9ec..5ba649e76 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/RegisterMock.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/RegisterMock.java
@@ -19,10 +19,8 @@
package org.apache.skywalking.oap.server.receiver.trace.mock;
import io.grpc.ManagedChannel;
-import java.util.concurrent.*;
+import java.util.concurrent.TimeUnit;
import org.apache.skywalking.apm.network.language.agent.*;
-import org.apache.skywalking.apm.util.RunnableWithExceptionProtection;
-import org.joda.time.DateTime;
import org.slf4j.*;
/**
@@ -32,37 +30,37 @@
private static final Logger logger = LoggerFactory.getLogger(RegisterMock.class);
- private ApplicationRegisterServiceGrpc.ApplicationRegisterServiceBlockingStub applicationRegisterServiceBlockingStub;
- private InstanceDiscoveryServiceGrpc.InstanceDiscoveryServiceBlockingStub instanceDiscoveryServiceBlockingStub;
- private ServiceNameDiscoveryServiceGrpc.ServiceNameDiscoveryServiceBlockingStub serviceNameDiscoveryServiceBlockingStub;
+ private final ApplicationRegisterServiceGrpc.ApplicationRegisterServiceBlockingStub applicationRegisterServiceBlockingStub;
+ private final InstanceDiscoveryServiceGrpc.InstanceDiscoveryServiceBlockingStub instanceDiscoveryServiceBlockingStub;
- void mock(ManagedChannel channel) throws InterruptedException {
+ RegisterMock(ManagedChannel channel) {
applicationRegisterServiceBlockingStub = ApplicationRegisterServiceGrpc.newBlockingStub(channel);
instanceDiscoveryServiceBlockingStub = InstanceDiscoveryServiceGrpc.newBlockingStub(channel);
- serviceNameDiscoveryServiceBlockingStub = ServiceNameDiscoveryServiceGrpc.newBlockingStub(channel);
- registerConsumer();
- registerProvider();
}
- private void registerConsumer() throws InterruptedException {
+ int registerService(String serviceName) throws InterruptedException {
Application.Builder application = Application.newBuilder();
- application.setApplicationCode("dubbox-consumer");
+ application.setApplicationCode(serviceName);
ApplicationMapping applicationMapping;
do {
applicationMapping = applicationRegisterServiceBlockingStub.applicationCodeRegister(application.build());
- logger.debug("application id: {}", applicationMapping.getApplication().getValue());
+ logger.debug("service id: {}", applicationMapping.getApplication().getValue());
TimeUnit.MILLISECONDS.sleep(20);
}
while (applicationMapping.getApplication().getValue() == 0);
+ return applicationMapping.getApplication().getValue();
+ }
+
+ int registerServiceInstance(int serviceId, String agentName) throws InterruptedException {
ApplicationInstance.Builder instance = ApplicationInstance.newBuilder();
- instance.setApplicationId(applicationMapping.getApplication().getValue());
- instance.setAgentUUID("pengys");
- instance.setRegisterTime(new DateTime("2017-01-01T00:01:01.001").getMillis());
+ instance.setApplicationId(serviceId);
+ instance.setAgentUUID(agentName);
+ instance.setRegisterTime(System.currentTimeMillis());
OSInfo.Builder osInfo = OSInfo.newBuilder();
- osInfo.setHostname("pengys");
+ osInfo.setHostname(agentName);
osInfo.setOsName("MacOS XX");
osInfo.setProcessNo(1001);
osInfo.addIpv4S("10.0.0.3");
@@ -77,89 +75,6 @@ private void registerConsumer() throws InterruptedException {
}
while (instanceMapping.getApplicationInstanceId() == 0);
- ServiceNameCollection.Builder serviceNameCollection = ServiceNameCollection.newBuilder();
- ServiceNameElement.Builder serviceNameElement = ServiceNameElement.newBuilder();
- serviceNameElement.setApplicationId(applicationMapping.getApplication().getValue());
- serviceNameElement.setServiceName("org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()");
- serviceNameElement.setSrcSpanType(SpanType.Exit);
- serviceNameCollection.addElements(serviceNameElement);
-
-// registerServiceName(serviceNameCollection);
-
-// heartBeatScheduled(instanceMapping.getApplicationInstanceId());
- }
-
- private void registerProvider() throws InterruptedException {
- Application.Builder application = Application.newBuilder();
- application.setApplicationCode("dubbox-provider");
-
- ApplicationMapping applicationMapping;
- do {
- applicationMapping = applicationRegisterServiceBlockingStub.applicationCodeRegister(application.build());
- logger.debug("application id: {}", applicationMapping.getApplication().getValue());
- Thread.sleep(20);
- }
- while (applicationMapping.getApplication().getValue() == 0);
-
- ApplicationInstance.Builder instance = ApplicationInstance.newBuilder();
- instance.setApplicationId(applicationMapping.getApplication().getValue());
- instance.setAgentUUID("peng-yongsheng");
- instance.setRegisterTime(new DateTime("2017-01-01T00:01:01.001").getMillis());
-
- OSInfo.Builder osInfo = OSInfo.newBuilder();
- osInfo.setHostname("peng-yongsheng");
- osInfo.setOsName("MacOS X");
- osInfo.setProcessNo(1000);
- osInfo.addIpv4S("10.0.0.1");
- osInfo.addIpv4S("10.0.0.2");
- instance.setOsinfo(osInfo);
-
- ApplicationInstanceMapping instanceMapping;
- do {
- instanceMapping = instanceDiscoveryServiceBlockingStub.registerInstance(instance.build());
- logger.debug("instance id: {}", instanceMapping.getApplicationInstanceId());
- Thread.sleep(20);
- }
- while (instanceMapping.getApplicationInstanceId() == 0);
-
- ServiceNameCollection.Builder serviceNameCollection = ServiceNameCollection.newBuilder();
- ServiceNameElement.Builder serviceNameElement = ServiceNameElement.newBuilder();
- serviceNameElement.setApplicationId(applicationMapping.getApplication().getValue());
- serviceNameElement.setServiceName("org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()");
- serviceNameElement.setSrcSpanType(SpanType.Entry);
- serviceNameCollection.addElements(serviceNameElement);
-
-// registerServiceName(serviceNameCollection);
-
-// heartBeatScheduled(instanceMapping.getApplicationInstanceId());
- }
-
- private void registerServiceName(ServiceNameCollection.Builder serviceNameCollection) throws InterruptedException {
- ServiceNameMappingCollection serviceNameMappingCollection;
- do {
- logger.debug("register service name: {}", serviceNameCollection.getElements(0).getServiceName());
- serviceNameMappingCollection = serviceNameDiscoveryServiceBlockingStub.discovery(serviceNameCollection.build());
- logger.debug("service name mapping collection size: {}", serviceNameMappingCollection.getElementsCount());
- if (serviceNameMappingCollection.getElementsCount() > 0) {
- logger.debug("service id: {}", serviceNameMappingCollection.getElements(0).getServiceId());
- }
- Thread.sleep(20);
- }
- while (serviceNameMappingCollection.getElementsCount() == 0 || serviceNameMappingCollection.getElements(0).getServiceId() == 0);
- }
-
- private void heartBeatScheduled(int instanceId) {
- Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(
- new RunnableWithExceptionProtection(() -> heartBeat(instanceId),
- t -> logger.error("instance heart beat scheduled error.", t)), 4, 1, TimeUnit.SECONDS);
- }
-
- private void heartBeat(int instanceId) {
- long now = System.currentTimeMillis();
- logger.debug("instance heart beat, instance id: {}, time: {}", instanceId, now);
- ApplicationInstanceHeartbeat.Builder heartbeat = ApplicationInstanceHeartbeat.newBuilder();
- heartbeat.setApplicationInstanceId(instanceId);
- heartbeat.setHeartbeatTime(now);
- instanceDiscoveryServiceBlockingStub.heartbeat(heartbeat.build());
+ return instanceMapping.getApplicationInstanceId();
}
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ConsumerMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceAMock.java
similarity index 54%
rename from oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ConsumerMock.java
rename to oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceAMock.java
index f91d4a207..4da1f80f5 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ConsumerMock.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceAMock.java
@@ -26,12 +26,28 @@
/**
* @author peng-yongsheng
*/
-class ConsumerMock {
+class ServiceAMock {
- void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder globalTraceId,
+ static String REST_ENDPOINT = "/dubbox-case/case/dubbox-rest";
+ static String DUBBO_ENDPOINT = "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()";
+ static String DUBBO_ADDRESS = "DubboIPAddress:1000";
+ private final RegisterMock registerMock;
+ private static int SERVICE_ID;
+ static int SERVICE_INSTANCE_ID;
+
+ ServiceAMock(RegisterMock registerMock) {
+ this.registerMock = registerMock;
+ }
+
+ void register() throws InterruptedException {
+ SERVICE_ID = registerMock.registerService("dubbox-consumer");
+ SERVICE_INSTANCE_ID = registerMock.registerServiceInstance(SERVICE_ID, "pengysA");
+ }
+
+ void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder traceId,
UniqueId.Builder segmentId, long startTimestamp, boolean isPrepare) {
UpstreamSegment.Builder upstreamSegment = UpstreamSegment.newBuilder();
- upstreamSegment.addGlobalTraceIds(globalTraceId);
+ upstreamSegment.addGlobalTraceIds(traceId);
upstreamSegment.setSegment(createSegment(startTimestamp, segmentId, isPrepare));
streamObserver.onNext(upstreamSegment.build());
@@ -40,14 +56,11 @@ void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder globa
private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId, boolean isPrepare) {
TraceSegmentObject.Builder segment = TraceSegmentObject.newBuilder();
segment.setTraceSegmentId(segmentId);
- segment.setApplicationId(2);
- segment.setApplicationInstanceId(2);
+ segment.setApplicationId(SERVICE_ID);
+ segment.setApplicationInstanceId(SERVICE_INSTANCE_ID);
segment.addSpans(createEntrySpan(startTimestamp, isPrepare));
segment.addSpans(createLocalSpan(startTimestamp, isPrepare));
- segment.addSpans(createMqEntrySpan(startTimestamp, isPrepare));
segment.addSpans(createExitSpan(startTimestamp, isPrepare));
- segment.addSpans(createMqEntrySpan2(startTimestamp, isPrepare));
- segment.addSpans(createExitSpan2(startTimestamp, isPrepare));
return segment.build().toByteString();
}
@@ -59,10 +72,10 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
span.setSpanLayer(SpanLayer.Http);
span.setParentSpanId(-1);
span.setStartTime(startTimestamp);
- span.setEndTime(startTimestamp + 2000);
+ span.setEndTime(startTimestamp + 6000);
span.setComponentId(ComponentsDefine.TOMCAT.getId());
if (isPrepare) {
- span.setOperationName("/dubbox-case/case/dubbox-rest");
+ span.setOperationName(REST_ENDPOINT);
} else {
span.setOperationNameId(2);
}
@@ -76,7 +89,7 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
span.setSpanType(SpanType.Local);
span.setParentSpanId(0);
span.setStartTime(startTimestamp + 100);
- span.setEndTime(startTimestamp + 1900);
+ span.setEndTime(startTimestamp + 500);
if (isPrepare) {
span.setOperationName("org.apache.skywalking.Local.do");
} else {
@@ -86,77 +99,21 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
return span;
}
- private SpanObject.Builder createMqEntrySpan(long startTimestamp, boolean isPrepare) {
- SpanObject.Builder span = SpanObject.newBuilder();
- span.setSpanId(2);
- span.setSpanType(SpanType.Entry);
- span.setSpanLayer(SpanLayer.MQ);
- span.setParentSpanId(1);
- span.setStartTime(startTimestamp + 110);
- span.setEndTime(startTimestamp + 1800);
- span.setComponentId(ComponentsDefine.ROCKET_MQ_CONSUMER.getId());
- if (isPrepare) {
- span.setOperationName("org.apache.skywalking.RocketMQ");
- } else {
- span.setOperationNameId(4);
- }
- span.setIsError(false);
- return span;
- }
-
private SpanObject.Builder createExitSpan(long startTimestamp, boolean isPrepare) {
SpanObject.Builder span = SpanObject.newBuilder();
- span.setSpanId(3);
+ span.setSpanId(2);
span.setSpanType(SpanType.Exit);
span.setSpanLayer(SpanLayer.RPCFramework);
- span.setParentSpanId(2);
- span.setStartTime(startTimestamp + 120);
- span.setEndTime(startTimestamp + 1780);
- span.setComponentId(ComponentsDefine.DUBBO.getId());
- if (isPrepare) {
- span.setPeer("172.25.0.4:20880");
- span.setOperationName("org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()");
- } else {
- span.setOperationNameId(5);
- span.setPeerId(3);
- }
- span.setIsError(false);
- return span;
- }
-
- private SpanObject.Builder createMqEntrySpan2(long startTimestamp, boolean isPrepare) {
- SpanObject.Builder span = SpanObject.newBuilder();
- span.setSpanId(4);
- span.setSpanType(SpanType.Entry);
- span.setSpanLayer(SpanLayer.MQ);
span.setParentSpanId(1);
- span.setStartTime(startTimestamp + 110);
- span.setEndTime(startTimestamp + 1800);
- span.setComponentId(ComponentsDefine.ROCKET_MQ_CONSUMER.getId());
- if (isPrepare) {
- span.setOperationName("org.apache.skywalking.RocketMQ");
- } else {
- span.setOperationNameId(4);
- }
- span.setIsError(false);
- return span;
- }
-
- private SpanObject.Builder createExitSpan2(long startTimestamp, boolean isPrepare) {
- SpanObject.Builder span = SpanObject.newBuilder();
- span.setSpanId(5);
- span.setSpanType(SpanType.Exit);
- span.setSpanLayer(SpanLayer.RPCFramework);
- span.setParentSpanId(4);
span.setStartTime(startTimestamp + 120);
- span.setEndTime(startTimestamp + 1780);
+ span.setEndTime(startTimestamp + 5800);
span.setComponentId(ComponentsDefine.DUBBO.getId());
if (isPrepare) {
- span.setPeer("172.25.0.4:20880");
- span.setOperationName("org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()");
+ span.setPeer(DUBBO_ADDRESS);
+ span.setOperationName(DUBBO_ENDPOINT);
} else {
- span.setOperationNameId(5);
- span.setPeerId(3);
+ span.setPeerId(2);
+ span.setOperationNameId(6);
}
span.setIsError(false);
return span;
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ProviderMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java
similarity index 63%
rename from oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ProviderMock.java
rename to oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java
index 42bb3b168..66023b4c7 100644
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ProviderMock.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceBMock.java
@@ -26,12 +26,27 @@
/**
* @author peng-yongsheng
*/
-class ProviderMock {
+class ServiceBMock {
- void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder globalTraceId,
+ private final RegisterMock registerMock;
+ private static int SERVICE_ID;
+ static int SERVICE_INSTANCE_ID;
+ static String ROCKET_MQ_ENDPOINT = "org.apache.skywalking.RocketMQ";
+ static String ROCKET_MQ_ADDRESS = "RocketMQAddress:2000";
+
+ ServiceBMock(RegisterMock registerMock) {
+ this.registerMock = registerMock;
+ }
+
+ void register() throws InterruptedException {
+ SERVICE_ID = registerMock.registerService("dubbox-provider");
+ SERVICE_INSTANCE_ID = registerMock.registerServiceInstance(SERVICE_ID, "pengysB");
+ }
+
+ void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder traceId,
UniqueId.Builder segmentId, UniqueId.Builder parentTraceSegmentId, long startTimestamp, boolean isPrepare) {
UpstreamSegment.Builder upstreamSegment = UpstreamSegment.newBuilder();
- upstreamSegment.addGlobalTraceIds(globalTraceId);
+ upstreamSegment.addGlobalTraceIds(traceId);
upstreamSegment.setSegment(createSegment(startTimestamp, segmentId, parentTraceSegmentId, isPrepare));
streamObserver.onNext(upstreamSegment.build());
@@ -41,10 +56,11 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
UniqueId.Builder parentTraceSegmentId, boolean isPrepare) {
TraceSegmentObject.Builder segment = TraceSegmentObject.newBuilder();
segment.setTraceSegmentId(segmentId);
- segment.setApplicationId(3);
- segment.setApplicationInstanceId(3);
- segment.addSpans(createExitSpan(startTimestamp, isPrepare));
+ segment.setApplicationId(SERVICE_ID);
+ segment.setApplicationInstanceId(SERVICE_INSTANCE_ID);
segment.addSpans(createEntrySpan(startTimestamp, parentTraceSegmentId, isPrepare));
+ segment.addSpans(createExitSpan(startTimestamp, isPrepare));
+ segment.addSpans(createMQExitSpan(startTimestamp, isPrepare));
return segment.build().toByteString();
}
@@ -52,31 +68,51 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
private TraceSegmentReference.Builder createReference(UniqueId.Builder parentTraceSegmentId, boolean isPrepare) {
TraceSegmentReference.Builder reference = TraceSegmentReference.newBuilder();
reference.setParentTraceSegmentId(parentTraceSegmentId);
- reference.setParentApplicationInstanceId(2);
- reference.setParentSpanId(1);
- reference.setEntryApplicationInstanceId(2);
+ reference.setParentApplicationInstanceId(ServiceAMock.SERVICE_INSTANCE_ID);
+ reference.setParentSpanId(2);
+ reference.setEntryApplicationInstanceId(ServiceAMock.SERVICE_INSTANCE_ID);
reference.setRefType(RefType.CrossProcess);
if (isPrepare) {
- reference.setParentServiceName("/dubbox-case/case/dubbox-rest");
- reference.setNetworkAddress("172.25.0.4:20880");
- reference.setEntryServiceName("/dubbox-case/case/dubbox-rest");
+ reference.setParentServiceName(ServiceAMock.REST_ENDPOINT);
+ reference.setNetworkAddress(ServiceAMock.DUBBO_ADDRESS);
+ reference.setEntryServiceName(ServiceAMock.REST_ENDPOINT);
} else {
reference.setParentServiceId(2);
- reference.setNetworkAddressId(3);
+ reference.setNetworkAddressId(2);
reference.setEntryServiceId(2);
}
return reference;
}
+ private SpanObject.Builder createEntrySpan(long startTimestamp, UniqueId.Builder uniqueId, boolean isPrepare) {
+ SpanObject.Builder span = SpanObject.newBuilder();
+ span.setSpanId(0);
+ span.setSpanType(SpanType.Entry);
+ span.setSpanLayer(SpanLayer.RPCFramework);
+ span.setParentSpanId(-1);
+ span.setStartTime(startTimestamp + 500);
+ span.setEndTime(startTimestamp + 5000);
+ span.setComponentId(ComponentsDefine.DUBBO.getId());
+ span.setIsError(false);
+ span.addRefs(createReference(uniqueId, isPrepare));
+
+ if (isPrepare) {
+ span.setOperationName(ServiceAMock.DUBBO_ENDPOINT);
+ } else {
+ span.setOperationNameId(4);
+ }
+ return span;
+ }
+
private SpanObject.Builder createExitSpan(long startTimestamp, boolean isPrepare) {
SpanObject.Builder span = SpanObject.newBuilder();
span.setSpanId(1);
span.setSpanType(SpanType.Exit);
span.setSpanLayer(SpanLayer.Database);
span.setParentSpanId(0);
- span.setStartTime(startTimestamp + 510);
- span.setEndTime(startTimestamp + 1490);
+ span.setStartTime(startTimestamp + 550);
+ span.setEndTime(startTimestamp + 1000);
span.setComponentId(ComponentsDefine.MONGO_DRIVER.getId());
span.setIsError(true);
@@ -84,28 +120,29 @@ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId
span.setOperationName("mongodb://[username:password@]host1[:port1][,host2[:port2],...[,hostN[:portN]]][/[database][?options]]");
span.setPeer("localhost:27017");
} else {
- span.setOperationNameId(6);
- span.setPeerId(2);
+ span.setOperationNameId(7);
+ span.setPeerId(3);
}
return span;
}
- private SpanObject.Builder createEntrySpan(long startTimestamp, UniqueId.Builder uniqueId, boolean isPrepare) {
+ private SpanObject.Builder createMQExitSpan(long startTimestamp, boolean isPrepare) {
SpanObject.Builder span = SpanObject.newBuilder();
- span.setSpanId(0);
- span.setSpanType(SpanType.Entry);
- span.setSpanLayer(SpanLayer.RPCFramework);
- span.setParentSpanId(-1);
- span.setStartTime(startTimestamp + 500);
+ span.setSpanId(2);
+ span.setSpanType(SpanType.Exit);
+ span.setSpanLayer(SpanLayer.MQ);
+ span.setParentSpanId(1);
+ span.setStartTime(startTimestamp + 1100);
span.setEndTime(startTimestamp + 1500);
- span.setComponentId(ComponentsDefine.DUBBO.getId());
+ span.setComponentId(ComponentsDefine.ROCKET_MQ_PRODUCER.getId());
span.setIsError(false);
- span.addRefs(createReference(uniqueId, isPrepare));
if (isPrepare) {
- span.setOperationName("org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()");
+ span.setOperationName(ROCKET_MQ_ENDPOINT);
+ span.setPeer(ROCKET_MQ_ADDRESS);
} else {
- span.setOperationNameId(7);
+ span.setOperationNameId(8);
+ span.setPeerId(4);
}
return span;
}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java
new file mode 100644
index 000000000..2126b4d3f
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/mock/ServiceCMock.java
@@ -0,0 +1,103 @@
+/*
+ * 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.receiver.trace.mock;
+
+import com.google.protobuf.ByteString;
+import io.grpc.stub.StreamObserver;
+import org.apache.skywalking.apm.network.language.agent.*;
+import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
+
+/**
+ * @author peng-yongsheng
+ */
+class ServiceCMock {
+
+ private final RegisterMock registerMock;
+ private static int SERVICE_ID;
+ private static int SERVICE_INSTANCE_ID;
+
+ ServiceCMock(RegisterMock registerMock) {
+ this.registerMock = registerMock;
+ }
+
+ void register() throws InterruptedException {
+ SERVICE_ID = registerMock.registerService("rocket-mq-consumer");
+ SERVICE_INSTANCE_ID = registerMock.registerServiceInstance(SERVICE_ID, "pengysC");
+ }
+
+ void mock(StreamObserver<UpstreamSegment> streamObserver, UniqueId.Builder traceId,
+ UniqueId.Builder segmentId, UniqueId.Builder parentTraceSegmentId, long startTimestamp, boolean isPrepare) {
+ UpstreamSegment.Builder upstreamSegment = UpstreamSegment.newBuilder();
+ upstreamSegment.addGlobalTraceIds(traceId);
+ upstreamSegment.setSegment(createSegment(startTimestamp, segmentId, parentTraceSegmentId, isPrepare));
+
+ streamObserver.onNext(upstreamSegment.build());
+ }
+
+ private ByteString createSegment(long startTimestamp, UniqueId.Builder segmentId,
+ UniqueId.Builder parentTraceSegmentId, boolean isPrepare) {
+ TraceSegmentObject.Builder segment = TraceSegmentObject.newBuilder();
+ segment.setTraceSegmentId(segmentId);
+ segment.setApplicationInstanceId(SERVICE_INSTANCE_ID);
+ segment.setApplicationId(SERVICE_ID);
+ segment.addSpans(createEntrySpan(startTimestamp, parentTraceSegmentId, isPrepare));
+
+ return segment.build().toByteString();
+ }
+
+ private SpanObject.Builder createEntrySpan(long startTimestamp, UniqueId.Builder uniqueId, boolean isPrepare) {
+ SpanObject.Builder span = SpanObject.newBuilder();
+ span.setSpanId(0);
+ span.setSpanType(SpanType.Entry);
+ span.setSpanLayer(SpanLayer.MQ);
+ span.setParentSpanId(-1);
+ span.setStartTime(startTimestamp + 3000);
+ span.setEndTime(startTimestamp + 5000);
+ span.setComponentId(ComponentsDefine.ROCKET_MQ_CONSUMER.getId());
+ span.setIsError(false);
+ span.addRefs(createReference(uniqueId, isPrepare));
+
+ if (isPrepare) {
+ span.setOperationName(ServiceBMock.ROCKET_MQ_ENDPOINT);
+ } else {
+ span.setOperationNameId(5);
+ }
+ return span;
+ }
+
+ private TraceSegmentReference.Builder createReference(UniqueId.Builder parentTraceSegmentId, boolean isPrepare) {
+ TraceSegmentReference.Builder reference = TraceSegmentReference.newBuilder();
+ reference.setParentTraceSegmentId(parentTraceSegmentId);
+ reference.setParentApplicationInstanceId(ServiceBMock.SERVICE_INSTANCE_ID);
+ reference.setParentSpanId(2);
+ reference.setEntryApplicationInstanceId(ServiceAMock.SERVICE_INSTANCE_ID);
+ reference.setRefType(RefType.CrossProcess);
+
+ if (isPrepare) {
+ reference.setParentServiceName(ServiceBMock.ROCKET_MQ_ENDPOINT);
+ reference.setNetworkAddress(ServiceBMock.ROCKET_MQ_ADDRESS);
+ reference.setEntryServiceName(ServiceAMock.REST_ENDPOINT);
+ } else {
+ reference.setParentServiceId(8);
+ reference.setNetworkAddressId(4);
+ reference.setEntryServiceId(2);
+ }
+ return reference;
+ }
+}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/ServiceInventoryCacheEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/ServiceInventoryCacheEsDAO.java
index 8c5e2c00a..e7e1ada27 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/ServiceInventoryCacheEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/cache/ServiceInventoryCacheEsDAO.java
@@ -18,14 +18,16 @@
package org.apache.skywalking.oap.server.storage.plugin.elasticsearch.cache;
+import java.util.*;
import org.apache.skywalking.oap.server.core.Const;
import org.apache.skywalking.oap.server.core.register.*;
import org.apache.skywalking.oap.server.core.storage.cache.IServiceInventoryCacheDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.oap.server.library.util.BooleanUtils;
import org.apache.skywalking.oap.server.storage.plugin.elasticsearch.base.EsDAO;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.search.SearchResponse;
-import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.query.*;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.slf4j.*;
@@ -85,4 +87,29 @@ private int get(String id) {
return null;
}
}
+
+ @Override public List<ServiceInventory> loadLastMappingUpdate() {
+ List<ServiceInventory> serviceInventories = new ArrayList<>();
+
+ try {
+ SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
+
+ BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
+ boolQuery.must().add(QueryBuilders.termQuery(ServiceInventory.IS_ADDRESS, BooleanUtils.TRUE));
+ boolQuery.must().add(QueryBuilders.rangeQuery(ServiceInventory.MAPPING_LAST_UPDATE_TIME).gte(System.currentTimeMillis() - 10000));
+
+ searchSourceBuilder.query(boolQuery);
+ searchSourceBuilder.size(50);
+
+ SearchResponse response = getClient().search(ServiceInventory.MODEL_NAME, searchSourceBuilder);
+
+ for (SearchHit searchHit : response.getHits().getHits()) {
+ serviceInventories.add(this.builder.map2Data(searchHit.getSourceAsMap()));
+ }
+ } catch (Throwable e) {
+ logger.error(e.getMessage());
+ }
+
+ return serviceInventories;
+ }
}
diff --git a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
index d0058e2dc..14dac1620 100644
--- a/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
+++ b/oap-server/server-storage-plugin/storage-elasticsearch-plugin/src/main/java/org/apache/skywalking/oap/server/storage/plugin/elasticsearch/query/MetadataQueryEsDAO.java
@@ -78,8 +78,6 @@ public MetadataQueryEsDAO(ElasticSearchClient client) {
BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
boolQueryBuilder.must().add(timeRangeQueryBuild(startTimestamp, endTimestamp));
- boolQueryBuilder.must().add(QueryBuilders.termQuery(NetworkAddressInventory.SRC_LAYER, srcLayer));
-
sourceBuilder.query(boolQueryBuilder);
sourceBuilder.size(0);
@@ -191,8 +189,15 @@ public Service searchService(String serviceCode) throws IOException {
serviceInstance.setName((String)sourceAsMap.get(ServiceInstanceInventory.NAME));
int languageId = ((Number)sourceAsMap.get(ServiceInstanceInventory.LANGUAGE)).intValue();
serviceInstance.setLanguage(LanguageTrans.INSTANCE.value(languageId));
- serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.OS_NAME, (String)sourceAsMap.get(ServiceInstanceInventory.OS_NAME)));
- serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.HOST_NAME, (String)sourceAsMap.get(ServiceInstanceInventory.HOST_NAME)));
+
+ String osName = (String)sourceAsMap.get(ServiceInstanceInventory.OS_NAME);
+ if (StringUtils.isNotEmpty(osName)) {
+ serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.OS_NAME, osName));
+ }
+ String hostName = (String)sourceAsMap.get(ServiceInstanceInventory.HOST_NAME);
+ if (StringUtils.isNotEmpty(hostName)) {
+ serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.HOST_NAME, hostName));
+ }
serviceInstance.getAttributes().add(new Attribute(ServiceInstanceInventory.PROCESS_NO, String.valueOf(((Number)sourceAsMap.get(ServiceInstanceInventory.PROCESS_NO)).intValue())));
List<String> ipv4s = ServiceInstanceInventory.AgentOsInfo.ipv4sDeserialize((String)sourceAsMap.get(ServiceInstanceInventory.IPV4S));
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 05689b665..ac3912ead 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
@@ -20,12 +20,12 @@
import java.io.IOException;
import java.util.*;
-import org.apache.skywalking.oap.server.core.UnexpectedException;
+import org.apache.skywalking.oap.server.core.*;
+import org.apache.skywalking.oap.server.core.analysis.indicator.Indicator;
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.*;
-import org.apache.skywalking.oap.server.core.source.*;
+import org.apache.skywalking.oap.server.core.source.DetectPoint;
import org.apache.skywalking.oap.server.core.storage.DownsampleingModelNameBuilder;
import org.apache.skywalking.oap.server.core.storage.query.ITopologyQueryDAO;
import org.apache.skywalking.oap.server.library.client.elasticsearch.ElasticSearchClient;
@@ -34,7 +34,7 @@
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.aggregations.AggregationBuilders;
-import org.elasticsearch.search.aggregations.bucket.terms.*;
+import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.builder.SearchSourceBuilder;
/**
@@ -58,7 +58,7 @@ public TopologyQueryEsDAO(ElasticSearchClient client) {
setQueryCondition(sourceBuilder, startTB, endTB, serviceIds);
String indexName = DownsampleingModelNameBuilder.build(step, ServiceRelationServerSideIndicator.INDEX_NAME);
- return load(sourceBuilder, indexName, ServiceRelationServerSideIndicator.SOURCE_SERVICE_ID, ServiceRelationServerSideIndicator.DEST_SERVICE_ID, Source.Service);
+ return load(sourceBuilder, indexName, DetectPoint.SERVER);
}
@Override
@@ -73,7 +73,7 @@ public TopologyQueryEsDAO(ElasticSearchClient client) {
setQueryCondition(sourceBuilder, startTB, endTB, serviceIds);
String indexName = DownsampleingModelNameBuilder.build(step, ServiceRelationClientSideIndicator.INDEX_NAME);
- return load(sourceBuilder, indexName, ServiceRelationClientSideIndicator.SOURCE_SERVICE_ID, ServiceRelationClientSideIndicator.DEST_SERVICE_ID, Source.Service);
+ return load(sourceBuilder, indexName, DetectPoint.CLIENT);
}
private void setQueryCondition(SearchSourceBuilder sourceBuilder, long startTB, long endTB,
@@ -100,7 +100,7 @@ private void setQueryCondition(SearchSourceBuilder sourceBuilder, long startTB,
sourceBuilder.query(QueryBuilders.rangeQuery(ServiceRelationServerSideIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
sourceBuilder.size(0);
- return load(sourceBuilder, indexName, ServiceRelationServerSideIndicator.SOURCE_SERVICE_ID, ServiceRelationServerSideIndicator.DEST_SERVICE_ID, Source.Service);
+ return load(sourceBuilder, indexName, DetectPoint.SERVER);
}
@Override public List<Call> loadClientSideServiceRelations(Step step, long startTB, long endTB) throws IOException {
@@ -109,60 +109,7 @@ private void setQueryCondition(SearchSourceBuilder sourceBuilder, long startTB,
sourceBuilder.query(QueryBuilders.rangeQuery(ServiceRelationServerSideIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
sourceBuilder.size(0);
- return load(sourceBuilder, indexName, ServiceRelationClientSideIndicator.SOURCE_SERVICE_ID, ServiceRelationClientSideIndicator.DEST_SERVICE_ID, Source.Service);
- }
-
- @Override public List<ServiceMapping> loadServiceMappings(Step step, long startTB, long endTB) throws IOException {
- String indexName = DownsampleingModelNameBuilder.build(step, ServiceMappingIndicator.INDEX_NAME);
- SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
- sourceBuilder.query(QueryBuilders.rangeQuery(ServiceMappingIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
- sourceBuilder.size(0);
-
- TermsAggregationBuilder sourceAggregation = AggregationBuilders.terms(ServiceMappingIndicator.SERVICE_ID).field(ServiceMappingIndicator.SERVICE_ID).size(1000);
- sourceAggregation.subAggregation(AggregationBuilders.terms(ServiceMappingIndicator.MAPPING_SERVICE_ID).field(ServiceMappingIndicator.MAPPING_SERVICE_ID).size(1000));
- sourceBuilder.aggregation(sourceAggregation);
-
- SearchResponse response = getClient().search(indexName, sourceBuilder);
-
- List<ServiceMapping> serviceMappings = new ArrayList<>();
- Terms serviceIdTerms = response.getAggregations().get(ServiceMappingIndicator.SERVICE_ID);
- for (Terms.Bucket serviceIdBucket : serviceIdTerms.getBuckets()) {
- Terms mappingServiceIdTerms = serviceIdBucket.getAggregations().get(ServiceMappingIndicator.MAPPING_SERVICE_ID);
- for (Terms.Bucket mappingServiceIdBucket : mappingServiceIdTerms.getBuckets()) {
- ServiceMapping serviceMapping = new ServiceMapping();
- serviceMapping.setServiceId(serviceIdBucket.getKeyAsNumber().intValue());
- serviceMapping.setMappingServiceId(mappingServiceIdBucket.getKeyAsNumber().intValue());
- serviceMappings.add(serviceMapping);
- }
- }
- return serviceMappings;
- }
-
- @Override
- public List<ServiceComponent> loadServiceComponents(Step step, long startTB, long endTB) throws IOException {
- String indexName = DownsampleingModelNameBuilder.build(step, ServiceComponentIndicator.INDEX_NAME);
- SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
- sourceBuilder.query(QueryBuilders.rangeQuery(ServiceComponentIndicator.TIME_BUCKET).gte(startTB).lte(endTB));
- sourceBuilder.size(0);
-
- TermsAggregationBuilder sourceAggregation = AggregationBuilders.terms(ServiceComponentIndicator.SERVICE_ID).field(ServiceComponentIndicator.SERVICE_ID).size(1000);
- sourceAggregation.subAggregation(AggregationBuilders.terms(ServiceComponentIndicator.COMPONENT_ID).field(ServiceComponentIndicator.COMPONENT_ID).size(1000));
- sourceBuilder.aggregation(sourceAggregation);
-
- SearchResponse response = getClient().search(indexName, sourceBuilder);
-
- List<ServiceComponent> serviceComponents = new ArrayList<>();
- Terms serviceIdTerms = response.getAggregations().get(ServiceComponentIndicator.SERVICE_ID);
- for (Terms.Bucket serviceIdBucket : serviceIdTerms.getBuckets()) {
- Terms componentIdTerms = serviceIdBucket.getAggregations().get(ServiceComponentIndicator.COMPONENT_ID);
- for (Terms.Bucket componentIdBucket : componentIdTerms.getBuckets()) {
- ServiceComponent serviceComponent = new ServiceComponent();
- serviceComponent.setServiceId(serviceIdBucket.getKeyAsNumber().intValue());
- serviceComponent.setComponentId(componentIdBucket.getKeyAsNumber().intValue());
- serviceComponents.add(serviceComponent);
- }
- }
- return serviceComponents;
+ return load(sourceBuilder, indexName, DetectPoint.CLIENT);
}
@Override
@@ -183,40 +130,30 @@ private void setQueryCondition(SearchSourceBuilder sourceBuilder, long startTB,
sourceBuilder.query(boolQuery);
- return load(sourceBuilder, indexName, EndpointRelationServerSideIndicator.SOURCE_ENDPOINT_ID, EndpointRelationServerSideIndicator.DEST_ENDPOINT_ID, Source.Endpoint);
+ return load(sourceBuilder, indexName, DetectPoint.SERVER);
}
- private List<Call> load(SearchSourceBuilder sourceBuilder, String indexName, String sourceCName,
- String destCName, Source source) throws IOException {
- TermsAggregationBuilder sourceAggregation = AggregationBuilders.terms(sourceCName).field(sourceCName).size(1000);
- sourceAggregation.subAggregation(AggregationBuilders.terms(destCName).field(destCName).size(1000));
- sourceBuilder.aggregation(sourceAggregation);
+ private List<Call> load(SearchSourceBuilder sourceBuilder, String indexName,
+ DetectPoint detectPoint) throws IOException {
+ sourceBuilder.aggregation(AggregationBuilders.terms(Indicator.ENTITY_ID).field(Indicator.ENTITY_ID).size(1000));
SearchResponse response = getClient().search(indexName, sourceBuilder);
List<Call> calls = new ArrayList<>();
- Terms sourceTerms = response.getAggregations().get(sourceCName);
- for (Terms.Bucket sourceBucket : sourceTerms.getBuckets()) {
- Terms destTerms = sourceBucket.getAggregations().get(destCName);
- for (Terms.Bucket destBucket : destTerms.getBuckets()) {
- Call value = new Call();
- value.setSource(sourceBucket.getKeyAsNumber().intValue());
- value.setTarget(destBucket.getKeyAsNumber().intValue());
- switch (source) {
- case Service:
- value.setId(ServiceRelation.buildEntityId(value.getSource(), value.getTarget()));
- break;
- case Endpoint:
- value.setId(EndpointRelation.buildEntityId(value.getSource(), value.getTarget()));
- break;
- }
- calls.add(value);
+ Terms entityTerms = response.getAggregations().get(Indicator.ENTITY_ID);
+ for (Terms.Bucket entityBucket : entityTerms.getBuckets()) {
+ String entityId = entityBucket.getKeyAsString();
+ String[] ids = entityId.split(Const.ID_SPLIT);
+ if (ids.length == 3) {
+ Call call = new Call();
+ call.setId(entityBucket.getKeyAsString());
+ call.setSource(Integer.valueOf(ids[0]));
+ call.setTarget(Integer.valueOf(ids[1]));
+ call.setComponentId(Integer.valueOf(ids[2]));
+ call.setDetectPoint(detectPoint);
+ calls.add(call);
}
}
return calls;
}
-
- enum Source {
- Service, Endpoint
- }
}
----------------------------------------------------------------
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