[fix](auditlog) add missing audit log fields

Author: morningman

fe/fe-core/src/main/java/org/apache/doris/catalog/InternalSchema.java

@@ -139,9 +139,23 @@ public class InternalSchema {
         AUDIT_SCHEMA.add(new ColumnDef("scan_rows", TypeDef.create(PrimitiveType.BIGINT), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA
                 .add(new ColumnDef("return_rows", TypeDef.create(PrimitiveType.BIGINT), ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA
+                .add(new ColumnDef("shuffle_send_rows", TypeDef.create(PrimitiveType.BIGINT),
+                        ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA
+                .add(new ColumnDef("shuffle_send_bytes", TypeDef.create(PrimitiveType.BIGINT),
+                        ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA
+                .add(new ColumnDef("scan_bytes_from_local_storage", TypeDef.create(PrimitiveType.BIGINT),
+                        ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA
+                .add(new ColumnDef("scan_bytes_from_remote_storage", TypeDef.create(PrimitiveType.BIGINT),
+                        ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(new ColumnDef("stmt_id", TypeDef.create(PrimitiveType.BIGINT), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(new ColumnDef("stmt_type", TypeDef.createVarchar(48), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(new ColumnDef("is_query", TypeDef.create(PrimitiveType.TINYINT), ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA.add(
+                new ColumnDef("is_nereids", TypeDef.create(PrimitiveType.TINYINT), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(new ColumnDef("frontend_ip", TypeDef.createVarchar(128), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA
                 .add(new ColumnDef("cpu_time_ms", TypeDef.create(PrimitiveType.BIGINT), ColumnNullableType.NULLABLE));
@@ -151,6 +165,9 @@ public class InternalSchema {
                 new ColumnDef("peak_memory_bytes", TypeDef.create(PrimitiveType.BIGINT), ColumnNullableType.NULLABLE));
         AUDIT_SCHEMA.add(
                 new ColumnDef("workload_group", TypeDef.create(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
+        AUDIT_SCHEMA.add(
+                new ColumnDef("compute_group", TypeDef.create(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
+        // Keep stmt as last column. So that in fe.audit.log, it will be easier to get sql string
         AUDIT_SCHEMA.add(new ColumnDef("stmt", TypeDef.create(PrimitiveType.STRING), ColumnNullableType.NULLABLE));
     }
 

fe/fe-core/src/main/java/org/apache/doris/plugin/AuditEvent.java

@@ -80,9 +80,9 @@ public enum EventType {
     public String queryId = "";
     @AuditField(value = "IsQuery")
     public boolean isQuery = false;
-    @AuditField(value = "isNereids")
+    @AuditField(value = "IsNereids")
     public boolean isNereids = false;
-    @AuditField(value = "feIp")
+    @AuditField(value = "FeIp")
     public String feIp = "";
     @AuditField(value = "StmtType")
     public String stmtType = "";
@@ -96,22 +96,20 @@ public enum EventType {
     public long shuffleSendRows = -1;
     @AuditField(value = "SqlHash")
     public String sqlHash = "";
-    @AuditField(value = "peakMemoryBytes")
+    @AuditField(value = "PeakMemoryBytes")
     public long peakMemoryBytes = -1;
     @AuditField(value = "SqlDigest")
     public String sqlDigest = "";
-    @AuditField(value = "cloudClusterName")
+    @AuditField(value = "ComputeGroupName")
     public String cloudClusterName = "";
-    @AuditField(value = "TraceId")
-    public String traceId = "";
     @AuditField(value = "WorkloadGroup")
     public String workloadGroup = "";
     // note: newly added fields should be always before fuzzyVariables
     @AuditField(value = "FuzzyVariables")
     public String fuzzyVariables = "";
-    @AuditField(value = "scanBytesFromLocalStorage")
+    @AuditField(value = "ScanBytesFromLocalStorage")
     public long scanBytesFromLocalStorage = -1;
-    @AuditField(value = "scanBytesFromRemoteStorage")
+    @AuditField(value = "ScanBytesFromRemoteStorage")
     public long scanBytesFromRemoteStorage = -1;
 
     public long pushToAuditLogQueueTime;

fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLoader.java

@@ -35,8 +35,6 @@
 import org.apache.logging.log4j.Logger;
 
 import java.io.IOException;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 
@@ -48,9 +46,6 @@ public class AuditLoader extends Plugin implements AuditPlugin {
 
     public static final String AUDIT_LOG_TABLE = "audit_log";
 
-    private static final DateTimeFormatter DATETIME_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS")
-            .withZone(ZoneId.systemDefault());
-
     private StringBuilder auditLogBuffer = new StringBuilder();
     private int auditLogNum = 0;
     private long lastLoadTimeAuditLog = 0;
@@ -143,6 +138,7 @@ private void assembleAudit(AuditEvent event) {
     }
 
     private void fillLogBuffer(AuditEvent event, StringBuilder logBuffer) {
+        // should be same order as InternalSchema.AUDIT_SCHEMA
         logBuffer.append(event.queryId).append("\t");
         logBuffer.append(TimeUtils.longToTimeStringWithms(event.timestamp)).append("\t");
         logBuffer.append(event.clientIp).append("\t");
@@ -156,6 +152,10 @@ private void fillLogBuffer(AuditEvent event, StringBuilder logBuffer) {
         logBuffer.append(event.scanBytes).append("\t");
         logBuffer.append(event.scanRows).append("\t");
         logBuffer.append(event.returnRows).append("\t");
+        logBuffer.append(event.shuffleSendRows).append("\t");
+        logBuffer.append(event.shuffleSendBytes).append("\t");
+        logBuffer.append(event.scanBytesFromLocalStorage).append("\t");
+        logBuffer.append(event.scanBytesFromRemoteStorage).append("\t");
         logBuffer.append(event.stmtId).append("\t");
         logBuffer.append(event.stmtType).append("\t");
         logBuffer.append(event.isQuery ? 1 : 0).append("\t");
@@ -165,6 +165,7 @@ private void fillLogBuffer(AuditEvent event, StringBuilder logBuffer) {
         logBuffer.append(event.sqlDigest).append("\t");
         logBuffer.append(event.peakMemoryBytes).append("\t");
         logBuffer.append(event.workloadGroup).append("\t");
+        logBuffer.append(event.cloudClusterName).append("\t");
         // already trim the query in org.apache.doris.qe.AuditLogHelper#logAuditLog
         String stmt = event.stmt;
         if (LOG.isDebugEnabled()) {

fe/fe-core/src/main/java/org/apache/doris/plugin/audit/AuditLogBuilder.java

@@ -116,7 +116,7 @@ private void auditQueryLog(AuditEvent event) throws IllegalAccessException {
             if (af.value().equals("Time(ms)")) {
                 queryTime = (long) f.get(event);
             }
-            sb.append("|").append(af.value()).append("=").append(String.valueOf(f.get(event)));
+            sb.append("|").append(af.value()).append("=").append(f.get(event));
         }
 
         String auditLog = sb.toString();