[SLS gateway]feat: support sls pb serialize for java

alibabacloud-gateway-sls/Teafile

@@ -1,7 +1,7 @@
 {
   "scope": "alibabacloud",
   "name": "GatewaySLS",
-  "version": "0.1.1",
+  "version": "0.1.2",
   "main": "./main.tea",
   "maintainers": [
     {
@@ -92,7 +92,8 @@
   },
   "releaselog": {
     "changelog": [
-      "[go/java]feat: support parsing compressed data."
+      "[go/java]feat: support parsing compressed data.",
+      "[java]feat: support pb serialization."
     ],
     "compatible": true
   }

alibabacloud-gateway-sls/java/pom.xml

@@ -59,7 +59,7 @@
     <dependency>
       <groupId>com.aliyun</groupId>
       <artifactId>tea-util</artifactId>
-      <version>0.2.16</version>
+      <version>0.2.20</version>
     </dependency>
     <dependency>
       <groupId>com.aliyun</groupId>

alibabacloud-gateway-sls/java/src/main/java/com/aliyun/gateway/sls/Client.java

@@ -28,6 +28,7 @@ public void modifyRequest(com.aliyun.gateway.spi.models.InterceptorContext conte
         String accessKeyId = credential.getAccessKeyId();
         String accessKeySecret = credential.getAccessKeySecret();
         String securityToken = credential.getSecurityToken();
+        String bodyRawSize = "0";
         if (!com.aliyun.teautil.Common.empty(accessKeyId)) {
             request.headers.put("x-log-signaturemethod", "hmac-sha1");
         }
@@ -38,8 +39,12 @@ public void modifyRequest(com.aliyun.gateway.spi.models.InterceptorContext conte
 
         if (!com.aliyun.teautil.Common.isUnset(request.body)) {
             if (com.aliyun.darabonbastring.Client.equals(request.reqBodyType, "protobuf")) {
-                // var bodyMap = Util.assertAsMap(request.body);
-                // 缺少body的Content-MD5计算，以及protobuf处理
+                byte[] serializedBody = com.aliyun.gateway.sls.util.Client.serializeLogGroupToPB(request.body);
+                bodyRawSize = com.aliyun.gateway.sls.util.Client.getBytesLength(serializedBody);
+                String compressType = request.headers.get("x-log-compresstype");
+                byte[] compressedData = com.aliyun.gateway.sls.util.Client.readAndCompressBlock(serializedBody, compressType);
+                request.headers.put("content-md5", com.aliyun.darabonbastring.Client.toUpper(com.aliyun.darabonba.encode.Encoder.hexEncode(com.aliyun.darabonba.signature.Signer.MD5SignForBytes(compressedData))));
+                request.stream = Tea.toReadable(compressedData);
                 request.headers.put("content-type", "application/x-protobuf");
             } else if (com.aliyun.darabonbastring.Client.equals(request.reqBodyType, "json")) {
                 String bodyStr = com.aliyun.teautil.Common.toJSONString(request.body);
@@ -63,7 +68,7 @@ public void modifyRequest(com.aliyun.gateway.spi.models.InterceptorContext conte
                 new TeaPair("date", com.aliyun.teautil.Common.getDateUTCString()),
                 new TeaPair("user-agent", request.userAgent),
                 new TeaPair("x-log-apiversion", "0.6.0"),
-                new TeaPair("x-log-bodyrawsize", "0")
+                new TeaPair("x-log-bodyrawsize", bodyRawSize)
             ),
             request.headers
         );

alibabacloud-gateway-sls/main.tea

@@ -32,6 +32,7 @@ async function modifyRequest(context: SPI.InterceptorContext, attributeMap: SPI.
   var accessKeyId = credential.getAccessKeyId();
   var accessKeySecret = credential.getAccessKeySecret();
   var securityToken = credential.getSecurityToken();
+  var bodyRawSize = '0';
   if (!Util.empty(accessKeyId)) {
       request.headers['x-log-signaturemethod'] = 'hmac-sha1';
   }
@@ -40,8 +41,12 @@ async function modifyRequest(context: SPI.InterceptorContext, attributeMap: SPI.
   }
   if (!Util.isUnset(request.body)) {
     if (String.equals(request.reqBodyType, 'protobuf')) {
-      // var bodyMap = Util.assertAsMap(request.body);
-      // 缺少body的Content-MD5计算，以及protobuf处理
+      var serializedBody = SLS_Util.serializeLogGroupToPB(request.body);
+      bodyRawSize = SLS_Util.getBytesLength(serializedBody);
+      var compressType = request.headers['x-log-compresstype'];
+      var compressedData = SLS_Util.readAndCompressBlock(serializedBody, compressType);
+      request.headers['content-md5'] = String.toUpper(EncodeUtil.hexEncode(SignatureUtil.MD5SignForBytes(compressedData)));
+      request.stream = compressedData;
       request.headers['content-type'] = 'application/x-protobuf';
     } else if (String.equals(request.reqBodyType, 'json')) {
       var bodyStr = Util.toJSONString(request.body);
@@ -62,7 +67,7 @@ async function modifyRequest(context: SPI.InterceptorContext, attributeMap: SPI.
     date = Util.getDateUTCString(),
     user-agent = request.userAgent,
     x-log-apiversion = '0.6.0',
-    x-log-bodyrawsize = '0',
+    x-log-bodyrawsize = bodyRawSize,
     ...request.headers
   };
   request.headers['authorization'] = getAuthorization(request.pathname, request.method, request.query, request.headers, accessKeyId, accessKeySecret);

alibabacloud-gateway-sls/util/Teafile

@@ -1,7 +1,7 @@
 {
   "scope": "alibabacloud",
   "name": "GatewaySLS_Util",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "main": "./main.tea",
   "maintainers": [
     {

alibabacloud-gateway-sls/util/java/pom.xml

@@ -61,6 +61,11 @@
       <artifactId>tea-util</artifactId>
       <version>0.2.14</version>
     </dependency>
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-java</artifactId>
+      <version>2.5.0</version>
+    </dependency>
   </dependencies>
   <build>
     <plugins>

alibabacloud-gateway-sls/util/java/src/main/java/com/aliyun/gateway/sls/util/Client.java

@@ -1,8 +1,15 @@
 // This file is auto-generated, don't edit it. Thanks.
 package com.aliyun.gateway.sls.util;
 
+import com.aliyun.tea.TeaConverter;
+import com.aliyun.tea.TeaException;
+import com.aliyun.tea.TeaPair;
+
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.HashMap;
 
 public class Client {
 
@@ -13,4 +20,76 @@ public static InputStream readAndUncompressBlock(InputStream stream, String comp
         String data = new String(rawData, "UTF-8");
         return new ByteArrayInputStream(data.getBytes());
     }
-}
+
+    public static byte[] readAndCompressBlock(byte[] stream, String compressType) throws Exception {
+        byte[] compressedData = CompressorFactory.getCompressor(compressType).compress(stream);
+        return compressedData;
+    }
+
+    public static byte[] serializeLogGroupToPB(Object logGroup) throws Exception {
+        byte[] logBytes = null;
+        Logs.LogGroup.Builder logs = Logs.LogGroup.newBuilder();
+        HashMap<String, Object> body;
+        if (logGroup instanceof HashMap) {
+            body = (HashMap<String, Object>) logGroup;
+        } else {
+            throw new IllegalArgumentException("Invalid body type " + logGroup.getClass());
+        }
+
+        String topic = (String) body.get("Topic");
+        if (topic != null) {
+            logs.setTopic((String) body.get("Topic"));
+        } else {
+            throw new IllegalArgumentException("Topic is null");
+        }
+        String source = (String) body.get("Source");
+        if (source != null) {
+            logs.setSource((String) body.get("Source"));
+        } else {
+            try {
+                logs.setSource(InetAddress.getLocalHost().getHostAddress());
+            } catch (Exception e) {}
+        }
+
+        ArrayList<Object> logTags = (ArrayList<Object>) body.get("LogTags");
+        if (logTags != null && logTags.size() > 0) {
+            for (Object obj : logTags) {
+                HashMap<String, String> tag = (HashMap<String, String>) obj;
+                Logs.LogTag.Builder tagBuilder = logs.addLogTagsBuilder();
+                tagBuilder.setKey(tag.get("Key"));
+                tagBuilder.setValue(tag.get("Value"));
+            }
+        }
+        ArrayList<Object> logItems = (ArrayList<Object>) body.get("Logs");
+        for (Object obj : logItems) {
+            Logs.Log.Builder logsBuilder = logs.addLogsBuilder();
+            HashMap<String, Object> logItem = (HashMap<String, Object>) obj;
+            logsBuilder.setTime((Integer) logItem.get("Time"));
+            ArrayList<Object> contents = (ArrayList<Object>) logItem.get("Contents");
+            for (Object content : contents) {
+                HashMap<String, String> realContent = (HashMap<String, String>) content;
+                Logs.Log.Content.Builder contentBuilder = logsBuilder.addContentsBuilder();
+                contentBuilder.setKey(realContent.get("Key"));
+                contentBuilder.setValue(realContent.get("Value"));
+            }
+        }
+        logBytes = logs.build().toByteArray();
+
+        int bodySize = logBytes.length;
+        if (bodySize > 50 * 1024 * 1024) {
+            throw new TeaException(TeaConverter.buildMap(
+                    new TeaPair("code", "InvalidLogSize"),
+                    new TeaPair("message", "logItems' size exceeds maximum limitation : " + 50 * 1024 * 1024 + " bytes")));
+        } else if (bodySize > 10 * 1024 * 1024) {
+            throw new TeaException(TeaConverter.buildMap(
+                    new TeaPair("code", "PostBodyTooLarge"),
+                    new TeaPair("message", "body size " + bodySize + " must little than " + 10 * 1024 * 1024)));
+        }
+
+        return logBytes;
+    }
+
+    public static String getBytesLength(byte[] stream) throws Exception {
+        return String.valueOf(stream.length);
+    }
+}

alibabacloud-gateway-sls/util/java/src/main/java/com/aliyun/gateway/sls/util/CompressorFactory.java

@@ -0,0 +1,74 @@
+package com.aliyun.gateway.sls.util;
+
+import com.aliyun.tea.TeaConverter;
+import com.aliyun.tea.TeaException;
+import com.aliyun.tea.TeaPair;
+import net.jpountz.lz4.LZ4Compressor;
+import net.jpountz.lz4.LZ4Exception;
+import net.jpountz.lz4.LZ4Factory;
+
+import java.io.ByteArrayOutputStream;
+import java.util.zip.Deflater;
+
+interface Compressor {
+    byte[] compress(byte[] data) throws Exception;
+}
+
+class Lz4Compressor implements Compressor {
+    @Override
+    public byte[] compress(byte[] data) {
+        final int rawSize = data.length;
+        LZ4Factory factory = LZ4Factory.fastestInstance();
+
+        // compress data
+        LZ4Compressor compressor = factory.fastCompressor();
+
+        int maxCompressedLength = compressor.maxCompressedLength(rawSize);
+        int encodingSize = 0;
+        byte[] rawCompressed = new byte[maxCompressedLength];
+        try {
+            encodingSize = compressor.compress(data, 0, rawSize, rawCompressed, 0, maxCompressedLength);
+        } catch (LZ4Exception e) {
+            throw new TeaException(TeaConverter.buildMap(new TeaPair("CompressException", e.getMessage())));
+        }
+
+        if (encodingSize <= 0) {
+            throw new TeaException(TeaConverter.buildMap(new TeaPair("CompressException", "Invalid encoding size")));
+        }
+
+        byte[] ret = new byte[encodingSize];
+        System.arraycopy(rawCompressed, 0, ret, 0, encodingSize);
+
+        return ret;
+    }
+}
+
+class GzipCompressor implements Compressor {
+    @Override
+    public byte[] compress(byte[] data) {
+        ByteArrayOutputStream out = new ByteArrayOutputStream(data.length);
+        Deflater compressor = new Deflater();
+        try {
+            compressor.setInput(data);
+            compressor.finish();
+            byte[] buf = new byte[10240];
+            while (!compressor.finished()) {
+                int count = compressor.deflate(buf);
+                out.write(buf, 0, count);
+            }
+            return out.toByteArray();
+        } finally {
+            compressor.end();
+        }
+    }
+}
+
+public class CompressorFactory {
+    public static Compressor getCompressor(String compressType) {
+        if ("lz4".equals(compressType)) {
+            return new Lz4Compressor();
+        }
+
+        throw new IllegalArgumentException("Invalid compressType: " + compressType);
+    }
+}