This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 41f47bd2e2d branch-4.2: [opt](audit) Compress audit log stream load
payloads (#68083) (#69105)
41f47bd2e2d is described below
commit 41f47bd2e2d7969d2c0d1fba7a0de497a4756d0a
Author: morrySnow <[email protected]>
AuthorDate: Thu Oct 15 16:54:47 2026 +0800
branch-4.2: [opt](audit) Compress audit log stream load payloads (#68083)
(#69105)
### What problem does this PR solve?
Issue Number: N/A
Related PR: https://github.com/apache/doris/pull/68083
Problem Summary: Built-in audit loading sent uncompressed Stream Load
payloads. Compress the body using gzip and send compress_type: gz while
preserving the branch-4.2 audit-loader structure.
### Release note
Enable gzip compression for built-in audit log Stream Load requests.
### Backport validation
- Test: Unit Test: AuditStreamLoaderTest; FE Checkstyle (0 violations).
- Combined eight-backport validation: 283 tests across 13 classes, 0
failures/errors after correcting identity-cast assertions. This
standalone backport has not had a separate unit-test run.
- An isolated combined eight-backport ASAN cluster is deployed, but no
targeted regression suite for this PR or full regression suite was run.
- Full build.sh --fe and ASAN build.sh --be succeeded on the combined
eight-backport tree; deployment smoke validation passed.
- Behavior changed: Yes (audit Stream Load payloads use gzip)
- Does this need documentation: No
(cherry picked from commit ca30be14a5fa32ef837ead479a05f5c5e0b98489)
### Scope
This is an independent backport of #68083 onto branch-4.2.
The branch contains one commit and does not include the other seven
picks.
The 283-test result above was obtained from the combined backport tree;
this standalone branch has not had a separate unit-test run.
### Check List (For Author)
- Test:
- [x] Unit Test: 283 targeted tests passed on the combined backport
tree.
- [x] Manual test: standalone patch matches the validated pick; FE
Checkstyle passed on the combined tree.
- [ ] Regression test: no targeted regression suite for this PR was run;
the full regression suite was not run.
- Behavior changed: see the backport validation details above.
- Does this need documentation: No.
---
.../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 fb9ffde02dc..f0a5e1979e2 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
@@ -30,16 +30,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;
public class AuditStreamLoader {
private static final Logger LOG =
LogManager.getLogger(AuditStreamLoader.class);
private static String loadUrlPattern = "http://%s/api/%s/%s/_stream_load?";;
// 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 hostPort;
private String db;
private String auditLogTbl;
@@ -55,7 +59,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();
conn.setInstanceFollowRedirects(false);
@@ -65,6 +70,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));
@@ -88,6 +94,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 ");
@@ -116,6 +123,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();
@@ -138,9 +151,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]
