You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ta...@apache.org on 2020/02/23 02:58:48 UTC
[skywalking] branch rest_api created (now 47bb0c0)
This is an automated email from the ASF dual-hosted git repository.
tanjian pushed a change to branch rest_api
in repository https://gitbox.apache.org/repos/asf/skywalking.git.
at 47bb0c0 support http api.
This branch includes the following new commits:
new 47bb0c0 support http api.
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: support http api.
Posted by ta...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
tanjian pushed a commit to branch rest_api
in repository https://gitbox.apache.org/repos/asf/skywalking.git
commit 47bb0c0247290e42923496b2b164eb75ec7cdfe6
Author: JaredTan95 <ji...@daocloud.io>
AuthorDate: Sun Feb 23 10:58:16 2020 +0800
support http api.
---
.../register/provider/RegisterModuleProvider.java | 11 ++
.../v6/rest/ServiceInstancePingServletHandler.java | 107 ++++++++++++++++++
.../ServiceInstanceRegisterServletHandler.java | 121 ++++++++++++++++++++
.../v6/rest/ServiceRegisterServletHandler.java | 80 +++++++++++++
.../trace/provider/TraceModuleProvider.java | 25 +++--
.../v6/rest/TraceSegmentCollectServletHandler.java | 82 ++++++++++++++
.../rest/reader/KeyStringValuePairJsonReader.java | 52 +++++++++
.../handler/v6/rest/reader/LogJsonReader.java | 58 ++++++++++
.../v6/rest/reader/ReferenceJsonReader.java | 90 +++++++++++++++
.../handler/v6/rest/reader/SegmentJsonReader.java | 80 +++++++++++++
.../handler/v6/rest/reader/SpanJsonReader.java | 124 +++++++++++++++++++++
.../handler/v6/rest/reader/StreamJsonReader.java | 26 +++++
.../handler/v6/rest/reader/TraceSegment.java | 44 ++++++++
.../handler/v6/rest/reader/UniqueIdJsonReader.java | 38 +++++++
.../v6/rest/reader/UpstreamSegmentJsonReader.java | 63 +++++++++++
15 files changed, 994 insertions(+), 7 deletions(-)
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
index 798dea3..f7b3dc0 100644
--- a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/RegisterModuleProvider.java
@@ -20,12 +20,16 @@ package org.apache.skywalking.oap.server.receiver.register.provider;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
+import org.apache.skywalking.oap.server.core.server.JettyHandlerRegister;
import org.apache.skywalking.oap.server.library.module.ModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
import org.apache.skywalking.oap.server.receiver.register.module.RegisterModule;
import org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.grpc.RegisterServiceHandler;
import org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.grpc.ServiceInstancePingServiceHandler;
+import org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.rest.ServiceInstancePingServletHandler;
+import org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.rest.ServiceInstanceRegisterServletHandler;
+import org.apache.skywalking.oap.server.receiver.register.provider.handler.v6.rest.ServiceRegisterServletHandler;
import org.apache.skywalking.oap.server.receiver.sharing.server.SharingServerModule;
public class RegisterModuleProvider extends ModuleProvider {
@@ -56,6 +60,13 @@ public class RegisterModuleProvider extends ModuleProvider {
.getService(GRPCHandlerRegister.class);
grpcHandlerRegister.addHandler(new RegisterServiceHandler(getManager()));
grpcHandlerRegister.addHandler(new ServiceInstancePingServiceHandler(getManager()));
+
+ JettyHandlerRegister jettyHandlerRegister = getManager().find(SharingServerModule.NAME)
+ .provider()
+ .getService(JettyHandlerRegister.class);
+ jettyHandlerRegister.addHandler(new ServiceRegisterServletHandler(getManager()));
+ jettyHandlerRegister.addHandler(new ServiceInstanceRegisterServletHandler(getManager()));
+ jettyHandlerRegister.addHandler(new ServiceInstancePingServletHandler(getManager()));
}
@Override
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstancePingServletHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstancePingServletHandler.java
new file mode 100644
index 0000000..2724036
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstancePingServletHandler.java
@@ -0,0 +1,107 @@
+/*
+ * 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.register.provider.handler.v6.rest;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import java.util.Objects;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.skywalking.apm.network.common.Command;
+import org.apache.skywalking.apm.network.common.Commands;
+import org.apache.skywalking.apm.network.trace.component.command.ServiceResetCommand;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cache.ServiceInstanceInventoryCache;
+import org.apache.skywalking.oap.server.core.command.CommandService;
+import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory;
+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;
+import org.apache.skywalking.oap.server.library.server.jetty.ArgumentsParseException;
+import org.apache.skywalking.oap.server.library.server.jetty.JettyJsonHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ServiceInstancePingServletHandler extends JettyJsonHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(ServiceInstancePingServletHandler.class);
+
+ private final IServiceInstanceInventoryRegister serviceInstanceInventoryRegister;
+ private final ServiceInstanceInventoryCache serviceInstanceInventoryCache;
+ private final IServiceInventoryRegister serviceInventoryRegister;
+ private final CommandService commandService;
+ private final Gson gson = new Gson();
+
+ private static final String INSTANCE_ID = "instance_id";
+ private static final String TIME = "time";
+ private static final String SERVICE_INSTANCE_UUID = "service_instance_UUID";
+ private static final String COMMANDS = "commands";
+
+ public ServiceInstancePingServletHandler(ModuleManager moduleManager) {
+ this.serviceInstanceInventoryRegister = moduleManager.find(CoreModule.NAME).provider().getService(
+ IServiceInstanceInventoryRegister.class);
+ this.serviceInstanceInventoryCache = moduleManager.find(CoreModule.NAME).provider().getService(
+ ServiceInstanceInventoryCache.class);
+ this.serviceInventoryRegister = moduleManager.find(CoreModule.NAME).provider().getService(
+ IServiceInventoryRegister.class);
+ this.commandService = moduleManager.find(CoreModule.NAME).provider().getService(CommandService.class);
+ }
+
+ @Override
+ public String pathSpec() {
+ return "/v2/instance/heartbeat";
+ }
+
+ @Override
+ protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException, IOException {
+ JsonObject responseJson = new JsonObject();
+ try {
+ JsonObject heartBeat = gson.fromJson(req.getReader(), JsonObject.class);
+ int instanceId = heartBeat.get(INSTANCE_ID).getAsInt();
+ long heartBeatTime = heartBeat.get(TIME).getAsLong();
+ String serviceInstanceUUID = heartBeat.get(SERVICE_INSTANCE_UUID).getAsString();
+
+ serviceInstanceInventoryRegister.heartbeat(instanceId, heartBeatTime);
+ ServiceInstanceInventory serviceInstanceInventory = serviceInstanceInventoryCache.get(instanceId);
+ if (Objects.nonNull(serviceInstanceInventory)) {
+ serviceInventoryRegister.heartbeat(serviceInstanceInventory.getServiceId(), heartBeatTime);
+ } else {
+ logger.warn(
+ "Can't found service by service instance id from cache, service instance id is: {}", instanceId);
+
+ final ServiceResetCommand resetCommand = commandService.newResetCommand(
+ instanceId, heartBeatTime, serviceInstanceUUID);
+ final Command command = resetCommand.serialize().build();
+ final Commands nextCommands = Commands.newBuilder().addCommands(command).build();
+ responseJson.add(COMMANDS, gson.toJsonTree(nextCommands, Commands.class));
+ }
+ responseJson.addProperty(INSTANCE_ID, instanceId);
+ responseJson.addProperty(TIME, heartBeatTime);
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+ return responseJson;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstanceRegisterServletHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstanceRegisterServletHandler.java
new file mode 100644
index 0000000..cefd13d
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceInstanceRegisterServletHandler.java
@@ -0,0 +1,121 @@
+/*
+ * 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.register.provider.handler.v6.rest;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cache.ServiceInventoryCache;
+import org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory;
+import org.apache.skywalking.oap.server.core.register.ServiceInventory;
+import org.apache.skywalking.oap.server.core.register.service.IServiceInstanceInventoryRegister;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.server.jetty.ArgumentsParseException;
+import org.apache.skywalking.oap.server.library.server.jetty.JettyJsonHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory.PropertyUtil.HOST_NAME;
+import static org.apache.skywalking.oap.server.core.register.ServiceInstanceInventory.PropertyUtil.PROCESS_NO;
+
+public class ServiceInstanceRegisterServletHandler extends JettyJsonHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(ServiceInstanceRegisterServletHandler.class);
+
+ private final IServiceInstanceInventoryRegister serviceInstanceInventoryRegister;
+ private final ServiceInventoryCache serviceInventoryCache;
+ private final Gson gson = new Gson();
+
+ private static final String SERVICE_ID = "service_id";
+ private static final String AGENT_UUID = "agent_uuid";
+ private static final String REGISTER_TIME = "register_time";
+ private static final String INSTANCE_ID = "instance_id";
+ private static final String OS_INFO = "os_info";
+
+ public ServiceInstanceRegisterServletHandler(ModuleManager moduleManager) {
+ this.serviceInventoryCache = moduleManager.find(CoreModule.NAME)
+ .provider()
+ .getService(ServiceInventoryCache.class);
+ this.serviceInstanceInventoryRegister = moduleManager.find(CoreModule.NAME).provider().getService(
+ IServiceInstanceInventoryRegister.class);
+ }
+
+ @Override
+ public String pathSpec() {
+ return "/v2/instance/register";
+ }
+
+ @Override
+ protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException {
+ JsonObject responseJson = new JsonObject();
+ try {
+ JsonObject instance = gson.fromJson(req.getReader(), JsonObject.class);
+ int serviceId = instance.get(SERVICE_ID).getAsInt();
+ String agentUUID = instance.get(AGENT_UUID).getAsString();
+ long registerTime = instance.get(REGISTER_TIME).getAsLong();
+ JsonObject osInfoJson = instance.get(OS_INFO).getAsJsonObject();
+
+ List<String> ipv4sList = new ArrayList<>();
+ JsonArray ipv4s = osInfoJson.get("ipv4s").getAsJsonArray();
+ ipv4s.forEach(ipv4 -> ipv4sList.add(ipv4.getAsString()));
+
+ ServiceInventory serviceInventory = serviceInventoryCache.get(serviceId);
+
+ JsonObject instanceProperties = new JsonObject();
+ instanceProperties.addProperty(
+ ServiceInstanceInventory.PropertyUtil.HOST_NAME, osInfoJson.get("host_name").getAsString());
+ instanceProperties.addProperty(
+ ServiceInstanceInventory.PropertyUtil.OS_NAME, osInfoJson.get("os_name").getAsString());
+ instanceProperties.addProperty(
+ ServiceInstanceInventory.PropertyUtil.PROCESS_NO, osInfoJson.get("process_id").getAsInt() + "");
+ instanceProperties.addProperty(
+ ServiceInstanceInventory.PropertyUtil.IPV4S,
+ ServiceInstanceInventory.PropertyUtil.ipv4sSerialize(ipv4sList)
+ );
+
+ String instanceName = serviceInventory.getName();
+ if (instanceProperties.has(PROCESS_NO)) {
+ instanceName += "-pid:" + instanceProperties.get(PROCESS_NO).getAsString();
+ }
+ if (instanceProperties.has(HOST_NAME)) {
+ instanceName += "@" + instanceProperties.get(HOST_NAME).getAsString();
+ }
+
+ int instanceId = serviceInstanceInventoryRegister.getOrCreate(
+ serviceId, instanceName, agentUUID, registerTime, instanceProperties);
+ responseJson.addProperty(SERVICE_ID, serviceId);
+ responseJson.addProperty(INSTANCE_ID, instanceId);
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+ //TODO: Commands
+ return responseJson;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceRegisterServletHandler.java b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceRegisterServletHandler.java
new file mode 100644
index 0000000..c33616e
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-register-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/register/provider/handler/v6/rest/ServiceRegisterServletHandler.java
@@ -0,0 +1,80 @@
+/*
+ * 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.register.provider.handler.v6.rest;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.register.service.IServiceInventoryRegister;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.server.jetty.ArgumentsParseException;
+import org.apache.skywalking.oap.server.library.server.jetty.JettyJsonHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ServiceRegisterServletHandler extends JettyJsonHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(ServiceRegisterServletHandler.class);
+
+ private final IServiceInventoryRegister serviceInventoryRegister;
+ private final Gson gson = new Gson();
+ private static final String SERVICE_NAME = "service_name";
+ private static final String SERVICE_ID = "service_id";
+
+ public ServiceRegisterServletHandler(ModuleManager moduleManager) {
+ serviceInventoryRegister = moduleManager.find(CoreModule.NAME)
+ .provider()
+ .getService(IServiceInventoryRegister.class);
+ }
+
+ @Override
+ public String pathSpec() {
+ return "/v2/service/register";
+ }
+
+ @Override
+ protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException {
+ JsonArray responseArray = new JsonArray();
+ try {
+ JsonArray serviceCodes = gson.fromJson(req.getReader(), JsonArray.class);
+ for (int i = 0; i < serviceCodes.size(); i++) {
+ JsonObject service = serviceCodes.get(i).getAsJsonObject();
+ String serviceCode = service.get(SERVICE_NAME).getAsString();
+ int serviceId = serviceInventoryRegister.getOrCreate(serviceCode, null);
+ JsonObject mapping = new JsonObject();
+ mapping.addProperty(SERVICE_NAME, serviceCode);
+ mapping.addProperty(SERVICE_ID, serviceId);
+ responseArray.add(mapping);
+ //
+ }
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+ return responseArray;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
index 3eb3d5f..b087241 100755
--- a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/TraceModuleProvider.java
@@ -18,10 +18,12 @@
package org.apache.skywalking.oap.server.receiver.trace.provider;
+import java.io.IOException;
import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule;
import org.apache.skywalking.oap.server.configuration.api.DynamicConfigurationService;
import org.apache.skywalking.oap.server.core.CoreModule;
import org.apache.skywalking.oap.server.core.server.GRPCHandlerRegister;
+import org.apache.skywalking.oap.server.core.server.JettyHandlerRegister;
import org.apache.skywalking.oap.server.library.module.ModuleConfig;
import org.apache.skywalking.oap.server.library.module.ModuleDefine;
import org.apache.skywalking.oap.server.library.module.ModuleProvider;
@@ -30,6 +32,7 @@ import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedExcepti
import org.apache.skywalking.oap.server.receiver.sharing.server.SharingServerModule;
import org.apache.skywalking.oap.server.receiver.trace.module.TraceModule;
import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v6.grpc.TraceSegmentReportServiceHandler;
+import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v6.rest.TraceSegmentCollectServletHandler;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.ISegmentParserService;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParseV2;
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserListenerManager;
@@ -41,8 +44,6 @@ import org.apache.skywalking.oap.server.receiver.trace.provider.parser.listener.
import org.apache.skywalking.oap.server.receiver.trace.provider.parser.standardization.SegmentStandardizationWorker;
import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
-import java.io.IOException;
-
public class TraceModuleProvider extends ModuleProvider {
private final TraceServiceModuleConfig moduleConfig;
@@ -80,7 +81,8 @@ public class TraceModuleProvider extends ModuleProvider {
segmentProducerV2 = new SegmentParseV2.Producer(getManager(), listenerManager(), moduleConfig);
- this.registerServiceImplementation(ISegmentParserService.class, new SegmentParserServiceImpl(segmentProducerV2));
+ this.registerServiceImplementation(
+ ISegmentParserService.class, new SegmentParserServiceImpl(segmentProducerV2));
}
public SegmentParserListenerManager listenerManager() {
@@ -99,19 +101,28 @@ public class TraceModuleProvider extends ModuleProvider {
public void start() throws ModuleStartException {
DynamicConfigurationService dynamicConfigurationService = getManager().find(ConfigurationModule.NAME)
.provider()
- .getService(DynamicConfigurationService.class);
+ .getService(
+ DynamicConfigurationService.class);
GRPCHandlerRegister grpcHandlerRegister = getManager().find(SharingServerModule.NAME)
.provider()
.getService(GRPCHandlerRegister.class);
+ JettyHandlerRegister jettyHandlerRegister = getManager().find(SharingServerModule.NAME)
+ .provider()
+ .getService(JettyHandlerRegister.class);
try {
dynamicConfigurationService.registerConfigChangeWatcher(thresholds);
dynamicConfigurationService.registerConfigChangeWatcher(uninstrumentedGatewaysConfig);
grpcHandlerRegister.addHandler(new TraceSegmentReportServiceHandler(segmentProducerV2, getManager()));
- SegmentStandardizationWorker standardizationWorkerV2 = new SegmentStandardizationWorker(getManager(), segmentProducerV2, moduleConfig
- .getBufferPath(), moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(), moduleConfig
- .isBufferFileCleanWhenRestart());
+ jettyHandlerRegister.addHandler(new TraceSegmentCollectServletHandler(segmentProducerV2));
+
+ SegmentStandardizationWorker standardizationWorkerV2 = new SegmentStandardizationWorker(
+ getManager(), segmentProducerV2, moduleConfig
+ .getBufferPath(), moduleConfig.getBufferOffsetMaxFileSize(), moduleConfig.getBufferDataMaxFileSize(),
+ moduleConfig
+ .isBufferFileCleanWhenRestart()
+ );
segmentProducerV2.setStandardizationWorker(standardizationWorkerV2);
} catch (IOException e) {
throw new ModuleStartException(e.getMessage(), e);
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/TraceSegmentCollectServletHandler.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/TraceSegmentCollectServletHandler.java
new file mode 100644
index 0000000..aadad13
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/TraceSegmentCollectServletHandler.java
@@ -0,0 +1,82 @@
+/*
+ * 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.handler.v6.rest;
+
+import com.google.gson.JsonElement;
+import com.google.gson.stream.JsonReader;
+import java.io.BufferedReader;
+import java.io.IOException;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.skywalking.oap.server.library.server.jetty.JettyJsonHandler;
+import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v6.rest.reader.TraceSegment;
+import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v6.rest.reader.UpstreamSegmentJsonReader;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParseV2;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentSource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TraceSegmentCollectServletHandler extends JettyJsonHandler {
+
+ private static final Logger logger = LoggerFactory.getLogger(TraceSegmentCollectServletHandler.class);
+
+ private final SegmentParseV2.Producer segmentProducer;
+
+ public TraceSegmentCollectServletHandler(SegmentParseV2.Producer segmentProducer) {
+ this.segmentProducer = segmentProducer;
+ }
+
+ @Override
+ public String pathSpec() {
+ return "/v2/segments";
+ }
+
+ @Override
+ protected JsonElement doGet(HttpServletRequest req) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected JsonElement doPost(HttpServletRequest req) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("receive stream segment");
+ }
+
+ try {
+ BufferedReader bufferedReader = req.getReader();
+ read(bufferedReader);
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+
+ return null;
+ }
+
+ private UpstreamSegmentJsonReader jsonReader = new UpstreamSegmentJsonReader();
+
+ private void read(BufferedReader bufferedReader) throws IOException {
+ JsonReader reader = new JsonReader(bufferedReader);
+
+ reader.beginArray();
+ while (reader.hasNext()) {
+ TraceSegment traceSegment = jsonReader.read(reader);
+ segmentProducer.send(traceSegment.getUpstreamSegment(), SegmentSource.Agent);
+ }
+ reader.endArray();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/KeyStringValuePairJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/KeyStringValuePairJsonReader.java
new file mode 100644
index 0000000..40ec472
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/KeyStringValuePairJsonReader.java
@@ -0,0 +1,52 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.common.KeyStringValuePair;
+
+public class KeyStringValuePairJsonReader implements StreamJsonReader<KeyStringValuePair> {
+
+ private static final String KEY = "key";
+ private static final String VALUE = "value";
+
+ @Override
+ public KeyStringValuePair read(JsonReader reader) throws IOException {
+ KeyStringValuePair.Builder builder = KeyStringValuePair.newBuilder();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case KEY:
+ builder.setKey(reader.nextString());
+ break;
+ case VALUE:
+ builder.setValue(reader.nextString());
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return builder.build();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/LogJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/LogJsonReader.java
new file mode 100644
index 0000000..9b24c5c
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/LogJsonReader.java
@@ -0,0 +1,58 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.language.agent.v2.Log;
+
+public class LogJsonReader implements StreamJsonReader<Log> {
+
+ private KeyStringValuePairJsonReader keyStringValuePairJsonReader = new KeyStringValuePairJsonReader();
+
+ private static final String TIME = "time";
+ private static final String DATA = "data";
+
+ @Override
+ public Log read(JsonReader reader) throws IOException {
+ Log.Builder builder = Log.newBuilder();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case TIME:
+ builder.setTime(reader.nextLong());
+ break;
+ case DATA:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addData(keyStringValuePairJsonReader.read(reader));
+ }
+ reader.endArray();
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return builder.build();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/ReferenceJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/ReferenceJsonReader.java
new file mode 100644
index 0000000..ed4653e
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/ReferenceJsonReader.java
@@ -0,0 +1,90 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.language.agent.v2.SegmentReference;
+
+public class ReferenceJsonReader implements StreamJsonReader<SegmentReference> {
+
+ private UniqueIdJsonReader uniqueIdJsonReader = new UniqueIdJsonReader();
+
+ private static final String REF_TYPE_VALUE = "ref_type";
+ private static final String PARENT_TRACE_SEGMENT_ID = "parent_trace_segment_id";
+ private static final String PARENT_SPAN_ID = "parent_span_id";
+ private static final String PARENT_SERVICE_INSTANCE_ID = "parent_service_instance_id";
+ private static final String NETWORK_ADDRESS = "network_address";
+ private static final String NETWORK_ADDRESS_ID = "network_address_id";
+ private static final String ENTRY_SERVICE_INSTANCE_ID = "entry_service_instance_id";
+ private static final String ENTRY_ENDPOINT = "entry_endpoint";
+ private static final String ENTRY_ENDPOINT_ID = "entry_endpoint_id";
+ private static final String PARENT_ENDPOINT = "parent_endpoint";
+ private static final String PARENT_ENDPOINT_ID = "parent_endpoint_id";
+
+ @Override
+ public SegmentReference read(JsonReader reader) throws IOException {
+ SegmentReference.Builder builder = SegmentReference.newBuilder();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case REF_TYPE_VALUE:
+ builder.setRefTypeValue(reader.nextInt());
+ break;
+ case PARENT_TRACE_SEGMENT_ID:
+ builder.setParentTraceSegmentId(uniqueIdJsonReader.read(reader));
+ break;
+ case PARENT_SPAN_ID:
+ builder.setParentSpanId(reader.nextInt());
+ break;
+ case PARENT_SERVICE_INSTANCE_ID:
+ builder.setParentServiceInstanceId(reader.nextInt());
+ break;
+ case NETWORK_ADDRESS:
+ builder.setNetworkAddress(reader.nextString());
+ break;
+ case NETWORK_ADDRESS_ID:
+ builder.setNetworkAddressId(reader.nextInt());
+ break;
+ case ENTRY_SERVICE_INSTANCE_ID:
+ builder.setEntryServiceInstanceId(reader.nextInt());
+ break;
+ case ENTRY_ENDPOINT:
+ builder.setEntryEndpoint(reader.nextString());
+ break;
+ case ENTRY_ENDPOINT_ID:
+ builder.setEntryEndpointId(reader.nextInt());
+ break;
+ case PARENT_ENDPOINT:
+ builder.setParentEndpoint(reader.nextString());
+ break;
+ case PARENT_ENDPOINT_ID:
+ builder.setParentEndpointId(reader.nextInt());
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return builder.build();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SegmentJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SegmentJsonReader.java
new file mode 100644
index 0000000..2982584
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SegmentJsonReader.java
@@ -0,0 +1,80 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.language.agent.v2.SegmentObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentJsonReader implements StreamJsonReader<SegmentObject.Builder> {
+
+ private static final Logger logger = LoggerFactory.getLogger(SegmentJsonReader.class);
+
+ private UniqueIdJsonReader uniqueIdJsonReader = new UniqueIdJsonReader();
+ private SpanJsonReader spanJsonReader = new SpanJsonReader();
+
+ private static final String TRACE_SEGMENT_ID = "trace_segment_id";
+ private static final String SPANS = "spans";
+ private static final String SERVICE_ID = "service_id";
+ private static final String SERVICE_INSTANCE_ID = "service_instance_id";
+ private static final String IS_SIZE_LIMITED = "is_size_limited";
+
+ @Override
+ public SegmentObject.Builder read(JsonReader reader) throws IOException {
+ SegmentObject.Builder builder = SegmentObject.newBuilder();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case TRACE_SEGMENT_ID:
+ builder.setTraceSegmentId(uniqueIdJsonReader.read(reader));
+ if (logger.isDebugEnabled()) {
+ StringBuilder segmentId = new StringBuilder();
+ builder.getTraceSegmentId().getIdPartsList().forEach(idPart -> segmentId.append(idPart));
+ logger.debug("segment id: {}", segmentId);
+ }
+ break;
+ case SERVICE_ID:
+ builder.setServiceId(reader.nextInt());
+ break;
+ case SERVICE_INSTANCE_ID:
+ builder.setServiceInstanceId(reader.nextInt());
+ break;
+ case IS_SIZE_LIMITED:
+ builder.setIsSizeLimited(reader.nextBoolean());
+ break;
+ case SPANS:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addSpans(spanJsonReader.read(reader));
+ }
+ reader.endArray();
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return builder;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SpanJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SpanJsonReader.java
new file mode 100644
index 0000000..11d51f3
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/SpanJsonReader.java
@@ -0,0 +1,124 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.language.agent.v2.SpanObjectV2;
+
+public class SpanJsonReader implements StreamJsonReader<SpanObjectV2> {
+
+ private KeyStringValuePairJsonReader keyStringValuePairJsonReader = new KeyStringValuePairJsonReader();
+ private LogJsonReader logJsonReader = new LogJsonReader();
+ private ReferenceJsonReader referenceJsonReader = new ReferenceJsonReader();
+
+ private static final String SPAN_ID = "span_id";
+ private static final String PARENT_SPAN_ID = "parent_span_id";
+ private static final String START_TIME = "start_time";
+ private static final String END_TIME = "end_time";
+ private static final String REFS = "refs";
+ private static final String OPERATION_NAME_ID = "operation_name_id";
+ private static final String OPERATION_NAME = "operation_name";
+ private static final String PEER_ID = "peer_id";
+ private static final String PEER = "peer";
+ private static final String SPAN_TYPE = "span_type";
+ private static final String SPAN_LAYER = "span_layer";
+ private static final String COMPONENT_ID = "component_id";
+ private static final String COMPONENT = "component";
+ private static final String IS_ERROR = "is_error";
+ private static final String TAGS = "tags";
+ private static final String LOGS = "logs";
+
+ @Override
+ public SpanObjectV2 read(JsonReader reader) throws IOException {
+ SpanObjectV2.Builder builder = SpanObjectV2.newBuilder();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case SPAN_ID:
+ builder.setSpanId(reader.nextInt());
+ break;
+ case SPAN_TYPE:
+ builder.setSpanTypeValue(reader.nextInt());
+ break;
+ case SPAN_LAYER:
+ builder.setSpanLayerValue(reader.nextInt());
+ break;
+ case PARENT_SPAN_ID:
+ builder.setParentSpanId(reader.nextInt());
+ break;
+ case START_TIME:
+ builder.setStartTime(reader.nextLong());
+ break;
+ case END_TIME:
+ builder.setEndTime(reader.nextLong());
+ break;
+ case COMPONENT_ID:
+ builder.setComponentId(reader.nextInt());
+ break;
+ case COMPONENT:
+ builder.setComponent(reader.nextString());
+ break;
+ case OPERATION_NAME_ID:
+ builder.setOperationNameId(reader.nextInt());
+ break;
+ case OPERATION_NAME:
+ builder.setOperationName(reader.nextString());
+ break;
+ case PEER_ID:
+ builder.setPeerId(reader.nextInt());
+ break;
+ case PEER:
+ builder.setPeer(reader.nextString());
+ break;
+ case IS_ERROR:
+ builder.setIsError(reader.nextBoolean());
+ break;
+ case REFS:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addRefs(referenceJsonReader.read(reader));
+ }
+ reader.endArray();
+ break;
+ case TAGS:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addTags(keyStringValuePairJsonReader.read(reader));
+ }
+ reader.endArray();
+ break;
+ case LOGS:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addLogs(logJsonReader.read(reader));
+ }
+ reader.endArray();
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return builder.build();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/StreamJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/StreamJsonReader.java
new file mode 100644
index 0000000..5e7e461
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/StreamJsonReader.java
@@ -0,0 +1,26 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+
+public interface StreamJsonReader<T> {
+ T read(JsonReader reader) throws IOException;
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/TraceSegment.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/TraceSegment.java
new file mode 100644
index 0000000..fe1f2f7
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/TraceSegment.java
@@ -0,0 +1,44 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import org.apache.skywalking.apm.network.language.agent.UniqueId;
+import org.apache.skywalking.apm.network.language.agent.UpstreamSegment;
+import org.apache.skywalking.apm.network.language.agent.v2.SegmentObject;
+
+public class TraceSegment {
+
+ private UpstreamSegment.Builder builder;
+
+ public TraceSegment() {
+ builder = UpstreamSegment.newBuilder();
+ }
+
+ public void addGlobalTraceId(UniqueId.Builder globalTraceId) {
+ builder.addGlobalTraceIds(globalTraceId);
+ }
+
+ public void setTraceSegmentBuilder(SegmentObject.Builder traceSegmentBuilder) {
+ builder.setSegment(traceSegmentBuilder.build().toByteString());
+ }
+
+ public UpstreamSegment getUpstreamSegment() {
+ return builder.build();
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UniqueIdJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UniqueIdJsonReader.java
new file mode 100644
index 0000000..ec87fea
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UniqueIdJsonReader.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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.apache.skywalking.apm.network.language.agent.UniqueId;
+
+public class UniqueIdJsonReader implements StreamJsonReader<UniqueId.Builder> {
+
+ @Override
+ public UniqueId.Builder read(JsonReader reader) throws IOException {
+ UniqueId.Builder builder = UniqueId.newBuilder();
+
+ reader.beginArray();
+ while (reader.hasNext()) {
+ builder.addIdParts(reader.nextLong());
+ }
+ reader.endArray();
+ return builder;
+ }
+}
diff --git a/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UpstreamSegmentJsonReader.java b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UpstreamSegmentJsonReader.java
new file mode 100644
index 0000000..c47583c
--- /dev/null
+++ b/oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v6/rest/reader/UpstreamSegmentJsonReader.java
@@ -0,0 +1,63 @@
+/*
+ * 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.handler.v6.rest.reader;
+
+import com.google.gson.stream.JsonReader;
+import java.io.IOException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class UpstreamSegmentJsonReader implements StreamJsonReader<TraceSegment> {
+
+ private static final Logger logger = LoggerFactory.getLogger(UpstreamSegmentJsonReader.class);
+
+ private UniqueIdJsonReader uniqueIdJsonReader = new UniqueIdJsonReader();
+ private SegmentJsonReader segmentJsonReader = new SegmentJsonReader();
+
+ private static final String GLOBAL_TRACE_IDS = "global_trace_ids";
+ private static final String SEGMENT = "segment";
+
+ @Override
+ public TraceSegment read(JsonReader reader) throws IOException {
+ TraceSegment traceSegment = new TraceSegment();
+
+ reader.beginObject();
+ while (reader.hasNext()) {
+ switch (reader.nextName()) {
+ case GLOBAL_TRACE_IDS:
+ reader.beginArray();
+ while (reader.hasNext()) {
+ traceSegment.addGlobalTraceId(uniqueIdJsonReader.read(reader));
+ }
+ reader.endArray();
+
+ break;
+ case SEGMENT:
+ traceSegment.setTraceSegmentBuilder(segmentJsonReader.read(reader));
+ break;
+ default:
+ reader.skipValue();
+ break;
+ }
+ }
+ reader.endObject();
+
+ return traceSegment;
+ }
+}