HDDS-14876. Make snapshot diff progress more granular

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/SnapshotDiffHandler.java

@@ -181,6 +181,13 @@ private ObjectNode getJsonObject(SnapshotDiffResponse diffResponse) {
     if (StringUtils.isNotEmpty(diffResponse.getReason())) {
       diffResponseNode.put("reason", diffResponse.getReason());
     }
+    if (diffResponse.getSubStatus() != null) {
+      SnapshotDiffResponse.SubStatus sub = diffResponse.getSubStatus();
+      diffResponseNode.put("subStatus", sub.name());
+      if (sub.hasProgress()) {
+        diffResponseNode.put("progressPercent", diffResponse.getProgressPercent());
+      }
+    }
     return diffResponseNode;
   }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -1468,12 +1468,19 @@ private SnapshotDiffResponse snapshotDiffInternal(String volumeName,
     OzoneManagerProtocolProtos.SnapshotDiffResponse diffResponse =
         omResponse.getSnapshotDiffResponse();
 
-    return new SnapshotDiffResponse(SnapshotDiffReportOzone.fromProtobuf(
-        diffResponse.getSnapshotDiffReport()),
+    SnapshotDiffResponse result = new SnapshotDiffResponse(
+        SnapshotDiffReportOzone.fromProtobuf(diffResponse.getSnapshotDiffReport()),
         JobStatus.fromProtobuf(diffResponse.getJobStatus()),
         diffResponse.getWaitTimeInMs(),
         diffResponse.getReason(),
         reportOnly);
+    if (diffResponse.hasSubStatus()) {
+      result.setSubStatus(SnapshotDiffResponse.SubStatus.fromProtoBuf(diffResponse.getSubStatus()));
+      if (diffResponse.hasProgressPercent()) {
+        result.setProgressPercent(diffResponse.getProgressPercent());
+      }
+    }
+    return result;
   }
 
   /**

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/snapshot/SnapshotDiffResponse.java

@@ -86,10 +86,18 @@ public String getReason() {
     return reason;
   }
 
+  public SubStatus getSubStatus() {
+    return subStatus;
+  }
+
   public void setSubStatus(SubStatus subStatus) {
     this.subStatus = subStatus;
   }
 
+  public double getProgressPercent() {
+    return progressPercent;
+  }
+
   public void setProgressPercent(double progressPercent) {
     this.progressPercent = progressPercent;
   }
@@ -143,11 +151,10 @@ public String toString() {
       str.append(".\n");
       if (subStatus != null) {
         str.append("SubStatus : ")
-            .append(subStatus);
-        if (subStatus.equals(SubStatus.OBJECT_ID_MAP_GEN_OBS) ||
-            subStatus.equals(SubStatus.OBJECT_ID_MAP_GEN_FSO)) {
-          str.append("Keys Processed Estimated Percentage : ")
-              .append(progressPercent);
+            .append(subStatus)
+            .append('\n');
+        if (subStatus.hasProgress()) {
+          str.append(String.format("Keys Processed Estimated Percentage : %.1f%n", progressPercent));
         }
       }
     }
@@ -183,7 +190,12 @@ public enum SubStatus {
     SST_FILE_DELTA_FULL_DIFF,
     OBJECT_ID_MAP_GEN_OBS,
     OBJECT_ID_MAP_GEN_FSO,
-    DIFF_REPORT_GEN;
+    DIFF_REPORT_GEN,
+    PATH_RESOLUTION_FSO;
+
+    public boolean hasProgress() {
+      return this == OBJECT_ID_MAP_GEN_OBS || this == OBJECT_ID_MAP_GEN_FSO;
+    }
 
     public static SubStatus fromProtoBuf(OzoneManagerProtocolProtos.SnapshotDiffResponse.SubStatus subStatusProto) {
       return SubStatus.valueOf(subStatusProto.name());

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/snapshot/TestSnapshotDiffResponse.java

@@ -17,12 +17,16 @@
 
 package org.apache.hadoop.ozone.snapshot;
 
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.Collections;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.JobStatus;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.EnumSource.Mode;
 
 class TestSnapshotDiffResponse {
 
@@ -57,20 +61,46 @@ void testReportOnlyFailedMessageIncludesReason() {
     assertTrue(message.contains("resubmit the job without using the --get-report option"));
   }
 
-  @Test
-  void testReportOnlyInProgressIncludesSubStatusAndProgress() {
+  @ParameterizedTest
+  @EnumSource(value = SubStatus.class, names = {"OBJECT_ID_MAP_GEN_OBS", "OBJECT_ID_MAP_GEN_FSO"})
+  void testInProgressWithMapGenSubStatusIncludesProgress(SubStatus subStatus) {
     SnapshotDiffResponse response = new SnapshotDiffResponse(createReport(),
         JobStatus.IN_PROGRESS, 1000L, true);
-    response.setSubStatus(SubStatus.OBJECT_ID_MAP_GEN_OBS);
+    response.setSubStatus(subStatus);
     response.setProgressPercent(55.5);
 
     String message = response.toString();
     assertTrue(message.contains("IN_PROGRESS"));
-    assertTrue(message.contains("OBJECT_ID_MAP_GEN_OBS"));
+    assertTrue(message.contains(subStatus.name()));
     assertTrue(message.contains("Keys Processed Estimated Percentage"));
     assertTrue(message.contains("55.5"));
   }
 
+  @ParameterizedTest
+  @EnumSource(value = SubStatus.class,
+      names = {"OBJECT_ID_MAP_GEN_OBS", "OBJECT_ID_MAP_GEN_FSO"},
+      mode = Mode.EXCLUDE)
+  void testInProgressWithNonMapGenSubStatusRendersSubStatusButNotProgress(SubStatus subStatus) {
+    SnapshotDiffResponse response = new SnapshotDiffResponse(createReport(), JobStatus.IN_PROGRESS, 1000L, true);
+    response.setSubStatus(subStatus);
+    response.setProgressPercent(55.5);
+
+    String message = response.toString();
+    assertTrue(message.contains("IN_PROGRESS"));
+    assertTrue(message.contains(subStatus.name()));
+    assertFalse(message.contains("Keys Processed Estimated Percentage"));
+    assertFalse(message.contains("55.5"));
+  }
+
+  @Test
+  void testInProgressWithNullSubStatusOmitsSubStatusAndProgressLines() {
+    SnapshotDiffResponse response = new SnapshotDiffResponse(createReport(), JobStatus.IN_PROGRESS, 1000L, true);
+    String message = response.toString();
+    assertTrue(message.contains("IN_PROGRESS"));
+    assertFalse(message.contains("SubStatus"));
+    assertFalse(message.contains("Keys Processed Estimated Percentage"));
+  }
+
   private SnapshotDiffReportOzone createReport() {
     return new SnapshotDiffReportOzone("snapshotRoot", "vol", "bucket", "fromSnap",
         "toSnap", Collections.emptyList(), null);

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -2256,13 +2256,15 @@ message SnapshotDiffResponse {
       OBJECT_ID_MAP_GEN_OBS = 3;
       OBJECT_ID_MAP_GEN_FSO = 4;
       DIFF_REPORT_GEN = 5;
+      PATH_RESOLUTION_FSO = 6;
   }
 
   optional SnapshotDiffReportProto snapshotDiffReport = 1;
   optional JobStatusProto jobStatus = 2;
   optional int64 waitTimeInMs = 3;
   optional string reason = 4;
   optional SubStatus subStatus = 5;
+  optional double progressPercent = 6;
 }
 
 message SubmitSnapshotDiffResponse {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/SnapshotDiffManager.java

@@ -62,6 +62,7 @@
 import static org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus.DIFF_REPORT_GEN;
 import static org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus.OBJECT_ID_MAP_GEN_FSO;
 import static org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus.OBJECT_ID_MAP_GEN_OBS;
+import static org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus.PATH_RESOLUTION_FSO;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
@@ -141,6 +142,7 @@
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse.SubStatus;
 import org.apache.hadoop.ozone.snapshot.SubmitSnapshotDiffResponse;
 import org.apache.hadoop.ozone.util.ClosableIterator;
+import org.apache.hadoop.util.Time;
 import org.apache.logging.log4j.util.Strings;
 import org.apache.ozone.rocksdb.util.SstFileInfo;
 import org.apache.ozone.rocksdiff.RocksDBCheckpointDiffer;
@@ -1112,27 +1114,27 @@ void generateSnapshotDiffReport(final String jobKey,
       // repetition while constantly checking if the job is cancelled.
       Callable<Void>[] methodCalls = new Callable[]{
           () -> {
-            recordActivity(jobKey, OBJECT_ID_MAP_GEN_OBS);
             getDeltaFilesAndDiffKeysToObjectIdToKeyMap(fsKeyTable, tsKeyTable,
                 fsInfo, tsInfo, performNonNativeDiff, tablePrefixes,
                 objectIdToKeyNameMapForFromSnapshot,
                 objectIdToKeyNameMapForToSnapshot, objectIdToIsDirMap,
-                oldParentIds, newParentIds, deltaFileComputer, jobKey);
+                oldParentIds, newParentIds, deltaFileComputer, jobKey, jobId);
             return null;
           },
           () -> {
             if (bucketLayout.isFileSystemOptimized()) {
-              recordActivity(jobKey, OBJECT_ID_MAP_GEN_FSO);
               getDeltaFilesAndDiffKeysToObjectIdToKeyMap(fsDirTable, tsDirTable,
                   fsInfo, tsInfo, performNonNativeDiff, tablePrefixes,
                   objectIdToKeyNameMapForFromSnapshot,
                   objectIdToKeyNameMapForToSnapshot, objectIdToIsDirMap,
-                  oldParentIds, newParentIds, deltaFileComputer, jobKey);
+                  oldParentIds, newParentIds, deltaFileComputer, jobKey, jobId);
             }
             return null;
           },
           () -> {
             if (bucketLayout.isFileSystemOptimized()) {
+              recordActivity(jobKey, PATH_RESOLUTION_FSO);
+              long pathResolutionStart = Time.monotonicNow();
               long bucketId = toSnapshot.getMetadataManager()
                   .getBucketId(volumeName, bucketName);
               String tablePrefix = tablePrefixes.getTablePrefix(fromSnapshot.getMetadataManager()
@@ -1145,11 +1147,19 @@ void generateSnapshotDiffReport(final String jobKey,
                   tablePrefix, bucketId,
                   toSnapshot.getMetadataManager().getDirectoryTable())
                   .getAbsolutePathForObjectIDs(newParentIds, true));
+              if (LOG.isDebugEnabled()) {
+                LOG.debug("Completed FSO path resolution for snapshot diff, resolved {} out of {} parent IDs, " +
+                        "elapsed: {}ms, jobId: {}",
+                    oldParentIdPathMap.get().size() + newParentIdPathMap.get().size(),
+                    oldParentIds.get().size() + newParentIds.get().size(),
+                    Time.monotonicNow() - pathResolutionStart, jobId);
+              }
             }
             return null;
           },
           () -> {
             recordActivity(jobKey, DIFF_REPORT_GEN);
+            long reportGenStart = Time.monotonicNow();
             Pair<Long, String> reportEntries = generateDiffReport(jobId,
                 fsKeyTable,
                 tsKeyTable,
@@ -1166,6 +1176,10 @@ void generateSnapshotDiffReport(final String jobKey,
             if (reportEntries.getKey() >= 0 &&
                 areDiffJobAndSnapshotsActive(volumeName, bucketName,
                     fromSnapshotName, toSnapshotName)) {
+              if (LOG.isDebugEnabled()) {
+                LOG.debug("Generated snapshot diff report, entry count: {}, elapsed: {}ms, jobId: {}",
+                    reportEntries.getKey(), Time.monotonicNow() - reportGenStart, jobId);
+              }
               updateJobStatusToDone(jobKey, reportEntries.getKey(), reportEntries.getValue());
             }
             return null;
@@ -1221,17 +1235,22 @@ private void getDeltaFilesAndDiffKeysToObjectIdToKeyMap(
       final PersistentMap<byte[], byte[]> newObjIdToKeyMap,
       final PersistentMap<byte[], Boolean> objectIdToIsDirMap,
       final Optional<Set<Long>> oldParentIds, final Optional<Set<Long>> newParentIds,
-      final DeltaFileComputer deltaFileComputer, final String jobKey) throws IOException, RocksDBException {
+      final DeltaFileComputer deltaFileComputer, final String jobKey,
+      final String jobId) throws IOException, RocksDBException {
 
+    long deltaFilesStart = Time.monotonicNow();
     Set<String> tablesToLookUp = Collections.singleton(fsTable.getName());
     Collection<Pair<Path, SstFileInfo>> deltaFiles = deltaFileComputer.getDeltaFiles(fsInfo, tsInfo,
         tablesToLookUp);
     if (LOG.isDebugEnabled()) {
-      LOG.debug("Computed Delta SST File Set, Total count = {} ", deltaFiles.size());
+      LOG.debug("Computed Delta SST File Set for table '{}', file count: {}, elapsed: {}ms, jobId: {}",
+          fsTable.getName(), deltaFiles.size(), Time.monotonicNow() - deltaFilesStart, jobId);
     }
+    recordActivity(jobKey,
+        fsTable.getName().equals(DIRECTORY_TABLE) ? OBJECT_ID_MAP_GEN_FSO : OBJECT_ID_MAP_GEN_OBS);
     addToObjectIdMap(fsTable, tsTable, deltaFiles.stream().map(Pair::getLeft).collect(Collectors.toList()),
         !skipNativeDiff, oldObjIdToKeyMap, newObjIdToKeyMap, objectIdToIsDirMap, oldParentIds,
-        newParentIds, tablePrefixes, jobKey);
+        newParentIds, tablePrefixes, jobKey, jobId);
   }
 
   @VisibleForTesting
@@ -1244,10 +1263,14 @@ void addToObjectIdMap(Table<String, ? extends WithParentObjectId> fsTable,
       PersistentMap<byte[], Boolean> objectIdToIsDirMap,
       Optional<Set<Long>> oldParentIds,
       Optional<Set<Long>> newParentIds,
-      TablePrefixInfo tablePrefixes, String jobKey) throws IOException, RocksDBException {
+      TablePrefixInfo tablePrefixes, String jobKey,
+      String jobId) throws IOException, RocksDBException {
+    updateProgress(jobKey, 0.0);
     if (deltaFiles.isEmpty()) {
       return;
     }
+    long objectIdMapStart = Time.monotonicNow();
+    AtomicLong keysProcessed = new AtomicLong(0);
     String tablePrefix = tablePrefixes.getTablePrefix(fsTable.getName());
     boolean isDirectoryTable = fsTable.getName().equals(DIRECTORY_TABLE);
     SstFileSetReader sstFileReader = new SstFileSetReader(deltaFiles);
@@ -1266,7 +1289,6 @@ void addToObjectIdMap(Table<String, ? extends WithParentObjectId> fsTable,
         : sstFileReader.getKeyStream(sstFileReaderLowerBound, sstFileReaderUpperBound);
          TableMergeIterator<String, WithParentObjectId> tableMergeIterator = new TableMergeIterator<>(keysToCheck,
              tablePrefix, (Table<String, WithParentObjectId>) fsTable, (Table<String, WithParentObjectId>) tsTable)) {
-      AtomicLong keysProcessed = new AtomicLong(0);
       while (tableMergeIterator.hasNext()) {
         Table.KeyValue<String, List<WithParentObjectId>> kvs = tableMergeIterator.next();
         String key = kvs.getKey();
@@ -1312,6 +1334,10 @@ void addToObjectIdMap(Table<String, ? extends WithParentObjectId> fsTable,
         }
       }
     }
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Generated object ID map for table '{}', keys scanned: {}, elapsed: {}ms, jobId: {}",
+          fsTable.getName(), keysProcessed.get(), Time.monotonicNow() - objectIdMapStart, jobId);
+    }
   }
 
   private void validateEstimatedKeyChangesAreInLimits(

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/snapshot/diff/delta/CompositeDeltaDiffComputer.java

@@ -86,6 +86,9 @@ Optional<Map<Path, Pair<Path, SstFileInfo>>> computeDeltaFiles(SnapshotInfo from
         updateActivity(SnapshotDiffResponse.SubStatus.SST_FILE_DELTA_DAG_WALK);
         deltaFiles = differComputer.computeDeltaFiles(fromSnapshotInfo, toSnapshotInfo, tablesToLookup,
             tablePrefixInfo).orElse(null);
+        if (deltaFiles == null) {
+          LOG.warn("DAG diff returned no result for tables {}, falling back to full diff.", tablesToLookup);
+        }
       }
     } catch (Exception e) {
       LOG.warn("Falling back to full diff.", e);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java

@@ -1482,6 +1482,12 @@ private SnapshotDiffResponse snapshotDiff(
       builder.setSnapshotDiffReport(
           response.getSnapshotDiffReport().toProtobuf());
     }
+    if (response.getSubStatus() != null) {
+      builder.setSubStatus(response.getSubStatus().toProtoBuf());
+      if (response.getSubStatus().hasProgress()) {
+        builder.setProgressPercent(response.getProgressPercent());
+      }
+    }
 
     return builder.build();
   }

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/snapshot/TestSnapshotDiffManager.java

@@ -158,6 +158,7 @@
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
 import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.Mock;
 import org.mockito.MockedConstruction;
@@ -494,12 +495,16 @@ public void testObjectIdMapWithTombstoneEntries(boolean nativeLibraryLoaded,
       Set<Long> oldParentIds = Sets.newHashSet();
       Set<Long> newParentIds = Sets.newHashSet();
 
+      SnapshotDiffJob dummyJob = new SnapshotDiffJob(System.currentTimeMillis(),
+          "", IN_PROGRESS, VOLUME_NAME, BUCKET_NAME, "from", "to", false, false, 0, null, 0.0, "");
+      db.get().put(snapDiffJobTable, codecRegistry.asRawData(""), codecRegistry.asRawData(dummyJob));
+
       snapshotDiffManager.addToObjectIdMap(toSnapshotTable,
           fromSnapshotTable, Sets.newHashSet(Paths.get("dummy.sst")),
           nativeLibraryLoaded, oldObjectIdKeyMap, newObjectIdKeyMap,
           objectIdsToCheck, Optional.of(oldParentIds),
           Optional.of(newParentIds),
-          new TablePrefixInfo(ImmutableMap.of(DIRECTORY_TABLE, "0", KEY_TABLE, "0", FILE_TABLE, "0")), "");
+          new TablePrefixInfo(ImmutableMap.of(DIRECTORY_TABLE, "0", KEY_TABLE, "0", FILE_TABLE, "0")), "", "");
 
       try (ClosableIterator<Map.Entry<byte[], byte[]>> oldObjectIdIter =
                oldObjectIdKeyMap.iterator()) {
@@ -1674,17 +1679,37 @@ public void testGetSnapshotDiffReportWhenDone() throws Exception {
         .containsExactlyElementsOf(expectedEntries);
   }
 
+  @ParameterizedTest
+  @EnumSource(value = SnapshotDiffResponse.SubStatus.class,
+      names = {"OBJECT_ID_MAP_GEN_OBS", "OBJECT_ID_MAP_GEN_FSO"})
+  public void testGetSnapshotDiffReportReportOnlyInProgressIncludesProgressDetails(
+      SnapshotDiffResponse.SubStatus subStatus) throws IOException {
+    SnapDiffTestContext ctx = setupRandomSnapDiffTestContext();
+    SnapshotDiffJob existing = new SnapshotDiffJob(0L, UUID.randomUUID().toString(),
+        IN_PROGRESS, ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName,
+        false, false, 0L, subStatus, 55.5, null);
+    snapshotDiffManager.getSnapDiffJobTable().put(ctx.diffJobKey, existing);
+
+    SnapshotDiffResponse response = snapshotDiffManager.getSnapshotDiffReport(
+        ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName, "", 1000);
+    assertEquals(IN_PROGRESS, response.getJobStatus());
+    assertEquals(subStatus, response.getSubStatus());
+    assertThat(response.getProgressPercent()).isEqualTo(55.5);
+  }
+
   @Test
-  public void testGetSnapshotDiffReportReportOnlyInProgressIncludesProgressDetails()
+  public void testGetSnapshotDiffReportInProgressWithPathResolutionSubStatus()
       throws IOException {
     SnapDiffTestContext ctx = setupRandomSnapDiffTestContext();
     SnapshotDiffJob existing = new SnapshotDiffJob(0L, UUID.randomUUID().toString(),
         IN_PROGRESS, ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName,
-        false, false, 0L, SnapshotDiffResponse.SubStatus.OBJECT_ID_MAP_GEN_OBS, 55.5, null);
+        false, false, 0L, SnapshotDiffResponse.SubStatus.PATH_RESOLUTION_FSO, 0.0, null);
     snapshotDiffManager.getSnapDiffJobTable().put(ctx.diffJobKey, existing);
 
     SnapshotDiffResponse response = snapshotDiffManager.getSnapshotDiffReport(
         ctx.volumeName, ctx.bucketName, ctx.fromSnapshotName, ctx.toSnapshotName, "", 1000);
     assertEquals(IN_PROGRESS, response.getJobStatus());
+    assertEquals(SnapshotDiffResponse.SubStatus.PATH_RESOLUTION_FSO, response.getSubStatus());
   }
+
 }