Support multiple ways to analyze segments in rest (#4745)

* support multiple segments analyze

* Support multiple ways to analyze segment data

* Change to support single or list of the segment data, add the document

* fix document and change the lua agent commit id
Co-authored-by: NMrproliu <mrproliu@lagou.com>

docs/en/protocols/HTTP-API-Protocol.md

 # HTTP API Protocol
 
 HTTP API Protocol defines the API data format, including api request and response data format.
+They use the HTTP1.1 wrapper of the official [SkyWalking Trace Data Protocol v3](Trace-Data-Protocol-v3.md). Read it for more details.
 
-HTTP APIs are the HTTP1.1 wrapper of the official [SkyWalking Trace Data Protocol v3](Trace-Data-Protocol-v3.md). Read it for more details.
+## Instance Management
 
+Detail information about data format can be found in [Instance Management](https://github.com/apache/skywalking-data-collect-protocol/blob/master/management/Management.proto).
+
+- Report service instance properties
+
+> POST http://localhost:12800/v3/management/reportProperties
+
+Input:
+
+```json
+{
+	"service": "User Service Name",
+	"serviceInstance": "User Service Instance Name",
+	"properties": [{
+		"language": "Lua"
+	}]
+}
+```
+
+Output JSON Array:
+
+```json
+{}
+```
+
+- Service instance ping
+
+> POST http://localhost:12800/v3/management/keepAlive
+
+Input:
+
+```json
+{
+	"service": "User Service Name",
+	"serviceInstance": "User Service Instance Name"
+}
+```
+
+OutPut:
+
+```json
+{}
+```
+
+## Trace Report
+
+Detail information about data format can be found in [Instance Management](https://github.com/apache/skywalking-data-collect-protocol/blob/master/language-agent/Tracing.proto).
+There are two ways to report segment data, one segment per request or segment array in the bulk mode.
+
+### POST http://localhost:12800/v3/segment
+
+Send a single segment object with JSON format.
+
+Input:
+
+```json
+{
+	"traceId": "a12ff60b-5807-463b-a1f8-fb1c8608219e",
+	"serviceInstance": "User_Service_Instance_Name",
+	"spans": [{
+		"operationName": "/ingress",
+		"startTime": 1588664577013,
+		"endTime": 1588664577028,
+		"spanType": "Exit",
+		"spanId": 1,
+		"isError": false,
+		"parentSpanId": 0,
+		"componentId": 6000,
+		"peer": "upstream service",
+		"spanLayer": "Http"
+	}, {
+		"operationName": "/ingress",
+		"startTime": 1588664577013,
+		"tags": [{
+			"key": "http.method",
+			"value": "GET"
+		}, {
+			"key": "http.params",
+			"value": "http://localhost/ingress"
+		}],
+		"endTime": 1588664577028,
+		"spanType": "Entry",
+		"spanId": 0,
+		"parentSpanId": -1,
+		"isError": false,
+		"spanLayer": "Http",
+		"componentId": 6000
+	}],
+	"service": "User_Service_Name",
+	"traceSegmentId": "a12ff60b-5807-463b-a1f8-fb1c8608219e"
+}
+```
+ OutPut:
+ 
+ ```json
+
+```
+
+### POST http://localhost:12800/v3/segments
+
+Send a segment object list with JSON format.
+
+Input:
+
+```json
+[{
+	"traceId": "a12ff60b-5807-463b-a1f8-fb1c8608219e",
+	"serviceInstance": "User_Service_Instance_Name",
+	"spans": [{
+		"operationName": "/ingress",
+		"startTime": 1588664577013,
+		"endTime": 1588664577028,
+		"spanType": "Exit",
+		"spanId": 1,
+		"isError": false,
+		"parentSpanId": 0,
+		"componentId": 6000,
+		"peer": "upstream service",
+		"spanLayer": "Http"
+	}, {
+		"operationName": "/ingress",
+		"startTime": 1588664577013,
+		"tags": [{
+			"key": "http.method",
+			"value": "GET"
+		}, {
+			"key": "http.params",
+			"value": "http://localhost/ingress"
+		}],
+		"endTime": 1588664577028,
+		"spanType": "Entry",
+		"spanId": 0,
+		"parentSpanId": -1,
+		"isError": false,
+		"spanLayer": "Http",
+		"componentId": 6000
+	}],
+	"service": "User_Service_Name",
+	"traceSegmentId": "a12ff60b-5807-463b-a1f8-fb1c8608219e"
+}, {
+	"traceId": "f956699e-5106-4ea3-95e5-da748c55bac1",
+	"serviceInstance": "User_Service_Instance_Name",
+	"spans": [{
+		"operationName": "/ingress",
+		"startTime": 1588664577250,
+		"endTime": 1588664577250,
+		"spanType": "Exit",
+		"spanId": 1,
+		"isError": false,
+		"parentSpanId": 0,
+		"componentId": 6000,
+		"peer": "upstream service",
+		"spanLayer": "Http"
+	}, {
+		"operationName": "/ingress",
+		"startTime": 1588664577250,
+		"tags": [{
+			"key": "http.method",
+			"value": "GET"
+		}, {
+			"key": "http.params",
+			"value": "http://localhost/ingress"
+		}],
+		"endTime": 1588664577250,
+		"spanType": "Entry",
+		"spanId": 0,
+		"parentSpanId": -1,
+		"isError": false,
+		"spanLayer": "Http",
+		"componentId": 6000
+	}],
+	"service": "User_Service_Name",
+	"traceSegmentId": "f956699e-5106-4ea3-95e5-da748c55bac1"
+}]
+```
+ OutPut:
+ 
+ ```json
+
+```
\ No newline at end of file

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

@@ -31,7 +31,8 @@ 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.v8.grpc.TraceSegmentReportServiceHandler;
-import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest.TraceSegmentReportServletHandler;
+import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest.TraceSegmentReportListServletHandler;
+import org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest.TraceSegmentReportSingleServletHandler;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.ISegmentParserService;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserListenerManager;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserServiceImpl;
@@ -99,7 +100,9 @@ public class TraceModuleProvider extends ModuleProvider {
             new TraceSegmentReportServiceHandler(getManager(), listenerManager(), moduleConfig));
 
         jettyHandlerRegister.addHandler(
-            new TraceSegmentReportServletHandler(getManager(), listenerManager(), moduleConfig));
+            new TraceSegmentReportListServletHandler(getManager(), listenerManager(), moduleConfig));
+        jettyHandlerRegister.addHandler(
+            new TraceSegmentReportSingleServletHandler(getManager(), listenerManager(), moduleConfig));
     }
 
     @Override

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/rest/TraceSegmentReportServletHandler.java → oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/rest/TraceSegmentReportBaseServletHandler.java

@@ -19,13 +19,13 @@
 package org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest;
 
 import com.google.gson.JsonElement;
-import java.io.BufferedReader;
+import java.io.IOException;
+import java.util.List;
 import javax.servlet.http.HttpServletRequest;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
 import org.apache.skywalking.oap.server.library.server.jetty.JettyJsonHandler;
-import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils;
 import org.apache.skywalking.oap.server.receiver.trace.provider.TraceServiceModuleConfig;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserListenerManager;
 import org.apache.skywalking.oap.server.receiver.trace.provider.parser.TraceAnalyzer;
@@ -36,7 +36,7 @@ import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
 @Slf4j
-public class TraceSegmentReportServletHandler extends JettyJsonHandler {
+public abstract class TraceSegmentReportBaseServletHandler extends JettyJsonHandler {
 
     private final ModuleManager moduleManager;
     private final SegmentParserListenerManager listenerManager;
@@ -44,9 +44,9 @@ public class TraceSegmentReportServletHandler extends JettyJsonHandler {
     private HistogramMetrics histogram;
     private CounterMetrics errorCounter;
 
-    public TraceSegmentReportServletHandler(ModuleManager moduleManager,
-                                            SegmentParserListenerManager listenerManager,
-                                            TraceServiceModuleConfig config) {
+    public TraceSegmentReportBaseServletHandler(ModuleManager moduleManager,
+                                                SegmentParserListenerManager listenerManager,
+                                                TraceServiceModuleConfig config) {
         this.moduleManager = moduleManager;
         this.listenerManager = listenerManager;
         this.config = config;
@@ -62,11 +62,6 @@ public class TraceSegmentReportServletHandler extends JettyJsonHandler {
         );
     }
 
-    @Override
-    public String pathSpec() {
-        return "/v3/segments";
-    }
-
     @Override
     protected JsonElement doGet(HttpServletRequest req) {
         throw new UnsupportedOperationException();
@@ -79,19 +74,13 @@ public class TraceSegmentReportServletHandler extends JettyJsonHandler {
         }
         HistogramMetrics.Timer timer = histogram.createTimer();
 
-        StringBuilder stringBuilder = new StringBuilder();
-        String line;
         try {
-            BufferedReader reader = req.getReader();
-            while ((line = reader.readLine()) != null) {
-                stringBuilder.append(line);
-            }
+            final List<SegmentObject> segments = parseSegments(req);
 
-            SegmentObject.Builder upstreamSegmentBuilder = SegmentObject.newBuilder();
-            ProtoBufJsonUtils.fromJSON(stringBuilder.toString(), upstreamSegmentBuilder);
-
-            final TraceAnalyzer traceAnalyzer = new TraceAnalyzer(moduleManager, listenerManager, config);
-            traceAnalyzer.doAnalysis(upstreamSegmentBuilder.build());
+            for (SegmentObject segment : segments) {
+                final TraceAnalyzer traceAnalyzer = new TraceAnalyzer(moduleManager, listenerManager, config);
+                traceAnalyzer.doAnalysis(segment);
+            }
         } catch (Exception e) {
             errorCounter.inc();
             log.error(e.getMessage(), e);
@@ -101,4 +90,10 @@ public class TraceSegmentReportServletHandler extends JettyJsonHandler {
 
         return null;
     }
+
+    /**
+     * parsing segment list from request
+     */
+    protected abstract List<SegmentObject> parseSegments(HttpServletRequest request) throws IOException;
+
 }

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/rest/TraceSegmentReportListServletHandler.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils;
+import org.apache.skywalking.oap.server.receiver.trace.provider.TraceServiceModuleConfig;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserListenerManager;
+
+import javax.servlet.http.HttpServletRequest;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Parsing segments from list json
+ */
+public class TraceSegmentReportListServletHandler extends TraceSegmentReportBaseServletHandler {
+
+    private final Gson gson = new Gson();
+
+    public TraceSegmentReportListServletHandler(ModuleManager moduleManager, SegmentParserListenerManager listenerManager, TraceServiceModuleConfig config) {
+        super(moduleManager, listenerManager, config);
+    }
+
+    @Override
+    protected List<SegmentObject> parseSegments(HttpServletRequest request) throws IOException {
+        BufferedReader reader = request.getReader();
+        String line;
+        StringBuilder stringBuilder = new StringBuilder();
+        while ((line = reader.readLine()) != null) {
+            stringBuilder.append(line);
+        }
+
+        final JsonArray array = gson.fromJson(stringBuilder.toString(), JsonArray.class);
+        if (array.size() == 0) {
+            return Collections.emptyList();
+        }
+
+        final ArrayList<SegmentObject> segments = new ArrayList<>(array.size());
+        for (JsonElement element : array) {
+            SegmentObject.Builder upstreamSegmentBuilder = SegmentObject.newBuilder();
+            ProtoBufJsonUtils.fromJSON(element.toString(), upstreamSegmentBuilder);
+            segments.add(upstreamSegmentBuilder.build());
+        }
+        return segments;
+    }
+
+    @Override
+    public String pathSpec() {
+        return "/v3/segments";
+    }
+}

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/rest/TraceSegmentReportSingleServletHandler.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest;
+
+import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.ProtoBufJsonUtils;
+import org.apache.skywalking.oap.server.receiver.trace.provider.TraceServiceModuleConfig;
+import org.apache.skywalking.oap.server.receiver.trace.provider.parser.SegmentParserListenerManager;
+
+import javax.servlet.http.HttpServletRequest;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Parsing segments from single json
+ */
+public class TraceSegmentReportSingleServletHandler extends TraceSegmentReportBaseServletHandler {
+
+    public TraceSegmentReportSingleServletHandler(ModuleManager moduleManager, SegmentParserListenerManager listenerManager, TraceServiceModuleConfig config) {
+        super(moduleManager, listenerManager, config);
+    }
+
+    @Override
+    protected List<SegmentObject> parseSegments(HttpServletRequest request) throws IOException {
+        BufferedReader reader = request.getReader();
+        String line;
+        StringBuilder stringBuilder = new StringBuilder();
+        while ((line = reader.readLine()) != null) {
+            stringBuilder.append(line);
+        }
+
+        SegmentObject.Builder segBuilder = SegmentObject.newBuilder();
+        ProtoBufJsonUtils.fromJSON(stringBuilder.toString(), segBuilder);
+        return Arrays.asList(segBuilder.build());
+    }
+
+    @Override
+    public String pathSpec() {
+        return "/v3/segment";
+    }
+}

oap-server/server-receiver-plugin/skywalking-trace-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/trace/provider/handler/v8/rest/TraceSegmentReportServletHandlerTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.receiver.trace.provider.handler.v8.rest;
+
+import org.apache.skywalking.apm.network.language.agent.v3.SegmentObject;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.none.MetricsCreatorNoop;
+import org.apache.skywalking.oap.server.telemetry.none.NoneTelemetryProvider;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.internal.util.reflection.Whitebox;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import javax.servlet.http.HttpServletRequest;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.StringReader;
+import java.util.List;
+
+import static org.mockito.Mockito.when;
+
+@RunWith(PowerMockRunner.class)
+@PowerMockIgnore({"javax.management.*"})
+public class TraceSegmentReportServletHandlerTest {
+
+    @Mock
+    private HttpServletRequest request;
+    @Mock
+    private ModuleManager moduleManager;
+    @Mock
+    private NoneTelemetryProvider telemetryProvider;
+
+    @Before
+    public void init() throws IOException {
+        TelemetryModule telemetryModule = Mockito.spy(TelemetryModule.class);
+        Whitebox.setInternalState(telemetryModule, "loadedProvider", telemetryProvider);
+        Mockito.when(moduleManager.find(TelemetryModule.NAME)).thenReturn(telemetryModule);
+
+        Mockito.when(telemetryProvider.getService(MetricsCreator.class))
+            .thenReturn(new MetricsCreatorNoop());
+    }
+
+    @Test
+    public void testSingle() throws IOException {
+        String singleJson = "{" +
+            "   \"traceId\":\"c480c738-b628-490d-ace7-69f7030d77cb\"," +
+            "   \"spans\":[" +
+            "       {\"operationName\":\"\\/ingress\"}" +
+            "   ]" +
+            "}";
+
+        final TraceSegmentReportSingleServletHandler singleServletHandler =
+            new TraceSegmentReportSingleServletHandler(moduleManager, null, null);
+
+        when(request.getReader()).thenReturn(new BufferedReader(new StringReader(singleJson)));
+        final List<SegmentObject> segmentObjects = singleServletHandler.parseSegments(request);
+        Assert.assertEquals(segmentObjects.size(), 1);
+    }
+
+    @Test
+    public void testListJson() throws IOException {
+        String listJson = "[{" +
+            "   \"traceId\":\"c480c738-b628-490d-ace7-69f7030d77cb\"," +
+            "   \"spans\":[" +
+            "       {\"operationName\":\"\\/ingress\"}" +
+            "   ]" +
+            "},{" +
+            "   \"traceId\":\"e9673310-cf3a-467e-8f47-eaec26b57f76\"," +
+            "   \"spans\":[" +
+            "       {\"operationName\":\"\\/ingress\"}" +
+            "   ]" +
+            "}]";
+
+        final TraceSegmentReportListServletHandler singleServletHandler =
+            new TraceSegmentReportListServletHandler(moduleManager, null, null);
+
+        when(request.getReader()).thenReturn(new BufferedReader(new StringReader(listJson)));
+        final List<SegmentObject> segmentObjects = singleServletHandler.parseSegments(request);
+        Assert.assertEquals(segmentObjects.size(), 2);
+    }
+
+}

test/e2e/e2e-test/docker/lua/Dockerfile.nginx

@@ -15,7 +15,7 @@
 
 FROM openresty/openresty
 
-ENV COMMIT_HASH=7b23a978bb27787c69a57f734e4fa5865494ef79
+ENV COMMIT_HASH=4c0eee3991e022f93703569c8e7bf114de7bb310
 
 WORKDIR /usr/share/skywalking-nginx-lua