HDDS-11455. optimized flow for obs

apache · Oct 11, 2024 · fae0f2e · fae0f2e
1 parent 3c91ad4
commit fae0f2e
Show file tree

Hide file tree

Showing 16 changed files with 2,640 additions and 154 deletions.
diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmKeyInfo.java
@@ -69,7 +69,7 @@ private static Codec<OmKeyInfo> newCodec(boolean ignorePipeline) {
   }
 
   public static Codec<OmKeyInfo> getCodec(boolean ignorePipeline) {
-    LOG.info("OmKeyInfo.getCodec ignorePipeline = {}", ignorePipeline);
+    // LOG.info("OmKeyInfo.getCodec ignorePipeline = {}", ignorePipeline);
     return ignorePipeline ? CODEC_TRUE : CODEC_FALSE;
   }
 

diff --git a/...p-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/helpers/OMAuditLogger.java b/...p-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/helpers/OMAuditLogger.java
@@ -26,6 +26,7 @@
 import org.apache.hadoop.ozone.audit.AuditMessage;
 import org.apache.hadoop.ozone.audit.OMAction;
 import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.ratis.execution.request.OmRequestBase;
 import org.apache.hadoop.ozone.om.request.OMClientRequest;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.ratis.server.protocol.TermIndex;
@@ -159,6 +160,34 @@ public static void log(OMAuditLogger.Builder builder, OMClientRequest request, O
     }
   }
 
+  public static void log(OMAuditLogger.Builder builder, OmRequestBase request, OzoneManager om,
+                         TermIndex termIndex, Throwable th) {
+    if (builder.isLog.get()) {
+      builder.getAuditLogger().logWrite(builder.getMessageBuilder().build());
+      return;
+    }
+
+    OMAction action = getAction(request.getOmRequest());
+    if (null == action) {
+      // no audit log defined
+      return;
+    }
+    if (builder.getAuditMap() == null) {
+      builder.setAuditMap(new HashMap<>());
+    }
+    try {
+      builder.getAuditMap().put("Command", request.getOmRequest().getCmdType().name());
+      builder.getAuditMap().put("Transaction", "" + termIndex.getIndex());
+      request.buildAuditMessage(action, builder.getAuditMap(),
+          th, request.getOmRequest().getUserInfo());
+      builder.setLog(true);
+      builder.setAuditLogger(om.getAuditLogger());
+      log(builder);
+    } catch (Exception ex) {
+      LOG.error("Exception occurred while write audit log, ", ex);
+    }
+  }
+
   public static Builder newBuilder() {
     return new Builder();
   }

diff --git a/...manager/src/main/java/org/apache/hadoop/ozone/om/ratis/execution/BucketQuotaResource.java b/...manager/src/main/java/org/apache/hadoop/ozone/om/ratis/execution/BucketQuotaResource.java
@@ -59,16 +59,16 @@ public static class BucketQuota {
     private AtomicLong incUsedBytes = new AtomicLong();
     private AtomicLong incUsedNamespace = new AtomicLong();
 
-    public void addUsedBytes(long bytes) {
-      incUsedBytes.addAndGet(bytes);
+    public long addUsedBytes(long bytes) {
+      return incUsedBytes.addAndGet(bytes);
     }
 
     public long getUsedBytes() {
       return incUsedBytes.get();
     }
 
-    public void addUsedNamespace(long bytes) {
-      incUsedNamespace.addAndGet(bytes);
+    public long addUsedNamespace(long bytes) {
+      return incUsedNamespace.addAndGet(bytes);
     }
 
     public long getUsedNamespace() {

diff --git a/...e-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/execution/DbChangesRecorder.java b/...e-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/execution/DbChangesRecorder.java
@@ -0,0 +1,47 @@
+package org.apache.hadoop.ozone.om.ratis.execution;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.utils.db.CodecBuffer;
+
+/**
+ * Records db changes.
+ */
+public class DbChangesRecorder {
+  private Map<String, Map<String, CodecBuffer>> tableRecordsMap = new HashMap<>();
+  private Map<String, Pair<Long, Long>> bucketUsedQuotaMap = new HashMap<>();
+
+  public void add(String name, String dbOpenKeyName, CodecBuffer omKeyCodecBuffer) {
+    Map<String, CodecBuffer> recordMap = tableRecordsMap.computeIfAbsent(name, k -> new HashMap<>());
+    recordMap.put(dbOpenKeyName, omKeyCodecBuffer);
+  }
+  public void add(String bucketName, long incUsedBytes, long incNamespace) {
+    Pair<Long, Long> quotaPair = bucketUsedQuotaMap.get(bucketName);
+    if (null == quotaPair) {
+      bucketUsedQuotaMap.put(bucketName, Pair.of(incUsedBytes, incNamespace));
+    } else {
+      bucketUsedQuotaMap.put(bucketName, Pair.of(incUsedBytes + quotaPair.getLeft(),
+          incNamespace + quotaPair.getRight()));
+    }
+  }
+  public Map<String, Map<String, CodecBuffer>> getTableRecordsMap() {
+    return tableRecordsMap;
+  }
+  public Map<String, Pair<Long, Long>> getBucketUsedQuotaMap() {
+    return bucketUsedQuotaMap;
+  }
+
+  public void clear() {
+    for (Map<String, CodecBuffer> records : tableRecordsMap.values()) {
+      records.values().forEach(e -> {
+        if (e != null) {
+          e.release();
+        }
+      });
+    }
+    tableRecordsMap.clear();
+    bucketUsedQuotaMap.clear();
+  }
+}