You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ke...@apache.org on 2022/03/15 07:10:49 UTC
[skywalking] branch zipkin created (now ae67fb8)
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a change to branch zipkin
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
at ae67fb8 Polish Zipkin HTTP receiver with Armeria APIs
This branch includes the following new commits:
new ae67fb8 Polish Zipkin HTTP receiver with Armeria APIs
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[skywalking] 01/01: Polish Zipkin HTTP receiver with Armeria APIs
Posted by ke...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kezhenxu94 pushed a commit to branch zipkin
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit ae67fb83def05a4c70c21d433f0d43b6b6e5a7bd
Author: kezhenxu94 <ke...@apache.org>
AuthorDate: Tue Mar 15 14:23:23 2022 +0800
Polish Zipkin HTTP receiver with Armeria APIs
---
.../receiver/zipkin/ZipkinReceiverConfig.java | 8 +-
.../receiver/zipkin/ZipkinReceiverProvider.java | 14 +--
.../ConsumesThrift.java} | 31 +++--
.../receiver/zipkin/handler/SpanProcessor.java | 71 ------------
.../receiver/zipkin/handler/SpanV1HTTPHandler.java | 78 -------------
.../receiver/zipkin/handler/SpanV2HTTPHandler.java | 80 -------------
.../handler/UnzippingBytesRequestConverter.java | 41 +++++++
.../zipkin/handler/ZipkinSpanHTTPHandler.java | 126 +++++++++++++++++++++
.../server/receiver/zipkin/trace/SpanForward.java | 2 +-
9 files changed, 191 insertions(+), 260 deletions(-)
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java
index 7577470..fe42705 100644
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java
@@ -31,9 +31,9 @@ public class ZipkinReceiverConfig extends ModuleConfig {
private String host;
private int port;
private String contextPath;
- private int jettyMaxThreads = 200;
- private long jettyIdleTimeOut = 30000;
- private int jettyAcceptorPriorityDelta = 0;
- private int jettyAcceptQueueSize = 0;
+ private int maxThreads = 200;
+ private long idleTimeOut = 30000;
+ private int acceptorPriorityDelta = 0;
+ private int acceptQueueSize = 0;
private List<String> instanceNameRule = new ArrayList<>();
}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverProvider.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverProvider.java
index 4508689..d18e57e 100644
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverProvider.java
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverProvider.java
@@ -26,8 +26,7 @@ import org.apache.skywalking.oap.server.library.module.ModuleStartException;
import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
import org.apache.skywalking.oap.server.library.server.http.HTTPServer;
import org.apache.skywalking.oap.server.library.server.http.HTTPServerConfig;
-import org.apache.skywalking.oap.server.receiver.zipkin.handler.SpanV1HTTPHandler;
-import org.apache.skywalking.oap.server.receiver.zipkin.handler.SpanV2HTTPHandler;
+import org.apache.skywalking.oap.server.receiver.zipkin.handler.ZipkinSpanHTTPHandler;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
public class ZipkinReceiverProvider extends ModuleProvider {
@@ -65,18 +64,17 @@ public class ZipkinReceiverProvider extends ModuleProvider {
.host(config.getHost())
.port(config.getPort())
.contextPath(config.getContextPath())
- .idleTimeOut(config.getJettyIdleTimeOut())
+ .idleTimeOut(config.getIdleTimeOut())
.acceptorPriorityDelta(
- config.getJettyAcceptorPriorityDelta())
- .maxThreads(config.getJettyMaxThreads())
- .acceptQueueSize(config.getJettyAcceptQueueSize())
+ config.getAcceptorPriorityDelta())
+ .maxThreads(config.getMaxThreads())
+ .acceptQueueSize(config.getAcceptQueueSize())
.build();
httpServer = new HTTPServer(httpServerConfig);
httpServer.initialize();
- httpServer.addHandler(new SpanV1HTTPHandler(config, getManager()));
- httpServer.addHandler(new SpanV2HTTPHandler(config, getManager()));
+ httpServer.addHandler(new ZipkinSpanHTTPHandler(config, getManager()));
}
@Override
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ConsumesThrift.java
similarity index 57%
copy from oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java
copy to oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ConsumesThrift.java
index 7577470..1b45f6b 100644
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/ZipkinReceiverConfig.java
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ConsumesThrift.java
@@ -16,24 +16,19 @@
*
*/
-package org.apache.skywalking.oap.server.receiver.zipkin;
+package org.apache.skywalking.oap.server.receiver.zipkin.handler;
-import lombok.Getter;
-import lombok.Setter;
-import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+import com.linecorp.armeria.server.annotation.Consumes;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
-import java.util.ArrayList;
-import java.util.List;
-
-@Setter
-@Getter
-public class ZipkinReceiverConfig extends ModuleConfig {
- private String host;
- private int port;
- private String contextPath;
- private int jettyMaxThreads = 200;
- private long jettyIdleTimeOut = 30000;
- private int jettyAcceptorPriorityDelta = 0;
- private int jettyAcceptQueueSize = 0;
- private List<String> instanceNameRule = new ArrayList<>();
+@Retention(RetentionPolicy.RUNTIME)
+@Target({
+ ElementType.TYPE,
+ ElementType.METHOD
+})
+@Consumes("application/x-thrift")
+@interface ConsumesThrift {
}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java
deleted file mode 100644
index 93e08a8..0000000
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanProcessor.java
+++ /dev/null
@@ -1,71 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.oap.server.receiver.zipkin.handler;
-
-import com.linecorp.armeria.common.HttpData;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.List;
-import java.util.zip.GZIPInputStream;
-
-import lombok.RequiredArgsConstructor;
-import org.apache.skywalking.oap.server.core.config.NamingControl;
-import org.apache.skywalking.oap.server.core.source.SourceReceiver;
-import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverConfig;
-import org.apache.skywalking.oap.server.receiver.zipkin.trace.SpanForward;
-import zipkin2.Span;
-import zipkin2.codec.SpanBytesDecoder;
-
-@RequiredArgsConstructor
-public class SpanProcessor {
- private final NamingControl namingControl;
- private final SourceReceiver receiver;
-
- void convert(ZipkinReceiverConfig config, SpanBytesDecoder decoder,
- String contentEncoding, HttpData request) throws IOException {
- try (InputStream inputStream = getInputStream(contentEncoding, request)) {
- ByteArrayOutputStream out = new ByteArrayOutputStream();
- byte[] buffer = new byte[2048];
- int readCntOnce;
-
- while ((readCntOnce = inputStream.read(buffer)) >= 0) {
- out.write(buffer, 0, readCntOnce);
- }
-
- List<Span> spanList = decoder.decodeList(out.toByteArray());
-
- SpanForward forward = new SpanForward(namingControl, receiver, config);
- forward.send(spanList);
- }
- }
-
- private InputStream getInputStream(String headEncoding, HttpData request) throws IOException {
- InputStream requestInStream;
-
- if (headEncoding != null && headEncoding.contains("gzip")) {
- requestInStream = new GZIPInputStream(request.toInputStream());
- } else {
- requestInStream = request.toInputStream();
- }
-
- return requestInStream;
- }
-
-}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV1HTTPHandler.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV1HTTPHandler.java
deleted file mode 100644
index 318d60b..0000000
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV1HTTPHandler.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.oap.server.receiver.zipkin.handler;
-
-import com.linecorp.armeria.common.HttpData;
-import com.linecorp.armeria.server.annotation.Header;
-import com.linecorp.armeria.server.annotation.Post;
-import com.linecorp.armeria.server.annotation.StatusCode;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.config.NamingControl;
-import org.apache.skywalking.oap.server.core.source.SourceReceiver;
-import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverConfig;
-import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
-import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
-import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
-import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
-import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
-import zipkin2.codec.SpanBytesDecoder;
-
-@Slf4j
-public class SpanV1HTTPHandler {
- private final ZipkinReceiverConfig config;
- private final SourceReceiver sourceReceiver;
- private final NamingControl namingControl;
- private final HistogramMetrics histogram;
- private final CounterMetrics errorCounter;
-
- public SpanV1HTTPHandler(ZipkinReceiverConfig config, ModuleManager manager) {
- sourceReceiver = manager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
- namingControl = manager.find(CoreModule.NAME).provider().getService(NamingControl.class);
- this.config = config;
- MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME)
- .provider()
- .getService(MetricsCreator.class);
- histogram = metricsCreator.createHistogramMetric(
- "trace_in_latency", "The process latency of trace data",
- new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-v1")
- );
- errorCounter = metricsCreator.createCounter("trace_analysis_error_count", "The error number of trace analysis",
- new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-v1")
- );
- }
-
- @Post("/api/v1/spans")
- @StatusCode(202)
- public void apiV1Spans(@Header("Content-Encoding") final String type,
- final HttpData request) {
- try (HistogramMetrics.Timer ignored = histogram.createTimer()) {
- int encode = type != null && type.contains("/x-thrift") ? SpanEncode.THRIFT : SpanEncode.JSON_V1;
-
- SpanBytesDecoder decoder = SpanEncode.isThrift(encode) ? SpanBytesDecoder.THRIFT : SpanBytesDecoder.JSON_V1;
-
- SpanProcessor processor = new SpanProcessor(namingControl, sourceReceiver);
- processor.convert(config, decoder, type, request);
- } catch (Throwable e) {
- errorCounter.inc();
- throw new RuntimeException(e);
- }
- }
-}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV2HTTPHandler.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV2HTTPHandler.java
deleted file mode 100644
index 777d694..0000000
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/SpanV2HTTPHandler.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.skywalking.oap.server.receiver.zipkin.handler;
-
-import com.linecorp.armeria.common.HttpData;
-import com.linecorp.armeria.server.annotation.Header;
-import com.linecorp.armeria.server.annotation.Post;
-import com.linecorp.armeria.server.annotation.StatusCode;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.config.NamingControl;
-import org.apache.skywalking.oap.server.core.source.SourceReceiver;
-import org.apache.skywalking.oap.server.library.module.ModuleManager;
-import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverConfig;
-import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
-import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
-import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
-import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
-import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
-import zipkin2.codec.SpanBytesDecoder;
-
-@Slf4j
-public class SpanV2HTTPHandler {
-
- private final ZipkinReceiverConfig config;
- private final SourceReceiver sourceReceiver;
- private final NamingControl namingControl;
- private final HistogramMetrics histogram;
- private final CounterMetrics errorCounter;
-
- public SpanV2HTTPHandler(ZipkinReceiverConfig config, ModuleManager manager) {
- sourceReceiver = manager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
- namingControl = manager.find(CoreModule.NAME).provider().getService(NamingControl.class);
- this.config = config;
- MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME)
- .provider()
- .getService(MetricsCreator.class);
- histogram = metricsCreator.createHistogramMetric(
- "trace_in_latency", "The process latency of trace data",
- new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-v2")
- );
- errorCounter = metricsCreator.createCounter("trace_analysis_error_count", "The error number of trace analysis",
- new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin-v2")
- );
- }
-
- @Post("/api/v2/spans")
- @StatusCode(202)
- public void apiV2Spans(@Header("Content-Encoding") final String type,
- HttpData request) {
- try (HistogramMetrics.Timer ignored = histogram.createTimer()) {
- int encode = type != null && type.contains("/x-protobuf") ? SpanEncode.PROTO3 : SpanEncode.JSON_V2;
-
- SpanBytesDecoder decoder = SpanEncode.isProto3(encode) ? SpanBytesDecoder.PROTO3 : SpanBytesDecoder.JSON_V2;
-
- SpanProcessor processor = new SpanProcessor(namingControl, sourceReceiver);
- processor.convert(config, decoder, type, request);
- } catch (Exception e) {
- errorCounter.inc();
- log.error(e.getMessage(), e);
- throw new RuntimeException(e);
- }
- }
-}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/UnzippingBytesRequestConverter.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/UnzippingBytesRequestConverter.java
new file mode 100644
index 0000000..412ccac
--- /dev/null
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/UnzippingBytesRequestConverter.java
@@ -0,0 +1,41 @@
+/*
+ * 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.handler;
+
+import com.linecorp.armeria.common.AggregatedHttpRequest;
+import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpHeaderNames;
+import com.linecorp.armeria.common.encoding.StreamDecoderFactory;
+import com.linecorp.armeria.server.ServiceRequestContext;
+
+final class UnzippingBytesRequestConverter {
+
+ static HttpData convertRequest(ServiceRequestContext ctx, AggregatedHttpRequest request) {
+ String encoding = request.headers().get(HttpHeaderNames.CONTENT_ENCODING);
+ HttpData content = request.content();
+ if (!content.isEmpty() && encoding != null && encoding.contains("gzip")) {
+ content = StreamDecoderFactory.gzip().newDecoder(ctx.alloc()).decode(content);
+ if (content.isEmpty()) {
+ content.close();
+ throw new IllegalArgumentException("Cannot unzip request content bytes");
+ }
+ }
+ return content;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ZipkinSpanHTTPHandler.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ZipkinSpanHTTPHandler.java
new file mode 100644
index 0000000..deacc8b
--- /dev/null
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/handler/ZipkinSpanHTTPHandler.java
@@ -0,0 +1,126 @@
+/*
+ * 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.handler;
+
+import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.server.annotation.ConsumesJson;
+import com.linecorp.armeria.server.annotation.ConsumesProtobuf;
+import com.linecorp.armeria.server.annotation.Post;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.config.NamingControl;
+import org.apache.skywalking.oap.server.core.source.SourceReceiver;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.receiver.zipkin.ZipkinReceiverConfig;
+import org.apache.skywalking.oap.server.receiver.zipkin.trace.SpanForward;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
+import org.apache.skywalking.oap.server.telemetry.api.HistogramMetrics;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
+import zipkin2.Span;
+import zipkin2.codec.SpanBytesDecoder;
+
+import static java.util.Objects.nonNull;
+
+@Slf4j
+public class ZipkinSpanHTTPHandler {
+ private final ZipkinReceiverConfig config;
+ private final SourceReceiver sourceReceiver;
+ private final NamingControl namingControl;
+ private final HistogramMetrics histogram;
+ private final CounterMetrics errorCounter;
+
+ public ZipkinSpanHTTPHandler(ZipkinReceiverConfig config, ModuleManager manager) {
+ sourceReceiver = manager.find(CoreModule.NAME).provider().getService(SourceReceiver.class);
+ namingControl = manager.find(CoreModule.NAME).provider().getService(NamingControl.class);
+ this.config = config;
+ MetricsCreator metricsCreator = manager.find(TelemetryModule.NAME)
+ .provider()
+ .getService(MetricsCreator.class);
+ histogram = metricsCreator.createHistogramMetric(
+ "trace_in_latency", "The process latency of trace data",
+ new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin")
+ );
+ errorCounter = metricsCreator.createCounter(
+ "trace_analysis_error_count", "The error number of trace analysis",
+ new MetricsTag.Keys("protocol"), new MetricsTag.Values("zipkin")
+ );
+ }
+
+ @Post("/api/v2/spans")
+ public HttpResponse collectV2Spans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.JSON_V2, ctx, req);
+ }
+
+ @Post("/api/v2/spans")
+ @ConsumesJson
+ public HttpResponse collectV2JsonSpans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.JSON_V2, ctx, req);
+ }
+
+ @Post("/api/v2/spans")
+ @ConsumesProtobuf
+ public HttpResponse collectV2ProtobufSpans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.PROTO3, ctx, req);
+ }
+
+ @Post("/api/v1/spans")
+ public HttpResponse collectV1Spans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.JSON_V1, ctx, req);
+ }
+
+ @Post("/api/v1/spans")
+ @ConsumesJson
+ public HttpResponse collectV1JsonSpans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.JSON_V1, ctx, req);
+ }
+
+ @Post("/api/v1/spans")
+ @ConsumesThrift
+ public HttpResponse collectV1ThriftSpans(ServiceRequestContext ctx, HttpRequest req) {
+ return doCollectSpans(SpanBytesDecoder.THRIFT, ctx, req);
+ }
+
+ HttpResponse doCollectSpans(final SpanBytesDecoder decoder,
+ final ServiceRequestContext ctx,
+ final HttpRequest req) {
+ final HistogramMetrics.Timer timer = histogram.createTimer();
+ final HttpResponse response = HttpResponse.from(req.aggregate().thenApply(request -> {
+ final HttpData httpData = UnzippingBytesRequestConverter.convertRequest(ctx, request);
+ final List<Span> spanList = decoder.decodeList(httpData.byteBuf().array());
+ final SpanForward forward = new SpanForward(namingControl, sourceReceiver, config);
+ forward.send(spanList);
+ return HttpResponse.of(HttpStatus.OK);
+ }));
+ response.whenComplete().handle((unused, throwable) -> {
+ if (nonNull(throwable)) {
+ errorCounter.inc();
+ }
+ timer.close();
+ return null;
+ });
+ return response;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java
index ba32102..77b15d5 100644
--- a/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java
+++ b/oap-server/server-receiver-plugin/zipkin-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/zipkin/trace/SpanForward.java
@@ -37,7 +37,7 @@ import zipkin2.codec.SpanBytesEncoder;
@RequiredArgsConstructor
public class SpanForward {
- private static final String DEFAULT_SERVICE_INSTANCE_NAME = " unknown_instance";
+ private static final String DEFAULT_SERVICE_INSTANCE_NAME = "unknown_instance";
private final NamingControl namingControl;
private final SourceReceiver receiver;
private final ZipkinReceiverConfig config;