support Lz4 compression for http-based collector (#796)

* Support lz4 compressed input * Remove extra change * use batch size as initial array size
FrankChen021 · May 28, 2024 · 3ef6fb3 · 3ef6fb3
1 parent e2c4dd1
commit 3ef6fb3
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 11 deletions.
diff --git a/server/collector/pom.xml b/server/collector/pom.xml
@@ -41,6 +41,16 @@
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java-util</artifactId>
     </dependency>
+
+    <!--Use LZ4 decompress
+    This lib supports dependent block while the org.lz4:lz4-java does not
+    -->
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-compress</artifactId>
+      <version>1.26.2</version>
+    </dependency>
+
   </dependencies>
 
   <build>

diff --git a/server/collector/src/main/java/org/bithon/server/collector/http/TraceHttpCollector.java b/server/collector/src/main/java/org/bithon/server/collector/http/TraceHttpCollector.java
@@ -23,6 +23,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Setter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.compress.compressors.lz4.FramedLZ4CompressorInputStream;
 import org.bithon.component.commons.concurrency.NamedThreadFactory;
 import org.bithon.component.commons.utils.ReflectionUtils;
 import org.bithon.component.commons.utils.StringUtils;
@@ -85,19 +86,29 @@ public void span(HttpServletRequest request, HttpServletResponse response) throw
 
         String encoding = request.getHeader("Content-Encoding");
         if (!StringUtils.isEmpty(encoding)) {
-            if ("gzip".equals(encoding)) {
-                is = new GZIPInputStream(is);
-            } else if ("deflate".equals(encoding)) {
-                is = new InflaterInputStream(is);
-            } else {
-                String message = StringUtils.format("Not supported Content-Encoding [%s] from remote [%s]", encoding, request.getRemoteAddr());
-                response.getWriter().println(message);
-                response.setStatus(HttpStatus.BAD_REQUEST.value());
-                return;
+            switch (encoding) {
+                case "gzip":
+                    is = new GZIPInputStream(is);
+                    break;
+
+                case "deflate":
+                    is = new InflaterInputStream(is);
+                    break;
+
+                case "lz4":
+                    // Currently only FramedLZ4 is supported
+                    is = new FramedLZ4CompressorInputStream(is);
+                    break;
+
+                default:
+                    String message = StringUtils.format("Not supported Content-Encoding [%s] from remote [%s]", encoding, request.getRemoteAddr());
+                    response.getWriter().println(message);
+                    response.setStatus(HttpStatus.BAD_REQUEST.value());
+                    return;
             }
         }
 
-        List<TraceSpan> spans = new ArrayList<>(256);
+        List<TraceSpan> spans = new ArrayList<>(config.getMaxRowsPerBatch());
 
         // Create parser manually so that this parser can be accessed in the catch handler
         try (JsonParser parser = om.createParser(is)) {
@@ -117,7 +128,7 @@ public void span(HttpServletRequest request, HttpServletResponse response) throw
                         process(spans);
 
                         // Create a new array to hold the next batch
-                        spans = new ArrayList<>(256);
+                        spans = new ArrayList<>(config.getMaxRowsPerBatch());
                     }
                 }
             } catch (IOException e) {