Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Restore Zipkin receiver based on new core #1932

Merged
merged 7 commits into from
Nov 18, 2018
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 12 additions & 3 deletions docs/en/setup/backend/backend-receivers.md
Original file line number Diff line number Diff line change
Expand Up @@ -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: /
```
14 changes: 14 additions & 0 deletions oap-server/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down Expand Up @@ -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>
Original file line number Diff line number Diff line change
Expand Up @@ -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};
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@
public class TraceModuleProvider extends ModuleProvider {

private final TraceServiceModuleConfig moduleConfig;
private SegmentParse.Producer segmentProducer;

public TraceModuleProvider() {
this.moduleConfig = new TraceServiceModuleConfig();
Expand All @@ -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));

Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
Original file line number Diff line number Diff line change
@@ -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);
}
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Original file line number Diff line number Diff line change
@@ -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);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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
*/
Expand Down Expand Up @@ -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);
Expand Down
27 changes: 26 additions & 1 deletion oap-server/server-receiver-plugin/zipkin-receiver-plugin/pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand All @@ -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>
Original file line number Diff line number Diff line change
@@ -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;
}
}
Loading