Restore Zipkin receiver based on new core (#1932)

* Change some codes to make zipkin receiver works, not done yet.

* @adriancole Zipkin receiver is coming back again.

* Fix document and fix for Zipkin receiver.

* Fix some bugs.

* Fix register bug.

* Fix transfer bug for new backend core.

docs/en/setup/backend/backend-receivers.md

@@ -10,23 +10,32 @@ We have following receivers, and `default` implementors are provided in our Apac
 1. **service-mesh**. gRPC services accept data from inbound mesh probes.
 1. **istio-telemetry**. Istio telemetry is from Istio official bypass adaptor, this receiver match its gRPC services.
 1. **receiver-jvm**. gRPC services accept JVM metric data.
+1. **receiver_zipkin**. HTTP service accepts Span in Zipkin v1 and v2 formats. Notice, this receiver only
+works as expected in backend single node mode. Cluster mode is not supported. Welcome anyone to improve this.
 
 The sample settings of these receivers should be already in default `application.yml`, and also list here
 ```yaml
+receiver-register:
+  default:
 receiver-trace:
   default:
-    bufferPath: ../buffer/  # Path to trace buffer files, suggest to use absolute path
+    bufferPath: ../trace-buffer/  # Path to trace buffer files, suggest to use absolute path
     bufferOffsetMaxFileSize: 100 # Unit is MB
     bufferDataMaxFileSize: 500 # Unit is MB
-    bufferFileCleanWhenRestart: false # Clean buffer file when backend restart.
+    bufferFileCleanWhenRestart: false
 receiver-jvm:
   default:
 service-mesh:
   default:
-    bufferPath: ../mesh-buffer/  # Path to mesh telemetry data buffer files, suggest to use absolute path
+    bufferPath: ../mesh-buffer/  # Path to trace buffer files, suggest to use absolute path
     bufferOffsetMaxFileSize: 100 # Unit is MB
     bufferDataMaxFileSize: 500 # Unit is MB
     bufferFileCleanWhenRestart: false
 istio-telemetry:
   default:
+receiver_zipkin:
+  default:
+    host: 0.0.0.0
+    port: 9411
+    contextPath: /
 ```

oap-server/pom.xml

@@ -61,6 +61,8 @@
         <joda-time.version>2.9.9</joda-time.version>
         <kubernetes.version>2.0.0</kubernetes.version>
         <hikaricp.version>3.1.0</hikaricp.version>
+        <zipkin.version>2.9.1</zipkin.version>
+        <caffeine.version>2.6.2</caffeine.version>
     </properties>
 
     <dependencies>
@@ -277,6 +279,18 @@
                 <artifactId>HikariCP</artifactId>
                 <version>${hikaricp.version}</version>
             </dependency>
+            <!-- for zipkin receiver -->
+            <dependency>
+                <groupId>io.zipkin.zipkin2</groupId>
+                <artifactId>zipkin</artifactId>
+                <version>${zipkin.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>com.github.ben-manes.caffeine</groupId>
+                <artifactId>caffeine</artifactId>
+                <version>${caffeine.version}</version>
+            </dependency>
+            <!-- -->
         </dependencies>
     </dependencyManagement>
 </project>

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/module/TraceModule.java

@@ -19,17 +19,19 @@
 package org.apache.skywalking.oap.server.receiver.trace.module;
 
 import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.ISegmentParserService;
 
 /**
  * @author peng-yongsheng
  */
 public class TraceModule extends ModuleDefine {
+    public static final String NAME = "receiver-trace";
 
     public TraceModule() {
-        super("receiver-trace");
+        super(NAME);
     }
 
     @Override public Class[] services() {
-        return new Class[0];
+        return new Class[] {ISegmentParserService.class};
     }
 }

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java

@@ -37,6 +37,7 @@
 public class TraceModuleProvider extends ModuleProvider {
 
     private final TraceServiceModuleConfig moduleConfig;
+    private SegmentParse.Producer segmentProducer;
 
     public TraceModuleProvider() {
         this.moduleConfig = new TraceServiceModuleConfig();
@@ -54,19 +55,21 @@ public TraceModuleProvider() {
         return moduleConfig;
     }
 
-    @Override public void prepare() {
-    }
-
-    @Override public void start() throws ModuleStartException {
+    @Override public void prepare() throws ServiceNotProvidedException {
         SegmentParserListenerManager listenerManager = new SegmentParserListenerManager();
         listenerManager.add(new MultiScopesSpanListener.Factory());
         listenerManager.add(new ServiceMappingSpanListener.Factory());
         listenerManager.add(new SegmentSpanListener.Factory());
 
+        segmentProducer = new SegmentParse.Producer(getManager(), listenerManager);
+        this.registerServiceImplementation(ISegmentParserService.class, new SegmentParserServiceImpl(segmentProducer));
+    }
+
+    @Override public void start() throws ModuleStartException {
         GRPCHandlerRegister grpcHandlerRegister = getManager().find(CoreModule.NAME).provider().getService(GRPCHandlerRegister.class);
         JettyHandlerRegister jettyHandlerRegister = getManager().find(CoreModule.NAME).provider().getService(JettyHandlerRegister.class);
         try {
-            SegmentParse.Producer segmentProducer = new SegmentParse.Producer(getManager(), listenerManager);
+
             grpcHandlerRegister.addHandler(new TraceSegmentServiceHandler(segmentProducer));
             jettyHandlerRegister.addHandler(new TraceSegmentServletHandler(segmentProducer));
 

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

@@ -0,0 +1,28 @@
+/*
+ * 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;
+
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.SpanListenerFactory;
+
+/**
+ * @author wusheng
+ */
+public interface ISegmentParserListenerManager {
+    void add(SpanListenerFactory spanListenerFactory);
+}

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

@@ -0,0 +1,29 @@
+/*
+ * 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;
+
+import org.apache.skywalking.apm.network.language.agent.UpstreamSegment;
+import org.apache.skywalking.oap.server.library.module.Service;
+
+/**
+ * @author wusheng
+ */
+public interface ISegmentParserService extends Service {
+    void send(UpstreamSegment segment);
+}

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

@@ -18,20 +18,22 @@
 
 package org.apache.skywalking.oap.server.receiver.trace.provider.parser;
 
-import java.util.*;
+import java.util.LinkedList;
+import java.util.List;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.SpanListenerFactory;
 
 /**
  * @author peng-yongsheng
  */
-public class SegmentParserListenerManager {
+public class SegmentParserListenerManager implements ISegmentParserListenerManager {
 
     private List<SpanListenerFactory> spanListenerFactories;
 
     public SegmentParserListenerManager() {
         this.spanListenerFactories = new LinkedList<>();
     }
 
+    @Override
     public void add(SpanListenerFactory spanListenerFactory) {
         spanListenerFactories.add(spanListenerFactory);
     }

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

@@ -0,0 +1,38 @@
+/*
+ * 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;
+
+import org.apache.skywalking.apm.network.language.agent.UpstreamSegment;
+
+/**
+ * @author wusheng
+ */
+public class SegmentParserServiceImpl implements ISegmentParserService {
+    private final SegmentParse.Producer segmentProducer;
+
+    public SegmentParserServiceImpl(
+        SegmentParse.Producer segmentProducer) {
+        this.segmentProducer = segmentProducer;
+    }
+
+    @Override
+    public void send(UpstreamSegment segment) {
+        segmentProducer.send(segment, SegmentParse.Source.Agent);
+    }
+}

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

@@ -35,9 +35,7 @@
  *
  * v5        |   v6
  *
- * 1. Application == Service
- * 2. Server == Service Instance
- * 3. Service == Endpoint
+ * 1. Application == Service 2. Server == Service Instance 3. Service == Endpoint
  *
  * @author peng-yongsheng, wusheng
  */
@@ -120,6 +118,9 @@ public void parseEntry(SpanDecorator spanDecorator, SegmentCoreInfo segmentCoreI
         SourceBuilder sourceBuilder = new SourceBuilder();
 
         int peerId = spanDecorator.getPeerId();
+        if (peerId == 0) {
+            return;
+        }
         int destServiceId = serviceInventoryCache.getServiceId(peerId);
         int mappingServiceId = serviceInventoryCache.get(destServiceId).getMappingServiceId();
         int destInstanceId = instanceInventoryCache.getServiceInstanceId(destServiceId, peerId);

oap-server/server-receiver-plugin/zipkin-receiver-plugin/pom.xml

@@ -17,7 +17,8 @@
   ~
   -->
 
-<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">
+<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>server-receiver-plugin</artifactId>
         <groupId>org.apache.skywalking</groupId>
@@ -28,4 +29,28 @@
     <artifactId>zipkin-receiver-plugin</artifactId>
     <packaging>jar</packaging>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>skywalking-trace-receiver-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>skywalking-register-receiver-plugin</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.zipkin.zipkin2</groupId>
+            <artifactId>zipkin</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.github.ben-manes.caffeine</groupId>
+            <artifactId>caffeine</artifactId>
+        </dependency>
+    </dependencies>
 </project>

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

@@ -0,0 +1,48 @@
+/*
+ * 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.zipkin;
+
+import org.apache.skywalking.oap.server.core.CoreModule;
+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;
+
+public class CoreRegisterLinker {
+    private static volatile ModuleManager MODULE_MANAGER;
+    private static volatile IServiceInventoryRegister SERVICE_INVENTORY_REGISTER;
+    private static volatile IServiceInstanceInventoryRegister SERVICE_INSTANCE_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).provider().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).provider().getService(IServiceInstanceInventoryRegister.class);
+        }
+        return SERVICE_INSTANCE_INVENTORY_REGISTER;
+    }
+}