HDDS-7810. Support namespace summaries (du, dist & counts) for OBJECT_STORE buckets. (apache#4245)

(cherry picked from commit 0a5fc69)

Author: ArafatKhan2198

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/handlers/BucketHandler.java

@@ -178,9 +178,7 @@ public static BucketHandler getBucketHandler(
             omMetadataManager, reconSCM, bucketInfo);
       } else if (bucketInfo.getBucketLayout()
           .equals(BucketLayout.OBJECT_STORE)) {
-        // TODO: HDDS-7810 Write a handler for object store bucket
-        // We can use LegacyBucketHandler for OBS bucket for now.
-        return new LegacyBucketHandler(reconNamespaceSummaryManager,
+        return new OBSBucketHandler(reconNamespaceSummaryManager,
             omMetadataManager, reconSCM, bucketInfo);
       } else {
         LOG.error("Unsupported bucket layout: " +

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/handlers/FSOBucketHandler.java

@@ -42,7 +42,7 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
 
 /**
- * Class for handling FSO buckets.
+ * Class for handling FSO buckets NameSpaceSummaries.
  */
 public class FSOBucketHandler extends BucketHandler {
   private static final Logger LOG =

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/handlers/LegacyBucketHandler.java

@@ -41,7 +41,7 @@
 import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
 
 /**
- * Class for handling Legacy buckets.
+ * Class for handling Legacy buckets NameSpaceSummaries.
  */
 public class LegacyBucketHandler extends BucketHandler {
 

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/api/handlers/OBSBucketHandler.java

@@ -0,0 +1,268 @@
+/*
+ * 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.hadoop.ozone.recon.api.handlers;
+
+
+import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.TableIterator;
+import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.OmBucketInfo;
+import org.apache.hadoop.ozone.om.helpers.OmDirectoryInfo;
+import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
+import org.apache.hadoop.ozone.recon.api.types.DUResponse;
+import org.apache.hadoop.ozone.recon.api.types.EntityType;
+import org.apache.hadoop.ozone.recon.api.types.NSSummary;
+import org.apache.hadoop.ozone.recon.recovery.ReconOMMetadataManager;
+import org.apache.hadoop.ozone.recon.spi.ReconNamespaceSummaryManager;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+
+/**
+ * Class for handling OBS buckets NameSpaceSummaries.
+ */
+public class OBSBucketHandler extends BucketHandler {
+
+  private final String vol;
+  private final String bucket;
+  private final OmBucketInfo omBucketInfo;
+
+  public OBSBucketHandler(
+      ReconNamespaceSummaryManager reconNamespaceSummaryManager,
+      ReconOMMetadataManager omMetadataManager,
+      OzoneStorageContainerManager reconSCM,
+      OmBucketInfo bucketInfo) {
+    super(reconNamespaceSummaryManager, omMetadataManager,
+        reconSCM);
+    this.omBucketInfo = bucketInfo;
+    this.vol = omBucketInfo.getVolumeName();
+    this.bucket = omBucketInfo.getBucketName();
+  }
+
+  /**
+   * Helper function to check if a path is a key, or invalid.
+   *
+   * @param keyName key name
+   * @return KEY, or UNKNOWN
+   * @throws IOException
+   */
+  @Override
+  public EntityType determineKeyPath(String keyName) throws IOException {
+    String key = OM_KEY_PREFIX + vol +
+        OM_KEY_PREFIX + bucket +
+        OM_KEY_PREFIX + keyName;
+
+    Table<String, OmKeyInfo> keyTable = getKeyTable();
+
+    try (
+        TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>>
+            iterator = keyTable.iterator()) {
+      iterator.seek(key);
+      if (iterator.hasNext()) {
+        Table.KeyValue<String, OmKeyInfo> kv = iterator.next();
+        String dbKey = kv.getKey();
+        if (dbKey.equals(key)) {
+          return EntityType.KEY;
+        }
+      }
+    }
+    return EntityType.UNKNOWN;
+  }
+
+  /**
+   * This method handles disk usage of direct keys.
+   *
+   * @param parentId       The identifier for the parent bucket.
+   * @param withReplica    if withReplica is enabled, set sizeWithReplica
+   *                       for each direct key's DU
+   * @param listFile       if listFile is enabled, append key DU as a children
+   *                       keys
+   * @param duData         the current DU data
+   * @param normalizedPath the normalized path request
+   * @return the total DU of all direct keys
+   * @throws IOException IOE
+   */
+  @Override
+  public long handleDirectKeys(long parentId, boolean withReplica,
+                               boolean listFile,
+                               List<DUResponse.DiskUsage> duData,
+                               String normalizedPath) throws IOException {
+
+    NSSummary nsSummary = getReconNamespaceSummaryManager()
+        .getNSSummary(parentId);
+    // Handle the case of an empty bucket.
+    if (nsSummary == null) {
+      return 0;
+    }
+
+    Table<String, OmKeyInfo> keyTable = getKeyTable();
+    long keyDataSizeWithReplica = 0L;
+
+    try (
+        TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>>
+            iterator = keyTable.iterator()) {
+
+      String seekPrefix = OM_KEY_PREFIX +
+          vol +
+          OM_KEY_PREFIX +
+          bucket +
+          OM_KEY_PREFIX;
+
+      iterator.seek(seekPrefix);
+
+      while (iterator.hasNext()) {
+        // KeyName : OmKeyInfo-Object
+        Table.KeyValue<String, OmKeyInfo> kv = iterator.next();
+        String dbKey = kv.getKey();
+
+        // Exit loop if the key doesn't match the seekPrefix.
+        if (!dbKey.startsWith(seekPrefix)) {
+          break;
+        }
+
+        OmKeyInfo keyInfo = kv.getValue();
+        if (keyInfo != null) {
+          DUResponse.DiskUsage diskUsage = new DUResponse.DiskUsage();
+          String objectName = keyInfo.getKeyName();
+          diskUsage.setSubpath(objectName);
+          diskUsage.setKey(true);
+          diskUsage.setSize(keyInfo.getDataSize());
+
+          if (withReplica) {
+            long keyDU = keyInfo.getReplicatedSize();
+            keyDataSizeWithReplica += keyDU;
+            diskUsage.setSizeWithReplica(keyDU);
+          }
+          // List all the keys for the OBS bucket if requested.
+          if (listFile) {
+            duData.add(diskUsage);
+          }
+        }
+      }
+    }
+
+    return keyDataSizeWithReplica;
+  }
+
+  /**
+   * Calculates the total disk usage (DU) for an Object Store Bucket (OBS) by
+   * summing the sizes of all keys contained within the bucket.
+   * Since OBS buckets operate on a flat hierarchy, this method iterates through
+   * all the keys in the bucket without the need to traverse directories.
+   *
+   * @param parentId The identifier for the parent bucket.
+   * @return The total disk usage of all keys within the specified OBS bucket.
+   * @throws IOException
+   */
+  @Override
+  public long calculateDUUnderObject(long parentId) throws IOException {
+    // Initialize the total disk usage variable.
+    long totalDU = 0L;
+
+    // Access the key table for the bucket.
+    Table<String, OmKeyInfo> keyTable = getKeyTable();
+
+    try (
+        TableIterator<String, ? extends Table.KeyValue<String, OmKeyInfo>>
+            iterator = keyTable.iterator()) {
+      // Construct the seek prefix to filter keys under this bucket.
+      String seekPrefix =
+          OM_KEY_PREFIX + vol + OM_KEY_PREFIX + bucket + OM_KEY_PREFIX;
+      iterator.seek(seekPrefix);
+
+      // Iterate over keys in the bucket.
+      while (iterator.hasNext()) {
+        Table.KeyValue<String, OmKeyInfo> kv = iterator.next();
+        String keyName = kv.getKey();
+
+        // Break the loop if the current key does not start with the seekPrefix.
+        if (!keyName.startsWith(seekPrefix)) {
+          break;
+        }
+
+        // Sum the size of each key to the total disk usage.
+        OmKeyInfo keyInfo = kv.getValue();
+        if (keyInfo != null) {
+          totalDU += keyInfo.getDataSize();
+        }
+      }
+    }
+
+    // Return the total disk usage of all keys in the bucket.
+    return totalDU;
+  }
+
+  /**
+   * Object stores do not support directories.
+   *
+   * @throws UnsupportedOperationException
+   */
+  @Override
+  public long getDirObjectId(String[] names)
+      throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Object stores do not support directories.");
+  }
+
+  /**
+   * Object stores do not support directories.
+   *
+   * @throws UnsupportedOperationException
+   */
+  @Override
+  public long getDirObjectId(String[] names, int cutoff)
+      throws UnsupportedOperationException {
+    throw new UnsupportedOperationException(
+        "Object stores do not support directories.");
+  }
+
+  /**
+   * Returns the keyInfo object from the KEY table.
+   * @return OmKeyInfo
+   */
+  @Override
+  public OmKeyInfo getKeyInfo(String[] names) throws IOException {
+    String ozoneKey = OM_KEY_PREFIX;
+    ozoneKey += String.join(OM_KEY_PREFIX, names);
+
+    return getKeyTable().getSkipCache(ozoneKey);
+  }
+
+  /**
+   * Object stores do not support directories.
+   *
+   * @throws UnsupportedOperationException
+   */
+  @Override
+  public OmDirectoryInfo getDirInfo(String[] names) throws IOException {
+    throw new UnsupportedOperationException(
+        "Object stores do not support directories.");
+  }
+
+  public Table<String, OmKeyInfo> getKeyTable() {
+    return getOmMetadataManager().getKeyTable(getBucketLayout());
+  }
+
+  public BucketLayout getBucketLayout() {
+    return BucketLayout.OBJECT_STORE;
+  }
+
+}

hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/tasks/NSSummaryTask.java

@@ -38,6 +38,7 @@
 import java.util.concurrent.Callable;
 import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Task to query data from OMDB and write into Recon RocksDB.
@@ -68,6 +69,7 @@ public class NSSummaryTask implements ReconOmTask {
   private final ReconOMMetadataManager reconOMMetadataManager;
   private final NSSummaryTaskWithFSO nsSummaryTaskWithFSO;
   private final NSSummaryTaskWithLegacy nsSummaryTaskWithLegacy;
+  private final NSSummaryTaskWithOBS nsSummaryTaskWithOBS;
   private final OzoneConfiguration ozoneConfiguration;
 
   @Inject
@@ -86,6 +88,9 @@ public NSSummaryTask(ReconNamespaceSummaryManager
     this.nsSummaryTaskWithLegacy = new NSSummaryTaskWithLegacy(
         reconNamespaceSummaryManager,
         reconOMMetadataManager, ozoneConfiguration);
+    this.nsSummaryTaskWithOBS = new NSSummaryTaskWithOBS(
+        reconNamespaceSummaryManager,
+        reconOMMetadataManager, ozoneConfiguration);
   }
 
   @Override
@@ -95,20 +100,28 @@ public String getTaskName() {
 
   @Override
   public Pair<String, Boolean> process(OMUpdateEventBatch events) {
-    boolean success;
-    success = nsSummaryTaskWithFSO.processWithFSO(events);
-    if (success) {
-      success = nsSummaryTaskWithLegacy.processWithLegacy(events);
-    } else {
+    boolean success = nsSummaryTaskWithFSO.processWithFSO(events);
+    if (!success) {
       LOG.error("processWithFSO failed.");
     }
+    success = nsSummaryTaskWithLegacy.processWithLegacy(events);
+    if (!success) {
+      LOG.error("processWithLegacy failed.");
+    }
+    success = nsSummaryTaskWithOBS.processWithOBS(events);
+    if (!success) {
+      LOG.error("processWithOBS failed.");
+    }
     return new ImmutablePair<>(getTaskName(), success);
   }
 
   @Override
   public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
+    // Initialize a list of tasks to run in parallel
     Collection<Callable<Boolean>> tasks = new ArrayList<>();
 
+    long startTime = System.nanoTime(); // Record start time
+
     try {
       // reinit Recon RocksDB's namespace CF.
       reconNamespaceSummaryManager.clearNSSummaryTable();
@@ -122,6 +135,8 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
         .reprocessWithFSO(omMetadataManager));
     tasks.add(() -> nsSummaryTaskWithLegacy
         .reprocessWithLegacy(reconOMMetadataManager));
+    tasks.add(() -> nsSummaryTaskWithOBS
+        .reprocessWithOBS(reconOMMetadataManager));
 
     List<Future<Boolean>> results;
     ThreadFactory threadFactory = new ThreadFactoryBuilder()
@@ -137,17 +152,25 @@ public Pair<String, Boolean> reprocess(OMMetadataManager omMetadataManager) {
         }
       }
     } catch (InterruptedException ex) {
-      LOG.error("Error while reprocessing NSSummary " +
-          "table in Recon DB. ", ex);
+      LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex);
       return new ImmutablePair<>(getTaskName(), false);
     } catch (ExecutionException ex2) {
-      LOG.error("Error while reprocessing NSSummary " +
-          "table in Recon DB. ", ex2);
+      LOG.error("Error while reprocessing NSSummary table in Recon DB.", ex2);
       return new ImmutablePair<>(getTaskName(), false);
     } finally {
       executorService.shutdown();
+
+      long endTime = System.nanoTime();
+      // Convert to milliseconds
+      long durationInMillis =
+          TimeUnit.NANOSECONDS.toMillis(endTime - startTime);
+
+      // Log performance metrics
+      LOG.info("Task execution time: {} milliseconds", durationInMillis);
     }
+
     return new ImmutablePair<>(getTaskName(), true);
   }
+
 }