HDDS-13785. Remove orphan versions from SnapshotLocalData Yaml file

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -4853,4 +4853,9 @@
     <value>10000</value>
     <description>Maximum number of lock objects that could be present in the pool.</description>
   </property>
+  <property>
+    <name>ozone.om.snapshot.local.data.manager.service.interval</name>
+    <value>5m</value>
+    <description>Interval for cleaning up orphan snapshot local data versions corresponding to snapshots</description>
+  </property>
 </configuration>

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

@@ -678,6 +678,9 @@ public final class OMConfigKeys {
   public static final String OZONE_OM_HIERARCHICAL_RESOURCE_LOCKS_HARD_LIMIT =
       "ozone.om.hierarchical.resource.locks.hard.limit";
   public static final int OZONE_OM_HIERARCHICAL_RESOURCE_LOCKS_HARD_LIMIT_DEFAULT = 10000;
+  public static final String OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL =
+      "ozone.om.snapshot.local.data.manager.service.interval";
+  public static final String OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL_DEFAULT = "5m";
 
   /**
    * Never constructed.

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

@@ -196,9 +196,10 @@ public final class OmSnapshotManager implements AutoCloseable {
   private final AtomicInteger inFlightSnapshotCount = new AtomicInteger(0);
 
   public OmSnapshotManager(OzoneManager ozoneManager) throws IOException {
-    this.snapshotLocalDataManager = new OmSnapshotLocalDataManager(ozoneManager.getMetadataManager());
-    boolean isFilesystemSnapshotEnabled =
-        ozoneManager.isFilesystemSnapshotEnabled();
+    OmMetadataManagerImpl omMetadataManager = (OmMetadataManagerImpl) ozoneManager.getMetadataManager();
+    this.snapshotLocalDataManager = new OmSnapshotLocalDataManager(ozoneManager.getMetadataManager(),
+        omMetadataManager.getSnapshotChainManager(), ozoneManager.getConfiguration());
+    boolean isFilesystemSnapshotEnabled = ozoneManager.isFilesystemSnapshotEnabled();
     LOG.info("Ozone filesystem snapshot feature is {}.",
         isFilesystemSnapshotEnabled ? "enabled" : "disabled");
 
@@ -344,6 +345,16 @@ public OmSnapshotManager(OzoneManager ozoneManager) throws IOException {
     }
   }
 
+  public static boolean isSnapshotPurged(SnapshotChainManager chainManager, OMMetadataManager omMetadataManager,
+      UUID snapshotId, TransactionInfo transactionInfo) throws IOException {
+    String tableKey = chainManager.getTableKey(snapshotId);
+    if (tableKey == null) {
+      return true;
+    }
+    return !omMetadataManager.getSnapshotInfoTable().isExist(tableKey) && transactionInfo != null &&
+        isTransactionFlushedToDisk(omMetadataManager, transactionInfo);
+  }
+
   /**
    * Help reject OM startup if snapshot feature is disabled
    * but there are snapshots remaining in this OM. Note: snapshots that are

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

@@ -17,6 +17,8 @@
 
 package org.apache.hadoop.ozone.om.snapshot;
 
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL;
+import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL_DEFAULT;
 import static org.apache.hadoop.ozone.om.OmSnapshotLocalDataYaml.YAML_FILE_EXTENSION;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -41,17 +43,21 @@
 import java.util.Stack;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.Scheduler;
 import org.apache.hadoop.hdds.utils.TransactionInfo;
 import org.apache.hadoop.hdds.utils.db.RDBStore;
 import org.apache.hadoop.ozone.om.OMMetadataManager;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalData;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalData.VersionMeta;
 import org.apache.hadoop.ozone.om.OmSnapshotLocalDataYaml;
 import org.apache.hadoop.ozone.om.OmSnapshotManager;
+import org.apache.hadoop.ozone.om.SnapshotChainManager;
 import org.apache.hadoop.ozone.om.helpers.SnapshotInfo;
 import org.apache.hadoop.ozone.om.lock.FlatResource;
 import org.apache.hadoop.ozone.om.lock.HierarchicalResourceLockManager;
@@ -73,6 +79,7 @@
 public class OmSnapshotLocalDataManager implements AutoCloseable {
 
   private static final Logger LOG = LoggerFactory.getLogger(OmSnapshotLocalDataManager.class);
+  private static final String LOCAL_DATA_MANAGER_SERVICE_NAME = "OmSnapshotLocalDataManagerService";
 
   private final ObjectSerializer<OmSnapshotLocalData> snapshotLocalDataSerializer;
   // In-memory DAG of snapshot-version dependencies. Each node represents a
@@ -101,8 +108,13 @@ public class OmSnapshotLocalDataManager implements AutoCloseable {
   private final ReadWriteLock internalLock;
   // Locks should be always acquired by iterating through the snapshot chain to avoid deadlocks.
   private HierarchicalResourceLockManager locks;
+  private Map<UUID, Integer> snapshotToBeCheckedForOrphans;
+  private Scheduler scheduler;
+  private volatile boolean closed;
 
-  public OmSnapshotLocalDataManager(OMMetadataManager omMetadataManager) throws IOException {
+  public OmSnapshotLocalDataManager(OMMetadataManager omMetadataManager,
+      SnapshotChainManager snapshotChainManager,
+      OzoneConfiguration configuration) throws IOException {
     this.localDataGraph = GraphBuilder.directed().build();
     this.omMetadataManager = omMetadataManager;
     this.snapshotLocalDataSerializer = new YamlSerializer<OmSnapshotLocalData>(
@@ -116,7 +128,7 @@ public void computeAndSetChecksum(Yaml yaml, OmSnapshotLocalData data) throws IO
     this.versionNodeMap = new ConcurrentHashMap<>();
     this.fullLock = new ReentrantReadWriteLock();
     this.internalLock = new ReentrantReadWriteLock();
-    init();
+    init(configuration, snapshotChainManager);
   }
 
   @VisibleForTesting
@@ -216,7 +228,7 @@ private LocalDataVersionNode getVersionNode(UUID snapshotId, int version) {
 
   private void addSnapshotVersionMeta(UUID snapshotId, SnapshotVersionsMeta snapshotVersionsMeta)
       throws IOException {
-    if (!versionNodeMap.containsKey(snapshotId)) {
+    if (!versionNodeMap.containsKey(snapshotId) && !snapshotVersionsMeta.getSnapshotVersions().isEmpty()) {
       for (LocalDataVersionNode versionNode : snapshotVersionsMeta.getSnapshotVersions().values()) {
         validateVersionAddition(versionNode);
         LocalDataVersionNode previousVersionNode =
@@ -261,8 +273,33 @@ void addVersionNodeWithDependents(OmSnapshotLocalData snapshotLocalData) throws
     }
   }
 
-  private void init() throws IOException {
+  private void incrementOrphanCheckCount(UUID snapshotId) {
+    if (snapshotId != null) {
+      this.snapshotToBeCheckedForOrphans.compute(snapshotId, (k, v) -> v == null ? 1 : (v + 1));
+    }
+  }
+
+  private void decrementOrphanCheckCount(UUID snapshotId, int decrementBy) {
+    this.snapshotToBeCheckedForOrphans.compute(snapshotId, (k, v) -> {
+      if (v == null) {
+        return null;
+      }
+      int newValue = v - decrementBy;
+      if (newValue <= 0) {
+        return null;
+      }
+      return newValue;
+    });
+  }
+
+  @VisibleForTesting
+  Map<UUID, Integer> getSnapshotToBeCheckedForOrphans() {
+    return snapshotToBeCheckedForOrphans;
+  }
+
+  private void init(OzoneConfiguration configuration, SnapshotChainManager chainManager) throws IOException {
     this.locks = omMetadataManager.getHierarchicalLockManager();
+    this.snapshotToBeCheckedForOrphans = new ConcurrentHashMap<>();
     RDBStore store = (RDBStore) omMetadataManager.getStore();
     String checkpointPrefix = store.getDbLocation().getName();
     File snapshotDir = new File(store.getSnapshotsParentDir());
@@ -283,6 +320,74 @@ private void init() throws IOException {
       }
       addVersionNodeWithDependents(snapshotLocalData);
     }
+    for (UUID snapshotId : versionNodeMap.keySet()) {
+      incrementOrphanCheckCount(snapshotId);
+    }
+    long snapshotLocalDataManagerServiceInterval = configuration.getTimeDuration(
+        OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL,
+        OZONE_OM_SNAPSHOT_LOCAL_DATA_MANAGER_SERVICE_INTERVAL_DEFAULT,
+        TimeUnit.MILLISECONDS);
+    if (snapshotLocalDataManagerServiceInterval > 0) {
+      this.scheduler = new Scheduler(LOCAL_DATA_MANAGER_SERVICE_NAME, true, 1);
+      this.scheduler.scheduleWithFixedDelay(
+          () -> {
+            try {
+              checkOrphanSnapshotVersions(omMetadataManager, chainManager);
+            } catch (Exception e) {
+              LOG.error("Exception while checking orphan snapshot versions", e);
+            }
+          }, snapshotLocalDataManagerServiceInterval, snapshotLocalDataManagerServiceInterval, TimeUnit.MILLISECONDS);
+    }
+
+  }
+
+  private void checkOrphanSnapshotVersions(OMMetadataManager metadataManager, SnapshotChainManager chainManager)
+      throws IOException {
+    for (Map.Entry<UUID, Integer> entry : snapshotToBeCheckedForOrphans.entrySet()) {
+      UUID snapshotId = entry.getKey();
+      int countBeforeCheck = entry.getValue();
+      checkOrphanSnapshotVersions(metadataManager, chainManager, snapshotId);
+      decrementOrphanCheckCount(snapshotId, countBeforeCheck);
+    }
+  }
+
+  @VisibleForTesting
+  void checkOrphanSnapshotVersions(OMMetadataManager metadataManager, SnapshotChainManager chainManager,
+      UUID snapshotId) throws IOException {
+    LOG.info("Checking orphan snapshot versions for snapshot {}", snapshotId);
+    try (WritableOmSnapshotLocalDataProvider snapshotLocalDataProvider = new WritableOmSnapshotLocalDataProvider(
+        snapshotId)) {
+      OmSnapshotLocalData snapshotLocalData = snapshotLocalDataProvider.getSnapshotLocalData();
+      boolean isSnapshotPurged = OmSnapshotManager.isSnapshotPurged(chainManager, metadataManager, snapshotId,
+          snapshotLocalData.getTransactionInfo());
+      for (Map.Entry<Integer, LocalDataVersionNode> integerLocalDataVersionNodeEntry : getVersionNodeMap()
+          .get(snapshotId).getSnapshotVersions().entrySet()) {
+        LocalDataVersionNode versionEntry = integerLocalDataVersionNodeEntry.getValue();
+        // remove the version entry if it is not referenced by any other snapshot version node. For version node 0
+        // a newly created snapshot version could point to a version with indegree 0 in such a scenario a version 0
+        // node can be only deleted if the snapshot is also purged.
+        internalLock.readLock().lock();
+        try {
+          boolean toRemove = localDataGraph.inDegree(versionEntry) == 0
+              && ((versionEntry.getVersion() != 0 && versionEntry.getVersion() != snapshotLocalData.getVersion())
+              || isSnapshotPurged);
+          if (toRemove) {
+            LOG.info("Removing snapshot Id : {} version: {} from local data, snapshotLocalDataVersion : {}, " +
+                    "snapshotPurged: {}, inDegree : {}", snapshotId, versionEntry.getVersion(),
+                snapshotLocalData.getVersion(), isSnapshotPurged, localDataGraph.inDegree(versionEntry));
+            snapshotLocalDataProvider.removeVersion(versionEntry.getVersion());
+          }
+        } finally {
+          internalLock.readLock().unlock();
+        }
+      }
+      // If Snapshot is purged but not flushed completely to disk then this needs to wait for the next iteration
+      // which can be done by incrementing the orphan check count for the snapshotId.
+      if (!snapshotLocalData.getVersionSstFileInfos().isEmpty() && snapshotLocalData.getTransactionInfo() != null) {
+        incrementOrphanCheckCount(snapshotId);
+      }
+      snapshotLocalDataProvider.commit();
+    }
   }
 
   /**
@@ -326,13 +431,19 @@ private void validateVersionAddition(LocalDataVersionNode versionNode) throws IO
   }
 
   @Override
-  public void close() {
-    if (snapshotLocalDataSerializer != null) {
-      try {
-        snapshotLocalDataSerializer.close();
-      } catch (IOException e) {
-        LOG.error("Failed to close snapshot local data serializer", e);
+  public synchronized void close() {
+    if (!closed) {
+      if (snapshotLocalDataSerializer != null) {
+        try {
+          snapshotLocalDataSerializer.close();
+        } catch (IOException e) {
+          LOG.error("Failed to close snapshot local data serializer", e);
+        }
       }
+      if (scheduler != null) {
+        scheduler.close();
+      }
+      closed = true;
     }
   }
 
@@ -730,30 +841,66 @@ public synchronized void commit() throws IOException {
       // Need to update the disk state if and only if the dirty bit is set.
       if (isDirty()) {
         String filePath = getSnapshotLocalPropertyYamlPath(super.snapshotId);
-        String tmpFilePath = filePath + ".tmp";
-        File tmpFile = new File(tmpFilePath);
-        boolean tmpFileExists = tmpFile.exists();
-        if (tmpFileExists) {
-          tmpFileExists = !tmpFile.delete();
-        }
-        if (tmpFileExists) {
-          throw new IOException("Unable to delete tmp file " + tmpFilePath);
+        File snapshotLocalDataFile = new File(filePath);
+        if (!localDataVersionNodes.getSnapshotVersions().isEmpty()) {
+          String tmpFilePath = filePath + ".tmp";
+          File tmpFile = new File(tmpFilePath);
+          boolean tmpFileExists = tmpFile.exists();
+          if (tmpFileExists) {
+            tmpFileExists = !tmpFile.delete();
+          }
+          if (tmpFileExists) {
+            throw new IOException("Unable to delete tmp file " + tmpFilePath);
+          }
+          snapshotLocalDataSerializer.save(new File(tmpFilePath), super.snapshotLocalData);
+          Files.move(tmpFile.toPath(), Paths.get(filePath), StandardCopyOption.ATOMIC_MOVE,
+              StandardCopyOption.REPLACE_EXISTING);
+        } else if (snapshotLocalDataFile.exists()) {
+          LOG.info("Deleting YAML file corresponding to snapshotId: {} in path : {}",
+              super.snapshotId, snapshotLocalDataFile.getAbsolutePath());
+          if (!snapshotLocalDataFile.delete()) {
+            throw new IOException("Unable to delete file " + snapshotLocalDataFile.getAbsolutePath());
+          }
         }
-        snapshotLocalDataSerializer.save(new File(tmpFilePath), super.snapshotLocalData);
-        Files.move(tmpFile.toPath(), Paths.get(filePath), StandardCopyOption.ATOMIC_MOVE,
-            StandardCopyOption.REPLACE_EXISTING);
-        upsertNode(super.snapshotId, localDataVersionNodes);
+        SnapshotVersionsMeta previousVersionMeta = upsertNode(super.snapshotId, localDataVersionNodes);
+        checkForOphanVersionsAndIncrementCount(super.snapshotId, previousVersionMeta, localDataVersionNodes,
+            getSnapshotLocalData().getTransactionInfo() != null);
         // Reset dirty bit
         resetDirty();
       }
     }
 
-    private void upsertNode(UUID snapshotId, SnapshotVersionsMeta snapshotVersions) throws IOException {
+    private void checkForOphanVersionsAndIncrementCount(UUID snapshotId, SnapshotVersionsMeta previousVersionsMeta,
+        SnapshotVersionsMeta currentVersionMeta, boolean isPurgeTransactionSet) {
+      if (previousVersionsMeta != null) {
+        Map<Integer, LocalDataVersionNode> currentVersionNodeMap = currentVersionMeta.getSnapshotVersions();
+        Map<Integer, LocalDataVersionNode> previousVersionNodeMap = previousVersionsMeta.getSnapshotVersions();
+        boolean versionsRemoved = previousVersionNodeMap.keySet().stream()
+            .anyMatch(version -> !currentVersionNodeMap.containsKey(version));
+
+        // The previous snapshotId could have become an orphan entry or could have orphan versions.(In case of
+        // version removals)
+        if (versionsRemoved || !Objects.equals(previousVersionsMeta.getPreviousSnapshotId(),
+            currentVersionMeta.getPreviousSnapshotId())) {
+          incrementOrphanCheckCount(previousVersionsMeta.getPreviousSnapshotId());
+        }
+        // If the transactionInfo set, this means the snapshot has been purged and the entire YAML file could have
+        // become an orphan. Otherwise if the version is updated it
+        // could mean that there could be some orphan version present within the
+        // same snapshot.
+        if (isPurgeTransactionSet || previousVersionsMeta.getVersion() != currentVersionMeta.getVersion()) {
+          incrementOrphanCheckCount(snapshotId);
+        }
+      }
+    }
+
+    private SnapshotVersionsMeta upsertNode(UUID snapshotId, SnapshotVersionsMeta snapshotVersions) throws IOException {
       internalLock.writeLock().lock();
       try {
         SnapshotVersionsMeta existingSnapVersions = getVersionNodeMap().remove(snapshotId);
         Map<Integer, LocalDataVersionNode> existingVersions = existingSnapVersions == null ? Collections.emptyMap() :
             existingSnapVersions.getSnapshotVersions();
+        Map<Integer, LocalDataVersionNode> newVersions = snapshotVersions.getSnapshotVersions();
         Map<Integer, List<LocalDataVersionNode>> predecessors = new HashMap<>();
         // Track all predecessors of the existing versions and remove the node from the graph.
         for (Map.Entry<Integer, LocalDataVersionNode> existingVersion : existingVersions.entrySet()) {
@@ -763,14 +910,16 @@ private void upsertNode(UUID snapshotId, SnapshotVersionsMeta snapshotVersions)
           predecessors.put(existingVersion.getKey(), new ArrayList<>(localDataGraph.predecessors(existingVersionNode)));
           localDataGraph.removeNode(existingVersionNode);
         }
+
         // Add the nodes to be added in the graph and map.
         addSnapshotVersionMeta(snapshotId, snapshotVersions);
         // Reconnect all the predecessors for existing nodes.
-        for (Map.Entry<Integer, LocalDataVersionNode> entry : snapshotVersions.getSnapshotVersions().entrySet()) {
+        for (Map.Entry<Integer, LocalDataVersionNode> entry : newVersions.entrySet()) {
           for (LocalDataVersionNode predecessor : predecessors.getOrDefault(entry.getKey(), Collections.emptyList())) {
             localDataGraph.putEdge(predecessor, entry.getValue());
           }
         }
+        return existingSnapVersions;
       } finally {
         internalLock.writeLock().unlock();
       }