This is an automated email from the ASF dual-hosted git repository.
morrySnow pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new ca30be14a5f [opt](audit) Compress audit log stream load payloads
(#68083)
ca30be14a5f is described below
commit ca30be14a5fa32ef837ead479a05f5c5e0b98489
Author: morrySnow <[email protected]>
AuthorDate: Thu Sep 17 11:05:57 2026 +0800
[opt](audit) Compress audit log stream load payloads (#68083)
### What problem does this PR solve?
Problem Summary:
The built-in audit loader currently sends every Stream Load batch as
uncompressed text. This PR gzip-compresses the request body and adds the
`compress_type: gz` Stream Load header, reducing network traffic for
audit batches. The optional external audit loader plugin is unchanged.
### Release note
Enable gzip compression for built-in audit log Stream Load requests.
---
.../doris/plugin/audit/AuditStreamLoader.java | 19 ++++--
.../doris/plugin/audit/AuditStreamLoaderTest.java | 67 ++++++++++++++++++++++
2 files changed, 82 insertions(+), 4 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java
b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java
index 73a78e7356e..ea8560a7653 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditStreamLoader.java
@@ -33,16 +33,20 @@ import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
+import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
+import java.nio.charset.StandardCharsets;
import java.util.Calendar;
import java.util.stream.Collectors;
+import java.util.zip.GZIPOutputStream;
import javax.net.ssl.HttpsURLConnection;
public class AuditStreamLoader {
private static final Logger LOG =
LogManager.getLogger(AuditStreamLoader.class);
// timeout for both connection and read. 10 seconds is long enough.
private static final int HTTP_TIMEOUT_MS = 10000;
+ private static final String COMPRESS_TYPE = "gz";
private String db;
private String auditLogTbl;
private String auditLogLoadUrlStr;
@@ -58,7 +62,8 @@ public class AuditStreamLoader {
this.feIdentity =
Env.getCurrentEnv().getSelfNode().getIdent().replaceAll("\\.",
"_").replaceAll(":", "_");
}
- private HttpURLConnection getConnection(String urlStr, String label,
String clusterToken) throws IOException {
+ private static HttpURLConnection getConnection(
+ String urlStr, String label, String clusterToken) throws
IOException {
URL url = new URL(urlStr);
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
if (conn instanceof HttpsURLConnection && Config.enable_https) {
@@ -73,6 +78,7 @@ public class AuditStreamLoader {
conn.addRequestProperty("Expect", "100-continue");
conn.addRequestProperty("Content-Type", "text/plain; charset=UTF-8");
conn.addRequestProperty("label", label);
+ conn.addRequestProperty("compress_type", COMPRESS_TYPE);
conn.setConnectTimeout(HTTP_TIMEOUT_MS);
conn.setReadTimeout(HTTP_TIMEOUT_MS);
conn.setRequestProperty("timeout",
String.valueOf(GlobalVariable.auditPluginLoadTimeoutS));
@@ -96,6 +102,7 @@ public class AuditStreamLoader {
sb.append("-H \"").append("Expect\":").append("\"100-continue\" \\\n
");
sb.append("-H \"").append("Content-Type\":").append("\"text/plain;
charset=UTF-8\" \\\n ");
sb.append("-H \"").append("max_filter_ratio\":").append("\"1.0\" \\\n
");
+ sb.append("-H
\"").append("compress_type\":").append("\"").append(COMPRESS_TYPE).append("\"
\\\n ");
sb.append("-H \"").append("columns\":")
.append("\"" + InternalSchema.AUDIT_SCHEMA.stream().map(c ->
c.getName()).collect(
Collectors.joining(",")) + "\" \\\n ");
@@ -124,6 +131,12 @@ public class AuditStreamLoader {
return response.toString();
}
+ private static void writeCompressedBody(OutputStream outputStream,
StringBuilder payload) throws IOException {
+ try (GZIPOutputStream gzipOutputStream = new GZIPOutputStream(new
BufferedOutputStream(outputStream))) {
+
gzipOutputStream.write(payload.toString().getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
public LoadResponse loadBatch(StringBuilder sb, String clusterToken) {
String label = genLabel();
@@ -146,9 +159,7 @@ public class AuditStreamLoader {
// build request and send to new be location
beConn = getConnection(location, label, clusterToken);
// send data to be
- try (BufferedOutputStream bos = new
BufferedOutputStream(beConn.getOutputStream())) {
- bos.write(sb.toString().getBytes());
- }
+ writeCompressedBody(beConn.getOutputStream(), sb);
// get respond
status = beConn.getResponseCode();
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditStreamLoaderTest.java
b/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditStreamLoaderTest.java
new file mode 100644
index 00000000000..1776906aee7
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/plugin/audit/AuditStreamLoaderTest.java
@@ -0,0 +1,67 @@
+// 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.doris.plugin.audit;
+
+import org.apache.doris.common.jmockit.Deencapsulation;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.nio.charset.StandardCharsets;
+import java.util.zip.GZIPInputStream;
+
+public class AuditStreamLoaderTest {
+
+ @Test
+ public void testConnectionUsesGzipCompression() throws Exception {
+ HttpURLConnection connection =
Deencapsulation.invoke(AuditStreamLoader.class, "getConnection",
+ "http://127.0.0.1:8030/api/db/table/_stream_load?";, "label",
"token");
+
+ Assertions.assertEquals("gz",
connection.getRequestProperty("compress_type"));
+ }
+
+ @Test
+ public void testWriteCompressedBody() throws Exception {
+ String payload = "audit row 中文\u001fselect 1\u001e";
+ ByteArrayOutputStream compressed = new ByteArrayOutputStream();
+
+ Deencapsulation.invoke(AuditStreamLoader.class, "writeCompressedBody",
+ compressed, new StringBuilder(payload));
+
+ byte[] bytes = compressed.toByteArray();
+ Assertions.assertEquals(0x1f, bytes[0] & 0xff);
+ Assertions.assertEquals(0x8b, bytes[1] & 0xff);
+ Assertions.assertEquals(payload, decompress(bytes));
+ }
+
+ private static String decompress(byte[] compressed) throws IOException {
+ try (GZIPInputStream gzipInputStream = new GZIPInputStream(new
ByteArrayInputStream(compressed));
+ ByteArrayOutputStream output = new ByteArrayOutputStream()) {
+ byte[] buffer = new byte[1024];
+ int bytesRead;
+ while ((bytesRead = gzipInputStream.read(buffer)) != -1) {
+ output.write(buffer, 0, bytesRead);
+ }
+ return new String(output.toByteArray(), StandardCharsets.UTF_8);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
