diff --git a/README.md b/README.md
index 54ab4ca..8b3f642 100644
--- a/README.md
+++ b/README.md
@@ -27,6 +27,7 @@ CREATE TABLE starrocks_audit_db__.starrocks_audit_tbl__ (
`queryId` VARCHAR(64) COMMENT "Unique ID of the query",
`timestamp` DATETIME NOT NULL COMMENT "Query start time",
`queryType` VARCHAR(12) COMMENT "Query type (query, slow_query, connection)",
+ `eventType` VARCHAR(16) NULL COMMENT "Audit event type (BEFORE_QUERY, AFTER_QUERY, CONNECTION). This field is NULL when audit_stmt_before_execute is disabled",
`clientIp` VARCHAR(32) COMMENT "Client IP",
`user` VARCHAR(64) COMMENT "Query username",
`authorizedUser` VARCHAR(64) COMMENT "Unique user identifier, i.e., user_identity",
@@ -254,8 +255,23 @@ The `queryType` types supported in the StarRocks audit table include: query, slo
For connection, StarRocks 3.0.6+ supports printing successful/failed connection information when a client connects in `fe.audit.log`. You can configure `audit_log_modules=slow_query,query,connection` in `fe.conf` and then restart FE to enable it. After enabling connection information, the AuditLoader plugin can also collect this type of client connection information and load it into the table `starrocks_audit_tbl__`. After loading, the `queryType` field of the audit table for this type of information will be connection, which you can use to audit user login information.
+For StarRocks versions that support `audit_stmt_before_execute`, the AuditLoader plugin automatically detects whether this FE configuration is enabled. When it is disabled or unsupported by the current FE version, the `eventType` field is loaded as `NULL`, and query statements still produce only the normal after-execution audit record. When `audit_stmt_before_execute` is enabled, AuditLoader also collects before-execution audit events. You can use `eventType` to distinguish `BEFORE_QUERY`, `AFTER_QUERY`, and `CONNECTION` records.
+
+If you are upgrading from an earlier AuditLoader table, add the nullable field before installing the new plugin package:
+
+```sql
+ALTER TABLE starrocks_audit_db__.starrocks_audit_tbl__
+ADD COLUMN `eventType` VARCHAR(16) NULL COMMENT "Audit event type (BEFORE_QUERY, AFTER_QUERY, CONNECTION). This field is NULL when audit_stmt_before_execute is disabled";
+```
+
### Release Notes:
+##### AuditLoader v4.2.2
+
+1. Added the `eventType` field. It is populated only when the FE configuration `audit_stmt_before_execute` is enabled; otherwise, it is loaded as `NULL`.
+
+2. Added collection of `BEFORE_QUERY` audit events when `audit_stmt_before_execute` is enabled, allowing before- and after-execution audit records to be distinguished in the audit table.
+
##### AuditLoader v4.2.1
1. Added the ability to configure encrypted passwords in plugin.conf
diff --git a/pom.xml b/pom.xml
index 4345210..843dbf1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
com.starrocks
fe-plugins-auditloader
- 4.2.1
+ 4.2.2
2.24.1
@@ -95,4 +95,4 @@
-
\ No newline at end of file
+
diff --git a/src/main/java/com/starrocks/plugin/audit/AuditLoaderPlugin.java b/src/main/java/com/starrocks/plugin/audit/AuditLoaderPlugin.java
index 4c7b62d..c507c54 100644
--- a/src/main/java/com/starrocks/plugin/audit/AuditLoaderPlugin.java
+++ b/src/main/java/com/starrocks/plugin/audit/AuditLoaderPlugin.java
@@ -17,6 +17,7 @@
package com.starrocks.plugin.audit;
+import com.starrocks.common.Config;
import com.starrocks.plugin.*;
import com.starrocks.sql.ast.StatementBase;
import com.starrocks.sql.common.SqlDigestBuilder;
@@ -82,6 +83,11 @@ public class AuditLoaderPlugin extends Plugin implements AuditPlugin {
* 是否包含新字段 queriedRelations，如果旧版本没有该字段则值为空
*/
private boolean queriedRelationsExists;
+ /**
+ * Whether the FE supports the audit_stmt_before_execute configuration.
+ * Older StarRocks versions do not have this field, so resolve it by reflection.
+ */
+ private Field auditStmtBeforeExecuteField;
@Override
public void init(PluginInfo info, PluginContext ctx) throws PluginException {
@@ -102,6 +108,7 @@ public void init(PluginInfo info, PluginContext ctx) throws PluginException {
candidateMvsExists = hasField(AuditEvent.class, "candidateMvs");
hitMVsExists = hasField(AuditEvent.class, "hitMVs");
queriedRelationsExists = hasField(AuditEvent.class, "queriedRelations");
+ auditStmtBeforeExecuteField = getField(Config.class, "audit_stmt_before_execute");
isInit = true;
}
@@ -151,6 +158,7 @@ public void close() throws IOException {
public boolean eventFilter(AuditEvent.EventType type) {
return type == AuditEvent.EventType.AFTER_QUERY ||
+ (type == AuditEvent.EventType.BEFORE_QUERY && isAuditStmtBeforeExecuteEnabled()) ||
type == AuditEvent.EventType.CONNECTION;
}
@@ -179,6 +187,7 @@ private void assembleAudit(AuditEvent event) {
String content = "{\"queryId\":\"" + getQueryId(queryType, event) + "\"," +
"\"timestamp\":\"" + longToTimeString(event.timestamp) + "\"," +
"\"queryType\":\"" + queryType + "\"," +
+ "\"eventType\":" + getEventTypeJsonValue(event) + "," +
"\"clientIp\":\"" + event.clientIp + "\"," +
"\"user\":\"" + event.user + "\"," +
"\"authorizedUser\":\"" + event.authorizedUser + "\"," +
@@ -301,6 +310,13 @@ private String getQueriedRelationsJson(AuditEvent event) {
}
}
+ private String getEventTypeJsonValue(AuditEvent event) {
+ if (!isAuditStmtBeforeExecuteEnabled() || event.type == null) {
+ return "null";
+ }
+ return "\"" + event.type.name() + "\"";
+ }
+
/**
* 类中是否包含指定字段
*
@@ -318,6 +334,26 @@ private boolean hasField(Class clazz, String fieldName) {
return false;
}
+ private Field getField(Class clazz, String fieldName) {
+ try {
+ return clazz.getDeclaredField(fieldName);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ private boolean isAuditStmtBeforeExecuteEnabled() {
+ if (auditStmtBeforeExecuteField == null) {
+ return false;
+ }
+ try {
+ return auditStmtBeforeExecuteField.getBoolean(null);
+ } catch (Exception e) {
+ LOG.debug("encounter exception when getting audit_stmt_before_execute from FE config", e);
+ return false;
+ }
+ }
+
public static class AuditLoaderConf {
public static final String PROP_MAX_BATCH_SIZE = "max_batch_size";
public static final String PROP_MAX_BATCH_INTERVAL_SEC = "max_batch_interval_sec";
diff --git a/src/main/java/com/starrocks/plugin/audit/StarrocksStreamLoader.java b/src/main/java/com/starrocks/plugin/audit/StarrocksStreamLoader.java
index 6b4ba85..d2773b8 100644
--- a/src/main/java/com/starrocks/plugin/audit/StarrocksStreamLoader.java
+++ b/src/main/java/com/starrocks/plugin/audit/StarrocksStreamLoader.java
@@ -89,7 +89,7 @@ private HttpURLConnection getConnection(String urlStr, String label) throws IOEx
conn.addRequestProperty("label", label);
conn.addRequestProperty("max_filter_ratio", "1.0");
- conn.addRequestProperty("columns", "queryId,timestamp,queryType,clientIp,user,authorizedUser,resourceGroup,catalog,db,state,errorCode,queryTime,scanBytes,scanRows,returnRows,cpuCostNs,memCostBytes,stmtId,isQuery,feIp,stmt,digest,planCpuCosts,planMemCosts,pendingTimeMs,candidateMVs,hitMvs,QueriedRelations,warehouse");
+ conn.addRequestProperty("columns", "queryId,timestamp,queryType,eventType,clientIp,user,authorizedUser,resourceGroup,catalog,db,state,errorCode,queryTime,scanBytes,scanRows,returnRows,cpuCostNs,memCostBytes,stmtId,isQuery,feIp,stmt,digest,planCpuCosts,planMemCosts,pendingTimeMs,candidateMVs,hitMvs,QueriedRelations,warehouse");
if(!StringUtils.isBlank(this.streamLoadFilter)) {
conn.addRequestProperty("where", streamLoadFilter);
}
@@ -114,7 +114,7 @@ private String toCurl(HttpURLConnection conn) {
if(!StringUtils.isBlank(this.streamLoadFilter)) {
sb.append("-H \"").append("where\":").append(streamLoadFilter).append(" \\\n ");
}
- sb.append("-H \"").append("columns\":").append("\"queryId, timestamp, queryType, clientIp, user, authorizedUser, resourceGroup, catalog, db, state, errorCode," +
+ sb.append("-H \"").append("columns\":").append("\"queryId, timestamp, queryType, eventType, clientIp, user, authorizedUser, resourceGroup, catalog, db, state, errorCode," +
"queryTime, scanBytes, scanRows, returnRows, cpuCostNs, memCostBytes, stmtId, isQuery, feIp, stmt, digest, planCpuCosts, planMemCosts, pendingTimeMs, candidateMVs, hitMvs, QueriedRelations, warehouse\" \\\n ");
sb.append("\"").append(conn.getURL()).append("\"");
return sb.toString();