Optimize audit logic for batch operations especially when audit is not enabled for DML

Avoid audit batch event creation if audit for batch is not enabled (random UIID generation is not very cheap)
Avoid String.format in a potential hot path

Patch by Dmitry Konstantinov; reviewed by Štefan Miklošovič for CASSANDRA-20885
Co-authored-by: Štefan Miklošovič

Author: netudima

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Optimize audit logic for batch operations especially when audit is not enabled for DML (CASSANDRA-20885)
  * Implement nodetool history (CASSANDRA-20851)
  * Expose StorageService.dropPreparedStatements via JMX (CASSANDRA-20870)
  * Expose Metric for Prepared Statement Cache Size (in bytes) (CASSANDRA-20864)

src/java/org/apache/cassandra/audit/AuditLogFilter.java

@@ -153,6 +153,17 @@ boolean isFiltered(AuditLogEntry auditLogEntry)
                || isFiltered(auditLogEntry.getUser(), includedUsers, excludedUsers);
     }
 
+    boolean isFiltered(AuditLogContext auditLogContext)
+    {
+        return isFiltered(auditLogContext.keyspace, includedKeyspaces, excludedKeyspaces)
+               || isFiltered(auditLogContext.auditLogEntryType.getCategory().toString(), includedCategories, excludedCategories);
+    }
+
+    boolean isFiltered(AuditLogEntryType auditLogEntryType)
+    {
+        return isFiltered(auditLogEntryType.getCategory().toString(), includedCategories, excludedCategories);
+    }
+
     /**
      * Checks whether given input is being filtered or not.
      * If excludeSet does not contain any items, by default nothing is excluded (unless there are

src/java/org/apache/cassandra/audit/AuditLogManager.java

@@ -26,7 +26,6 @@
 import java.security.AccessControlContext;
 import java.security.AccessController;
 import java.security.Principal;
-import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
@@ -315,12 +314,47 @@ else if (statement != null)
             log(entry, cause, query == null ? null : ImmutableList.of(query));
     }
 
-    public void batchSuccess(BatchStatement.Type batchType, List<? extends CQLStatement> statements, List<String> queries, List<List<ByteBuffer>> values, QueryOptions options, QueryState state, long queryTime, Message.Response response)
+    public void batchSuccess(BatchStatement.Type batchType,
+                             List<? extends CQLStatement> statements,
+                             List<String> queries,
+                             List<List<ByteBuffer>> values,
+                             QueryOptions options,
+                             QueryState state,
+                             long queryStartTimeMillis,
+                             Message.Response response)
     {
-        List<AuditLogEntry> entries = buildEntriesForBatch(statements, queries, state, options, queryTime);
-        for (AuditLogEntry auditLogEntry : entries)
+        UUID batchId = null;
+        if (!filter.isFiltered(AuditLogEntryType.BATCH))
         {
-            log(auditLogEntry);
+            batchId = UUID.randomUUID(); // lazy init only if needed, to reduce overheads
+            log(new AuditLogEntry.Builder(state)
+                                   .setOperation("BatchId:[" + batchId + "] - BATCH of [" + statements.size() + "] statements")
+                                   .setOptions(options)
+                                   .setTimestamp(queryStartTimeMillis)
+                                   .setBatch(batchId)
+                                   .setType(AuditLogEntryType.BATCH)
+                                   .build());
+        }
+
+        for (int i = 0; i < statements.size(); i++)
+        {
+            CQLStatement statement = statements.get(i);
+            if (filter.isFiltered(statement.getAuditLogContext()))
+                continue;
+
+            if (batchId == null)
+                batchId = UUID.randomUUID();
+
+            log(new AuditLogEntry.Builder(state)
+                                  .setType(statement.getAuditLogContext().auditLogEntryType)
+                                  .setOperation(queries.get(i))
+                                  .setTimestamp(queryStartTimeMillis)
+                                  .setScope(statement)
+                                  .setKeyspace(state, statement)
+                                  .setOptions(options)
+                                  .setBatch(batchId)
+                                  .build());
+
         }
     }
 
@@ -334,38 +368,6 @@ public void batchFailure(BatchStatement.Type batchType, List<? extends CQLStatem
         log(entry, cause, queries);
     }
 
-    private static List<AuditLogEntry> buildEntriesForBatch(List<? extends CQLStatement> statements, List<String> queries, QueryState state, QueryOptions options, long queryStartTimeMillis)
-    {
-        List<AuditLogEntry> auditLogEntries = new ArrayList<>(statements.size() + 1);
-        UUID batchId = UUID.randomUUID();
-        String queryString = String.format("BatchId:[%s] - BATCH of [%d] statements", batchId, statements.size());
-        AuditLogEntry entry = new AuditLogEntry.Builder(state)
-                              .setOperation(queryString)
-                              .setOptions(options)
-                              .setTimestamp(queryStartTimeMillis)
-                              .setBatch(batchId)
-                              .setType(AuditLogEntryType.BATCH)
-                              .build();
-        auditLogEntries.add(entry);
-
-        for (int i = 0; i < statements.size(); i++)
-        {
-            CQLStatement statement = statements.get(i);
-            entry = new AuditLogEntry.Builder(state)
-                    .setType(statement.getAuditLogContext().auditLogEntryType)
-                    .setOperation(queries.get(i))
-                    .setTimestamp(queryStartTimeMillis)
-                    .setScope(statement)
-                    .setKeyspace(state, statement)
-                    .setOptions(options)
-                    .setBatch(batchId)
-                    .build();
-            auditLogEntries.add(entry);
-        }
-
-        return auditLogEntries;
-    }
-
     public void prepareSuccess(CQLStatement statement, String query, QueryState state, long queryTime, ResultMessage.Prepared response)
     {
         AuditLogEntry entry = new AuditLogEntry.Builder(state).setOperation(query)