From 9d6bf6787b8b3751a08a087e8f6ac6532d80a9bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Wed, 26 Sep 2018 08:20:55 +0800 Subject: [PATCH] Finish mesh source dispatch (#1687) * Merge module and provider of mesh * Finish the skeleton of mesh receiver. * Finish mesh dispatcher. * Fix missing timebucket --- .../skywalking/oap/server/core/Const.java | 6 +- .../server/core/query/TopologyBuilder.java | 6 +- .../pom.xml | 8 + .../provider/IstioTelemetryGRPCHandler.java | 5 +- .../IstioTelemetryReceiverProvider.java | 3 +- .../mesh-receiver-module/pom.xml | 33 --- .../mesh-receiver-provider/pom.xml | 45 ---- .../skywalking-mesh-receiver-plugin/pom.xml | 6 +- .../receiver/mesh/CoreRegisterLinker.java | 61 +++++ .../mesh/MeshDataBufferFileCache.java | 97 ++++++++ .../receiver/mesh}/MeshGRPCHandler.java | 3 +- .../receiver/mesh/MeshModuleConfig.java | 33 +++ .../receiver/mesh}/MeshReceiverModule.java | 2 +- .../receiver/mesh}/MeshReceiverProvider.java | 22 +- .../mesh/ServiceMeshMetricDataDecorator.java | 132 +++++++++++ .../mesh/TelemetryDataDispatcher.java | 215 ++++++++++++++++++ ...ing.oap.server.library.module.ModuleDefine | 3 +- ...g.oap.server.library.module.ModuleProvider | 3 +- .../endpoint/MultiScopesSpanListener.java | 10 +- 19 files changed, 588 insertions(+), 105 deletions(-) delete mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/pom.xml delete mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/pom.xml create mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/CoreRegisterLinker.java create mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java rename oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/{mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider => src/main/java/org/apache/skywalking/aop/server/receiver/mesh}/MeshGRPCHandler.java (94%) create mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshModuleConfig.java rename oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/{mesh-receiver-module/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/module => src/main/java/org/apache/skywalking/aop/server/receiver/mesh}/MeshReceiverModule.java (94%) rename oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/{mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider => src/main/java/org/apache/skywalking/aop/server/receiver/mesh}/MeshReceiverProvider.java (77%) create mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/ServiceMeshMetricDataDecorator.java create mode 100644 oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java rename oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/{mesh-receiver-module => }/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine (91%) rename oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/{mesh-receiver-provider => }/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider (91%) diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/Const.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/Const.java index fbad92199..4324835fa 100644 --- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/Const.java +++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/Const.java @@ -26,9 +26,9 @@ public class Const { public static final String ID_SPLIT = "_"; public static final String KEY_VALUE_SPLIT = ","; public static final String ARRAY_SPLIT = "|"; - public static final int NONE_SERVICE_ID = 1; - public static final int NONE_INSTANCE_ID = 1; - public static final int NONE_ENDPOINT_ID = 1; + public static final int USER_SERVICE_ID = 1; + public static final int USER_INSTANCE_ID = 1; + public static final int USER_ENDPOINT_ID = 1; public static final String NONE_ENDPOINT_NAME = "None"; public static final String USER_CODE = "User"; public static final String SEGMENT_SPAN_SPLIT = "S"; 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 7be38bd04..a933a5082 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 @@ -95,7 +95,7 @@ class TopologyBuilder { ServiceInventory source = serviceInventoryCache.get(serverCall.getSource()); ServiceInventory target = serviceInventoryCache.get(serverCall.getTarget()); - if (source.getSequence() == Const.NONE_SERVICE_ID) { + if (source.getSequence() == Const.USER_SERVICE_ID) { if (!nodeIds.contains(source.getSequence())) { Node visualUserNode = new Node(); visualUserNode.setId(source.getSequence()); @@ -122,7 +122,7 @@ class TopologyBuilder { call.setTarget(target.getSequence()); call.setId(serverCall.getId()); - if (source.getSequence() == Const.NONE_SERVICE_ID) { + if (source.getSequence() == Const.USER_SERVICE_ID) { call.setCallType(Const.EMPTY_STRING); } else { call.setCallType(nodeCompMap.get(serverCall.getTarget())); @@ -147,7 +147,7 @@ class TopologyBuilder { serviceRelationServerCalls.forEach(serverCall -> { ServiceInventory source = serviceInventoryCache.get(serverCall.getSource()); - if (BooleanUtils.valueToBoolean(source.getIsAddress()) || source.getSequence() == Const.NONE_SERVICE_ID) { + if (BooleanUtils.valueToBoolean(source.getIsAddress()) || source.getSequence() == Const.USER_SERVICE_ID) { filteredCalls.add(serverCall); } }); diff --git a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/pom.xml b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/pom.xml index 1baefe53f..786b6d6b8 100644 --- a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/pom.xml +++ b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/pom.xml @@ -30,6 +30,14 @@ skywalking-istio-telemetry-receiver-plugin jar + + + org.apache.skywalking + skywalking-mesh-receiver-plugin + ${project.version} + + + diff --git a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java index c21a5ddf2..bd65ade96 100644 --- a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryGRPCHandler.java @@ -27,6 +27,7 @@ import io.istio.api.policy.v1beta1.TypeProto; import java.time.Duration; import java.time.Instant; import java.util.Map; +import org.apache.skywalking.aop.server.receiver.mesh.TelemetryDataDispatcher; import org.apache.skywalking.apm.network.common.DetectPoint; import org.apache.skywalking.apm.network.servicemesh.Protocol; import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; @@ -58,7 +59,7 @@ public class IstioTelemetryGRPCHandler extends HandleMetricServiceGrpc.HandleMet String endpoint; boolean status = true; Protocol netProtocol; - if (protocol.equals("http") || protocol.equals("https")) { + if (protocol.equals("http") || protocol.equals("https") || requestScheme.equals("http") || requestScheme.equals("https")) { endpoint = requestScheme + "/" + requestMethod + "/" + requestPath; status = responseCode >= 200 && responseCode < 400; netProtocol = Protocol.HTTP; @@ -83,6 +84,8 @@ public class IstioTelemetryGRPCHandler extends HandleMetricServiceGrpc.HandleMet .setDestServiceInstance(string(i, "destinationUID")).setEndpoint(endpoint).setLatency(latency) .setResponseCode(Math.toIntExact(responseCode)).setStatus(status).setProtocol(netProtocol).setDetectPoint(detectPoint).build(); logger.debug("Transformed metric {}", metric); + + TelemetryDataDispatcher.preProcess(metric); } responseObserver.onNext(ReportProto.ReportResult.newBuilder().build()); responseObserver.onCompleted(); diff --git a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java index 4421f2498..5ac2f15f5 100644 --- a/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/istio/telemetry/provider/IstioTelemetryReceiverProvider.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.receiver.istio.telemetry.provider; +import org.apache.skywalking.aop.server.receiver.mesh.MeshReceiverModule; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; import org.apache.skywalking.oap.server.library.module.ModuleConfig; @@ -53,6 +54,6 @@ public class IstioTelemetryReceiverProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[] {CoreModule.NAME}; + return new String[] {CoreModule.NAME, MeshReceiverModule.NAME}; } } \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/pom.xml b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/pom.xml deleted file mode 100644 index 874e25c2e..000000000 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/pom.xml +++ /dev/null @@ -1,33 +0,0 @@ - - - - - - skywalking-mesh-receiver-plugin - org.apache.skywalking - 6.0.0-alpha-SNAPSHOT - - 4.0.0 - - mesh-receiver-module - - - \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/pom.xml b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/pom.xml deleted file mode 100644 index 3119c3bc2..000000000 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/pom.xml +++ /dev/null @@ -1,45 +0,0 @@ - - - - - - skywalking-mesh-receiver-plugin - org.apache.skywalking - 6.0.0-alpha-SNAPSHOT - - 4.0.0 - - mesh-receiver-provider - - - org.apache.skywalking - mesh-receiver-module - ${project.version} - - - org.apache.skywalking - server-core - ${project.version} - - - - - \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/pom.xml b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/pom.xml index c8ec9e3d6..43a835120 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/pom.xml +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/pom.xml @@ -28,10 +28,6 @@ 4.0.0 skywalking-mesh-receiver-plugin - pom - - mesh-receiver-module - mesh-receiver-provider - + jar \ No newline at end of file diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/CoreRegisterLinker.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/CoreRegisterLinker.java new file mode 100644 index 000000000..2868b82d9 --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/CoreRegisterLinker.java @@ -0,0 +1,61 @@ +/* + * 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.aop.server.receiver.mesh; + +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.register.service.IEndpointInventoryRegister; +import org.apache.skywalking.oap.server.core.register.service.IServiceInstanceInventoryRegister; +import org.apache.skywalking.oap.server.core.register.service.IServiceInventoryRegister; +import org.apache.skywalking.oap.server.library.module.ModuleManager; + +/** + * @author wusheng + */ +public class CoreRegisterLinker { + private static volatile ModuleManager MODULE_MANAGER; + private static volatile IServiceInventoryRegister SERVICE_INVENTORY_REGISTER; + private static volatile IServiceInstanceInventoryRegister SERVICE_INSTANCE_INVENTORY_REGISTER; + private static volatile IEndpointInventoryRegister ENDPOINT_INVENTORY_REGISTER; + + public static void setModuleManager(ModuleManager moduleManager) { + CoreRegisterLinker.MODULE_MANAGER = moduleManager; + } + + public static IServiceInventoryRegister getServiceInventoryRegister() { + if (SERVICE_INVENTORY_REGISTER == null) { + SERVICE_INVENTORY_REGISTER = MODULE_MANAGER.find(CoreModule.NAME).getService(IServiceInventoryRegister.class); + } + return SERVICE_INVENTORY_REGISTER; + } + + public static IServiceInstanceInventoryRegister getServiceInstanceInventoryRegister() { + if (SERVICE_INSTANCE_INVENTORY_REGISTER == null) { + SERVICE_INSTANCE_INVENTORY_REGISTER = MODULE_MANAGER.find(CoreModule.NAME).getService(IServiceInstanceInventoryRegister.class); + } + return SERVICE_INSTANCE_INVENTORY_REGISTER; + } + + public static IEndpointInventoryRegister getEndpointInventoryRegister() { + if (ENDPOINT_INVENTORY_REGISTER == null) { + ENDPOINT_INVENTORY_REGISTER = MODULE_MANAGER.find(CoreModule.NAME).getService(IEndpointInventoryRegister.class); + } + return ENDPOINT_INVENTORY_REGISTER; + } + +} diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java new file mode 100644 index 000000000..2ea9237ca --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java @@ -0,0 +1,97 @@ +/* + * 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.aop.server.receiver.mesh; + +import java.io.IOException; +import java.util.List; +import org.apache.skywalking.apm.commons.datacarrier.DataCarrier; +import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer; +import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; +import org.apache.skywalking.oap.server.library.buffer.BufferStream; +import org.apache.skywalking.oap.server.library.buffer.DataStreamReader; + +public class MeshDataBufferFileCache implements IConsumer, DataStreamReader.CallBack { + private MeshModuleConfig config; + private DataCarrier dataCarrier; + private BufferStream stream; + + public MeshDataBufferFileCache(MeshModuleConfig config) { + this.config = config; + dataCarrier = new DataCarrier<>(3, 1024); + } + + void start() throws IOException { + dataCarrier.consume(this, 1); + BufferStream.Builder builder = new BufferStream.Builder<>(config.getBufferPath()); + builder.cleanWhenRestart(config.isBufferFileCleanWhenRestart()); + builder.dataFileMaxSize(config.getBufferDataMaxFileSize()); + builder.offsetFileMaxSize(config.getBufferOffsetMaxFileSize()); + builder.parser(ServiceMeshMetric.parser()); + builder.callBack(this); + + stream = builder.build(); + stream.initialize(); + } + + @Override public void init() { + + } + + public void in(ServiceMeshMetric metric) { + dataCarrier.produce(new ServiceMeshMetricDataDecorator(metric)); + } + + /** + * Queue callback, make sure concurrency doesn't happen + * + * @param data + */ + @Override public void consume(List data) { + for (ServiceMeshMetricDataDecorator decorator : data) { + if (decorator.tryMetaDataRegister()) { + TelemetryDataDispatcher.doDispatch(decorator); + } else { + stream.write(decorator.getMetric()); + } + } + } + + @Override public void onError(List data, Throwable t) { + + } + + @Override public void onExit() { + + } + + /** + * File buffer callback. Block reading from buffer file, until metadata register done. + * + * @param message + * @return + */ + @Override public boolean call(ServiceMeshMetric message) { + ServiceMeshMetricDataDecorator decorator = new ServiceMeshMetricDataDecorator(message); + if (decorator.tryMetaDataRegister()) { + TelemetryDataDispatcher.doDispatch(decorator); + return true; + } + return false; + } +} diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshGRPCHandler.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java similarity index 94% rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshGRPCHandler.java rename to oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java index 889382106..d8852dae9 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshGRPCHandler.java +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshGRPCHandler.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.aop.server.receiver.mesh.provider; +package org.apache.skywalking.aop.server.receiver.mesh; import io.grpc.stub.StreamObserver; import org.apache.skywalking.apm.network.servicemesh.MeshProbeDownstream; @@ -35,6 +35,7 @@ public class MeshGRPCHandler extends ServiceMeshMetricServiceGrpc.ServiceMeshMet if (logger.isDebugEnabled()) { logger.debug("Received mesh metric: {}", metric); } + TelemetryDataDispatcher.preProcess(metric); } @Override public void onError(Throwable throwable) { diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshModuleConfig.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshModuleConfig.java new file mode 100644 index 000000000..5f650ce6b --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshModuleConfig.java @@ -0,0 +1,33 @@ +/* + * 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.aop.server.receiver.mesh; + +import lombok.Getter; +import lombok.Setter; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; + +/** + * @author peng-yongsheng + */ +public class MeshModuleConfig extends ModuleConfig { + @Setter @Getter private String bufferPath; + @Setter @Getter private int bufferOffsetMaxFileSize; + @Setter @Getter private int bufferDataMaxFileSize; + @Setter @Getter private boolean bufferFileCleanWhenRestart; +} diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/module/MeshReceiverModule.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverModule.java similarity index 94% rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/module/MeshReceiverModule.java rename to oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverModule.java index 23e50cd55..5809afbdb 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/module/MeshReceiverModule.java +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverModule.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.aop.server.receiver.mesh.module; +package org.apache.skywalking.aop.server.receiver.mesh; import org.apache.skywalking.oap.server.library.module.ModuleDefine; diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshReceiverProvider.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java similarity index 77% rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshReceiverProvider.java rename to oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java index cedb1b433..05c6d4904 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/provider/MeshReceiverProvider.java +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshReceiverProvider.java @@ -16,9 +16,9 @@ * */ -package org.apache.skywalking.aop.server.receiver.mesh.provider; +package org.apache.skywalking.aop.server.receiver.mesh; -import org.apache.skywalking.aop.server.receiver.mesh.module.MeshReceiverModule; +import java.io.IOException; import org.apache.skywalking.oap.server.core.CoreModule; import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister; import org.apache.skywalking.oap.server.library.module.ModuleConfig; @@ -28,6 +28,12 @@ import org.apache.skywalking.oap.server.library.module.ModuleStartException; import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; public class MeshReceiverProvider extends ModuleProvider { + private MeshModuleConfig config; + + public MeshReceiverProvider() { + config = new MeshModuleConfig(); + } + @Override public String name() { return "default"; } @@ -37,13 +43,21 @@ public class MeshReceiverProvider extends ModuleProvider { } @Override public ModuleConfig createConfigBeanIfAbsent() { - return null; + return config; } @Override public void prepare() throws ServiceNotProvidedException, ModuleStartException { + MeshDataBufferFileCache cache = new MeshDataBufferFileCache(config); + try { + cache.start(); + TelemetryDataDispatcher.setCache(cache, getManager()); + } catch (IOException e) { + throw new ModuleStartException(e.getMessage(), e); + } } @Override public void start() throws ServiceNotProvidedException, ModuleStartException { + CoreRegisterLinker.setModuleManager(getManager()); GRPCHandlerRegister service = getManager().find(CoreModule.NAME).getService(GRPCHandlerRegister.class); service.addHandler(new MeshGRPCHandler()); } @@ -53,6 +67,6 @@ public class MeshReceiverProvider extends ModuleProvider { } @Override public String[] requiredModules() { - return new String[]{CoreModule.NAME}; + return new String[] {CoreModule.NAME}; } } diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/ServiceMeshMetricDataDecorator.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/ServiceMeshMetricDataDecorator.java new file mode 100644 index 000000000..7da1ac19b --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/ServiceMeshMetricDataDecorator.java @@ -0,0 +1,132 @@ +/* + * 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.aop.server.receiver.mesh; + +import org.apache.skywalking.apm.network.common.DetectPoint; +import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; +import org.apache.skywalking.oap.server.core.Const; +import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory; + +/** + * @author wusheng + */ +public class ServiceMeshMetricDataDecorator { + private ServiceMeshMetric origin; + private ServiceMeshMetric rebuiltData; + private ServiceMeshMetric.Builder newDataBuilder; + private int endpointId; + + public ServiceMeshMetricDataDecorator(ServiceMeshMetric origin) { + this.origin = origin; + } + + boolean tryMetaDataRegister() { + int sourceServiceId = 0; + int sourceServiceInstanceId = 0; + int destServiceId = 0; + int destServiceInstanceId = 0; + boolean isRegistered = true; + sourceServiceId = origin.getSourceServiceId(); + if (sourceServiceId == Const.NONE) { + sourceServiceId = CoreRegisterLinker.getServiceInventoryRegister().getOrCreate(origin.getSourceServiceName()); + if (sourceServiceId != Const.NONE) { + getNewDataBuilder().setSourceServiceId(sourceServiceId); + } else { + isRegistered = false; + } + } + sourceServiceInstanceId = origin.getSourceServiceInstanceId(); + if (sourceServiceId != Const.NONE && sourceServiceInstanceId == Const.NONE) { + sourceServiceInstanceId = CoreRegisterLinker.getServiceInstanceInventoryRegister().getOrCreate(sourceServiceId, origin.getSourceServiceInstance(), origin.getEndTime(), + getOSInfoForMesh(origin.getSourceServiceInstance())); + if (sourceServiceInstanceId != Const.NONE) { + getNewDataBuilder().setSourceServiceInstanceId(sourceServiceInstanceId); + } else { + isRegistered = false; + } + } + destServiceId = origin.getDestServiceId(); + if (destServiceId == Const.NONE) { + destServiceId = CoreRegisterLinker.getServiceInventoryRegister().getOrCreate(origin.getDestServiceName()); + if (destServiceId != Const.NONE) { + getNewDataBuilder().setDestServiceId(destServiceId); + } else { + isRegistered = false; + } + } + destServiceInstanceId = origin.getDestServiceInstanceId(); + if (destServiceId != Const.NONE && destServiceInstanceId == Const.NONE) { + destServiceInstanceId = CoreRegisterLinker.getServiceInstanceInventoryRegister().getOrCreate(destServiceId, origin.getDestServiceInstance(), origin.getEndTime(), + getOSInfoForMesh(origin.getSourceServiceInstance())); + if (destServiceInstanceId != Const.NONE) { + getNewDataBuilder().setDestServiceInstanceId(destServiceInstanceId); + } else { + isRegistered = false; + } + } + String endpoint = origin.getEndpoint(); + if (destServiceId != Const.NONE) { + DetectPoint point = origin.getDetectPoint(); + /** + * Detect point matches {@link SpanType}, value options: + * + * Entry = 0; Server side. + * Exit = 1; Client side + * Local = 2; Never used. + * + */ + int detectPointInt = DetectPoint.client.equals(point) ? 1 : 0; + endpointId = CoreRegisterLinker.getEndpointInventoryRegister().getOrCreate(destServiceId, endpoint, detectPointInt); + if (endpointId != Const.NONE) { + } else { + isRegistered = false; + } + } + + return isRegistered; + } + + public ServiceMeshMetric getMetric() { + if (newDataBuilder != null) { + if (rebuiltData == null) { + rebuiltData = newDataBuilder.build(); + } + return rebuiltData; + } else { + return origin; + } + } + + public int getEndpointId() { + return endpointId; + } + + private ServiceMeshMetric.Builder getNewDataBuilder() { + if (newDataBuilder == null) { + newDataBuilder = origin.toBuilder(); + } + return newDataBuilder; + } + + private ServiceInstanceInventory.AgentOsInfo getOSInfoForMesh(String instanceName) { + ServiceInstanceInventory.AgentOsInfo osInfo = new ServiceInstanceInventory.AgentOsInfo(); + osInfo.setHostname(instanceName); + return osInfo; + } +} diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java new file mode 100644 index 000000000..4dfcc4e19 --- /dev/null +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java @@ -0,0 +1,215 @@ +/* + * 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.aop.server.receiver.mesh; + +import org.apache.logging.log4j.util.Strings; +import org.apache.skywalking.apm.network.servicemesh.Protocol; +import org.apache.skywalking.apm.network.servicemesh.ServiceMeshMetric; +import org.apache.skywalking.oap.server.core.CoreModule; +import org.apache.skywalking.oap.server.core.cache.ServiceInstanceInventoryCache; +import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache; +import org.apache.skywalking.oap.server.core.source.DetectPoint; +import org.apache.skywalking.oap.server.core.source.Endpoint; +import org.apache.skywalking.oap.server.core.source.RequestType; +import org.apache.skywalking.oap.server.core.source.Service; +import org.apache.skywalking.oap.server.core.source.ServiceInstance; +import org.apache.skywalking.oap.server.core.source.ServiceInstanceRelation; +import org.apache.skywalking.oap.server.core.source.ServiceRelation; +import org.apache.skywalking.oap.server.core.source.SourceReceiver; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.util.TimeBucketUtils; + +/** + * TelemetryDataDispatcher processes the {@link ServiceMeshMetric} format telemetry data, transfers it to source + * dispatcher. + * + * @author wusheng + */ +public class TelemetryDataDispatcher { + private static MeshDataBufferFileCache CACHE; + private static ServiceInventoryCache SERVICE_CACHE; + private static ServiceInstanceInventoryCache SERVICE_INSTANCE_CACHE; + private static SourceReceiver SOURCE_RECEIVER; + + private TelemetryDataDispatcher() { + + } + + public static void setCache(MeshDataBufferFileCache cache, ModuleManager moduleManager) { + CACHE = cache; + SERVICE_CACHE = moduleManager.find(CoreModule.NAME).getService(ServiceInventoryCache.class); + SERVICE_INSTANCE_CACHE = moduleManager.find(CoreModule.NAME).getService(ServiceInstanceInventoryCache.class); + SOURCE_RECEIVER = moduleManager.find(CoreModule.NAME).getService(SourceReceiver.class); + } + + public static void preProcess(ServiceMeshMetric data) { + CACHE.in(data); + } + + /** + * The {@link ServiceMeshMetricDataDecorator} is standard, all metadata registered through {@link #CACHE} + * + * @param decorator + */ + static void doDispatch(ServiceMeshMetricDataDecorator decorator) { + ServiceMeshMetric metric = decorator.getMetric(); + long minuteTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(metric.getStartTime()); + toService(decorator, minuteTimeBucket); + toServiceRelation(decorator, minuteTimeBucket); + toServiceInstance(decorator, minuteTimeBucket); + toServiceInstanceRelation(decorator, minuteTimeBucket); + toEndpoint(decorator, minuteTimeBucket); + } + + private static void toService(ServiceMeshMetricDataDecorator decorator, long minuteTimeBucket) { + ServiceMeshMetric metric = decorator.getMetric(); + Service service = new Service(); + service.setTimeBucket(minuteTimeBucket); + service.setId(metric.getDestServiceId()); + service.setName(getServiceName(metric.getDestServiceId(), metric.getDestServiceName())); + service.setServiceInstanceName(getServiceInstanceName(metric.getDestServiceInstanceId(), metric.getDestServiceInstance())); + service.setEndpointName(metric.getEndpoint()); + service.setLatency(metric.getLatency()); + service.setStatus(metric.getStatus()); + service.setType(protocol2Type(metric.getProtocol())); + + SOURCE_RECEIVER.receive(service); + } + + private static void toServiceRelation(ServiceMeshMetricDataDecorator decorator, long minuteTimeBucket) { + ServiceMeshMetric metric = decorator.getMetric(); + ServiceRelation serviceRelation = new ServiceRelation(); + serviceRelation.setTimeBucket(minuteTimeBucket); + serviceRelation.setSourceServiceId(metric.getSourceServiceId()); + serviceRelation.setSourceServiceName(getServiceName(metric.getSourceServiceId(), metric.getSourceServiceName())); + serviceRelation.setSourceServiceInstanceName(getServiceInstanceName(metric.getSourceServiceInstanceId(), metric.getSourceServiceInstance())); + + serviceRelation.setDestServiceId(metric.getDestServiceId()); + serviceRelation.setDestServiceName(getServiceName(metric.getDestServiceId(), metric.getDestServiceName())); + serviceRelation.setDestServiceInstanceName(getServiceInstanceName(metric.getDestServiceInstanceId(), metric.getDestServiceInstance())); + + serviceRelation.setEndpoint(metric.getEndpoint()); + serviceRelation.setLatency(metric.getLatency()); + serviceRelation.setStatus(metric.getStatus()); + serviceRelation.setType(protocol2Type(metric.getProtocol())); + serviceRelation.setResponseCode(metric.getResponseCode()); + serviceRelation.setDetectPoint(detectPointMapping(metric.getDetectPoint())); + + SOURCE_RECEIVER.receive(serviceRelation); + } + + private static void toServiceInstance(ServiceMeshMetricDataDecorator decorator, long minuteTimeBucket) { + ServiceMeshMetric metric = decorator.getMetric(); + ServiceInstance serviceInstance = new ServiceInstance(); + serviceInstance.setTimeBucket(minuteTimeBucket); + serviceInstance.setId(metric.getDestServiceId()); + serviceInstance.setName(getServiceInstanceName(metric.getDestServiceInstanceId(), metric.getDestServiceInstance())); + serviceInstance.setServiceId(metric.getDestServiceId()); + serviceInstance.setServiceName(getServiceName(metric.getDestServiceId(), metric.getDestServiceName())); + serviceInstance.setEndpointName(metric.getEndpoint()); + serviceInstance.setLatency(metric.getLatency()); + serviceInstance.setStatus(metric.getStatus()); + serviceInstance.setType(protocol2Type(metric.getProtocol())); + + SOURCE_RECEIVER.receive(serviceInstance); + } + + private static void toServiceInstanceRelation(ServiceMeshMetricDataDecorator decorator, long minuteTimeBucket) { + ServiceMeshMetric metric = decorator.getMetric(); + ServiceInstanceRelation serviceRelation = new ServiceInstanceRelation(); + serviceRelation.setTimeBucket(minuteTimeBucket); + serviceRelation.setSourceServiceInstanceId(metric.getSourceServiceInstanceId()); + serviceRelation.setSourceServiceInstanceName(getServiceInstanceName(metric.getSourceServiceInstanceId(), metric.getSourceServiceInstance())); + serviceRelation.setSourceServiceId(metric.getSourceServiceId()); + serviceRelation.setSourceServiceName(getServiceName(metric.getSourceServiceId(), metric.getSourceServiceName())); + + serviceRelation.setDestServiceInstanceId(metric.getDestServiceInstanceId()); + serviceRelation.setDestServiceInstanceName(getServiceInstanceName(metric.getDestServiceInstanceId(), metric.getDestServiceInstance())); + serviceRelation.setDestServiceId(metric.getDestServiceId()); + serviceRelation.setDestServiceName(getServiceName(metric.getDestServiceId(), metric.getDestServiceName())); + + serviceRelation.setEndpoint(metric.getEndpoint()); + serviceRelation.setLatency(metric.getLatency()); + serviceRelation.setStatus(metric.getStatus()); + serviceRelation.setType(protocol2Type(metric.getProtocol())); + serviceRelation.setResponseCode(metric.getResponseCode()); + serviceRelation.setDetectPoint(detectPointMapping(metric.getDetectPoint())); + + SOURCE_RECEIVER.receive(serviceRelation); + } + + private static void toEndpoint(ServiceMeshMetricDataDecorator decorator, long minuteTimeBucket) { + ServiceMeshMetric metric = decorator.getMetric(); + Endpoint endpoint = new Endpoint(); + endpoint.setTimeBucket(minuteTimeBucket); + endpoint.setId(decorator.getEndpointId()); + endpoint.setName(metric.getEndpoint()); + endpoint.setServiceId(metric.getDestServiceId()); + endpoint.setServiceName(getServiceName(metric.getDestServiceId(), metric.getDestServiceName())); + endpoint.setServiceInstanceId(metric.getDestServiceInstanceId()); + endpoint.setServiceInstanceName(getServiceInstanceName(metric.getDestServiceInstanceId(), metric.getDestServiceInstance())); + + endpoint.setLatency(metric.getLatency()); + endpoint.setStatus(metric.getStatus()); + endpoint.setType(protocol2Type(metric.getProtocol())); + + SOURCE_RECEIVER.receive(endpoint); + } + + private static RequestType protocol2Type(Protocol protocol) { + switch (protocol) { + case gRPC: + return RequestType.gRPC; + case HTTP: + return RequestType.HTTP; + case UNRECOGNIZED: + default: + return RequestType.RPC; + } + } + + private static DetectPoint detectPointMapping(org.apache.skywalking.apm.network.common.DetectPoint detectPoint) { + switch (detectPoint) { + case client: + return DetectPoint.CLIENT; + case server: + return DetectPoint.SERVER; + case proxy: + return DetectPoint.PROXY; + default: + return DetectPoint.SERVER; + } + } + + private static String getServiceName(int serviceId, String serviceName) { + if (Strings.isBlank(serviceName)) { + return SERVICE_CACHE.get(serviceId).getName(); + } else { + return serviceName; + } + } + + private static String getServiceInstanceName(int serviceInstanceId, String serviceInstanceName) { + if (Strings.isBlank(serviceInstanceName)) { + return SERVICE_INSTANCE_CACHE.get(serviceInstanceId).getName(); + } else { + return serviceInstanceName; + } + } +} diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine similarity index 91% rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine rename to oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine index 017617b51..db6c844ec 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine @@ -17,4 +17,5 @@ # -org.apache.skywalking.aop.server.receiver.mesh.module.MeshReceiverModule \ No newline at end of file + +org.apache.skywalking.aop.server.receiver.mesh.MeshReceiverModule diff --git a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider similarity index 91% rename from oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider rename to oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider index d80490517..1da66cfbe 100644 --- a/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider +++ b/oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -16,5 +16,4 @@ # # - -org.apache.skywalking.aop.server.receiver.mesh.provider.MeshReceiverProvider \ No newline at end of file +org.apache.skywalking.aop.server.receiver.mesh.MeshReceiverProvider 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 1854533a2..c64305e4f 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 @@ -97,9 +97,9 @@ public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListe } } else { SourceBuilder sourceBuilder = new SourceBuilder(); - sourceBuilder.setSourceEndpointId(Const.NONE_ENDPOINT_ID); - sourceBuilder.setSourceServiceInstanceId(Const.NONE_INSTANCE_ID); - sourceBuilder.setSourceServiceId(Const.NONE_SERVICE_ID); + sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID); + sourceBuilder.setSourceServiceInstanceId(Const.USER_INSTANCE_ID); + sourceBuilder.setSourceServiceId(Const.USER_SERVICE_ID); sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId()); sourceBuilder.setDestServiceInstanceId(segmentCoreInfo.getApplicationInstanceId()); sourceBuilder.setDestServiceId(segmentCoreInfo.getApplicationId()); @@ -122,7 +122,7 @@ public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListe int destServiceId = serviceInventoryCache.getServiceId(peerId); int destInstanceId = instanceInventoryCache.getServiceInstanceId(destServiceId, peerId); - sourceBuilder.setSourceEndpointId(Const.NONE_ENDPOINT_ID); + sourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID); sourceBuilder.setSourceServiceInstanceId(segmentCoreInfo.getApplicationInstanceId()); sourceBuilder.setSourceServiceId(segmentCoreInfo.getApplicationId()); sourceBuilder.setDestEndpointId(spanDecorator.getOperationNameId()); @@ -175,7 +175,7 @@ public class MultiScopesSpanListener implements EntrySpanListener, ExitSpanListe if (nonNull(entrySpanDecorator)) { exitSourceBuilder.setSourceEndpointId(entrySpanDecorator.getOperationNameId()); } else { - exitSourceBuilder.setSourceEndpointId(Const.NONE_ENDPOINT_ID); + exitSourceBuilder.setSourceEndpointId(Const.USER_ENDPOINT_ID); } exitSourceBuilder.setSourceEndpointName(endpointInventoryCache.get(exitSourceBuilder.getSourceEndpointId()).getName());