Finish mesh source dispatch (#1687)

* Merge module and provider of mesh

* Finish the skeleton of mesh receiver.

* Finish mesh dispatcher.

* Fix missing timebucket

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";

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);
             }
         });

oap-server/server-receiver-plugin/skywalking-istio-telemetry-receiver-plugin/pom.xml

@@ -30,6 +30,14 @@
     <artifactId>skywalking-istio-telemetry-receiver-plugin</artifactId>
     <packaging>jar</packaging>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>skywalking-mesh-receiver-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
     <build>
         <plugins>
             <plugin>

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();

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

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-module/pom.xml

-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>skywalking-mesh-receiver-plugin</artifactId>
-        <groupId>org.apache.skywalking</groupId>
-        <version>6.0.0-alpha-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>mesh-receiver-module</artifactId>
-
-
-</project>
\ No newline at end of file

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/mesh-receiver-provider/pom.xml

-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-  ~ 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.
-  ~
-  -->
-
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>skywalking-mesh-receiver-plugin</artifactId>
-        <groupId>org.apache.skywalking</groupId>
-        <version>6.0.0-alpha-SNAPSHOT</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>mesh-receiver-provider</artifactId>
-    <dependencies>
-        <dependency>
-            <groupId>org.apache.skywalking</groupId>
-            <artifactId>mesh-receiver-module</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.skywalking</groupId>
-            <artifactId>server-core</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-    </dependencies>
-
-
-</project>
\ No newline at end of file

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/pom.xml

@@ -28,10 +28,6 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>skywalking-mesh-receiver-plugin</artifactId>
-    <packaging>pom</packaging>
-    <modules>
-        <module>mesh-receiver-module</module>
-        <module>mesh-receiver-provider</module>
-    </modules>
+    <packaging>jar</packaging>
 
 </project>
\ No newline at end of file

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/CoreRegisterLinker.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.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;
+    }
+
+}

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshDataBufferFileCache.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.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<ServiceMeshMetricDataDecorator>, DataStreamReader.CallBack<ServiceMeshMetric> {
+    private MeshModuleConfig config;
+    private DataCarrier<ServiceMeshMetricDataDecorator> dataCarrier;
+    private BufferStream<ServiceMeshMetric> stream;
+
+    public MeshDataBufferFileCache(MeshModuleConfig config) {
+        this.config = config;
+        dataCarrier = new DataCarrier<>(3, 1024);
+    }
+
+    void start() throws IOException {
+        dataCarrier.consume(this, 1);
+        BufferStream.Builder<ServiceMeshMetric> 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<ServiceMeshMetricDataDecorator> data) {
+        for (ServiceMeshMetricDataDecorator decorator : data) {
+            if (decorator.tryMetaDataRegister()) {
+                TelemetryDataDispatcher.doDispatch(decorator);
+            } else {
+                stream.write(decorator.getMetric());
+            }
+        }
+    }
+
+    @Override public void onError(List<ServiceMeshMetricDataDecorator> 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;
+    }
+}

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 → 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) {

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/MeshModuleConfig.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.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;
+}

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 → 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;
 

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 → 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};
     }
 }

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/ServiceMeshMetricDataDecorator.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.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;
+    }
+}

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.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;
+        }
+    }
+}

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 → 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

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 → 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

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());