swamirishi commented on code in PR #5339:
URL: https://github.com/apache/ozone/pull/5339#discussion_r1357234086
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java:
##########
@@ -223,12 +227,153 @@ public BackgroundTaskResult call() {
getOzoneManager().getMetadataManager().getTableLock(
OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
}
+
+ try {
+ if (remainNum > 0) {
+ expandSnapshotDirectories(remainNum);
+ }
+ } catch (Exception e) {
+ LOG.error("Error while running deep clean on snapshots. Will " +
+ "retry at next run.", e);
+ }
}
// place holder by returning empty results of this call back.
return BackgroundTaskResult.EmptyTaskResult.newResult();
}
+ private void expandSnapshotDirectories(long remainNum) throws IOException {
+ OmSnapshotManager omSnapshotManager =
+ getOzoneManager().getOmSnapshotManager();
+ Table<String, SnapshotInfo> snapshotInfoTable =
+ getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+ long dirNum = 0L;
+ long subDirNum = 0L;
+ long subFileNum = 0L;
+ int consumedSize = 0;
+ List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+ try (TableIterator<String, ? extends Table.KeyValue
+ <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+ while (remainNum > 0 && iterator.hasNext()) {
+ SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+ // Expand deleted dirs only on active snapshot. Deleted Snapshots
+ // will be cleaned up by SnapshotDeletingService.
+ if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
+ currSnapInfo.getExpandedDeletedDir()) {
+ continue;
+ }
+
+ long volumeId = getOzoneManager().getMetadataManager()
+ .getVolumeId(currSnapInfo.getVolumeName());
+ // Get bucketInfo for the snapshot bucket to get bucket layout.
+ String dbBucketKey = getOzoneManager().getMetadataManager()
+ .getBucketKey(currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName());
+ OmBucketInfo bucketInfo = getOzoneManager().getMetadataManager()
+ .getBucketTable().get(dbBucketKey);
+
+ if (bucketInfo == null) {
+ throw new IllegalStateException("Bucket " + "/" +
+ currSnapInfo.getVolumeName() + "/" + currSnapInfo
+ .getBucketName() + " is not found. BucketInfo should not be " +
+ "null for snapshotted bucket. The OM is in unexpected state.");
+ }
+
+ String dbBucketKeyForDir = getOzoneManager().getMetadataManager()
Review Comment:
Can we have some kind of a util function for this? or use one if it is
already there. As in appending OM_KEY_PREFIX at the end.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java:
##########
@@ -223,12 +227,153 @@ public BackgroundTaskResult call() {
getOzoneManager().getMetadataManager().getTableLock(
OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
}
+
+ try {
+ if (remainNum > 0) {
+ expandSnapshotDirectories(remainNum);
+ }
+ } catch (Exception e) {
+ LOG.error("Error while running deep clean on snapshots. Will " +
+ "retry at next run.", e);
+ }
}
// place holder by returning empty results of this call back.
return BackgroundTaskResult.EmptyTaskResult.newResult();
}
+ private void expandSnapshotDirectories(long remainNum) throws IOException {
+ OmSnapshotManager omSnapshotManager =
+ getOzoneManager().getOmSnapshotManager();
+ Table<String, SnapshotInfo> snapshotInfoTable =
+ getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+ long dirNum = 0L;
+ long subDirNum = 0L;
+ long subFileNum = 0L;
+ int consumedSize = 0;
+ List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+ try (TableIterator<String, ? extends Table.KeyValue
+ <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+ while (remainNum > 0 && iterator.hasNext()) {
+ SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+ // Expand deleted dirs only on active snapshot. Deleted Snapshots
+ // will be cleaned up by SnapshotDeletingService.
+ if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
+ currSnapInfo.getExpandedDeletedDir()) {
+ continue;
+ }
+
+ long volumeId = getOzoneManager().getMetadataManager()
+ .getVolumeId(currSnapInfo.getVolumeName());
+ // Get bucketInfo for the snapshot bucket to get bucket layout.
+ String dbBucketKey = getOzoneManager().getMetadataManager()
+ .getBucketKey(currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName());
+ OmBucketInfo bucketInfo = getOzoneManager().getMetadataManager()
+ .getBucketTable().get(dbBucketKey);
+
+ if (bucketInfo == null) {
+ throw new IllegalStateException("Bucket " + "/" +
+ currSnapInfo.getVolumeName() + "/" + currSnapInfo
+ .getBucketName() + " is not found. BucketInfo should not be " +
+ "null for snapshotted bucket. The OM is in unexpected state.");
+ }
+
+ String dbBucketKeyForDir = getOzoneManager().getMetadataManager()
+ .getBucketKey(Long.toString(volumeId),
+ Long.toString(bucketInfo.getObjectID())) + OM_KEY_PREFIX;
+
+ try (ReferenceCounted<IOmMetadataReader, SnapshotCache>
+ rcCurrOmSnapshot = omSnapshotManager.checkForSnapshot(
+ currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName(),
+ getSnapshotPrefix(currSnapInfo.getName()),
+ true)) {
+
+ OmSnapshot currOmSnapshot = (OmSnapshot) rcCurrOmSnapshot.get();
+ Table<String, OmKeyInfo> snapDeletedDirTable =
+ currOmSnapshot.getMetadataManager().getDeletedDirTable();
+
+ if (snapDeletedDirTable.isEmpty()) {
+ // TODO: [SNAPSHOT] Update Snapshot state using
+ // SetSnapshotProperty from HDDS-7743 (YET TO BE MERGED)
+ continue;
+ }
+
+ List<Pair<String, OmKeyInfo>> allSubDirList
+ = new ArrayList<>((int) remainNum);
+
+ try (TableIterator<String, ? extends Table.KeyValue<String,
+ OmKeyInfo>> deletedIterator = snapDeletedDirTable.iterator()) {
+
+ long startTime = Time.monotonicNow();
+ deletedIterator.seek(dbBucketKeyForDir);
+
+ while (remainNum > 0 && deletedIterator.hasNext()) {
+ Table.KeyValue<String, OmKeyInfo> deletedDirInfo =
+ deletedIterator.next();
+ String deletedDirKey = deletedDirInfo.getKey();
+
+ // Exit if it is out of the bucket scope.
+ if (!deletedDirKey.startsWith(dbBucketKeyForDir)) {
+ break;
+ }
+
+ PurgePathRequest request = prepareDeleteDirRequest(
+ remainNum, deletedDirInfo.getValue(),
+ deletedDirInfo.getKey(), allSubDirList,
+ currOmSnapshot.getKeyManager());
+ if (isBufferLimitCrossed(ratisByteLimit, consumedSize,
+ request.getSerializedSize())) {
+ if (purgePathRequestList.size() != 0) {
+ // if message buffer reaches max limit,
+ // avoid sending further
+ remainNum = 0;
+ break;
+ }
+ // if directory itself is having a lot of keys / files,
+ // reduce capacity to minimum level
+ remainNum = MIN_ERR_LIMIT_PER_TASK;
+ request = prepareDeleteDirRequest(
+ remainNum, deletedDirInfo.getValue(),
+ deletedDirInfo.getKey(), allSubDirList,
+ currOmSnapshot.getKeyManager());
+ }
+
+ consumedSize += request.getSerializedSize();
+ purgePathRequestList.add(request);
+ remainNum = remainNum - request.getDeletedSubFilesCount();
+ remainNum = remainNum - request.getMarkDeletedSubDirsCount();
+ // Count up the purgeDeletedDir, subDirs and subFiles
+ if (request.hasDeletedDir() &&
+ !request.getDeletedDir().isEmpty()) {
+ dirNum++;
+ }
+ subDirNum += request.getMarkDeletedSubDirsCount();
+ subFileNum += request.getDeletedSubFilesCount();
+ }
+
+ optimizeDirDeletesAndSubmitRequest(
Review Comment:
From what I understand this would put a request to expand the directory and
manipulate the directory table and key table entries. I am not sure how this
would impact snapshot diff. @hemantk-12 @prashantpogde .
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java:
##########
@@ -223,12 +227,153 @@ public BackgroundTaskResult call() {
getOzoneManager().getMetadataManager().getTableLock(
OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
}
+
+ try {
+ if (remainNum > 0) {
+ expandSnapshotDirectories(remainNum);
+ }
+ } catch (Exception e) {
+ LOG.error("Error while running deep clean on snapshots. Will " +
+ "retry at next run.", e);
+ }
}
// place holder by returning empty results of this call back.
return BackgroundTaskResult.EmptyTaskResult.newResult();
}
+ private void expandSnapshotDirectories(long remainNum) throws IOException {
+ OmSnapshotManager omSnapshotManager =
+ getOzoneManager().getOmSnapshotManager();
+ Table<String, SnapshotInfo> snapshotInfoTable =
+ getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+ long dirNum = 0L;
+ long subDirNum = 0L;
+ long subFileNum = 0L;
+ int consumedSize = 0;
+ List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+ try (TableIterator<String, ? extends Table.KeyValue
+ <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+ while (remainNum > 0 && iterator.hasNext()) {
+ SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+ // Expand deleted dirs only on active snapshot. Deleted Snapshots
+ // will be cleaned up by SnapshotDeletingService.
+ if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
Review Comment:
We should take a lock on this snapshot to avoid race condition
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java:
##########
@@ -223,12 +227,153 @@ public BackgroundTaskResult call() {
getOzoneManager().getMetadataManager().getTableLock(
OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
}
+
+ try {
+ if (remainNum > 0) {
+ expandSnapshotDirectories(remainNum);
+ }
+ } catch (Exception e) {
+ LOG.error("Error while running deep clean on snapshots. Will " +
+ "retry at next run.", e);
+ }
}
// place holder by returning empty results of this call back.
return BackgroundTaskResult.EmptyTaskResult.newResult();
}
+ private void expandSnapshotDirectories(long remainNum) throws IOException {
+ OmSnapshotManager omSnapshotManager =
+ getOzoneManager().getOmSnapshotManager();
+ Table<String, SnapshotInfo> snapshotInfoTable =
+ getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+ long dirNum = 0L;
+ long subDirNum = 0L;
+ long subFileNum = 0L;
+ int consumedSize = 0;
+ List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+ try (TableIterator<String, ? extends Table.KeyValue
+ <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+ while (remainNum > 0 && iterator.hasNext()) {
+ SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+ // Expand deleted dirs only on active snapshot. Deleted Snapshots
+ // will be cleaned up by SnapshotDeletingService.
+ if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
+ currSnapInfo.getExpandedDeletedDir()) {
+ continue;
+ }
+
+ long volumeId = getOzoneManager().getMetadataManager()
+ .getVolumeId(currSnapInfo.getVolumeName());
+ // Get bucketInfo for the snapshot bucket to get bucket layout.
+ String dbBucketKey = getOzoneManager().getMetadataManager()
+ .getBucketKey(currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName());
+ OmBucketInfo bucketInfo = getOzoneManager().getMetadataManager()
+ .getBucketTable().get(dbBucketKey);
+
+ if (bucketInfo == null) {
+ throw new IllegalStateException("Bucket " + "/" +
+ currSnapInfo.getVolumeName() + "/" + currSnapInfo
+ .getBucketName() + " is not found. BucketInfo should not be " +
+ "null for snapshotted bucket. The OM is in unexpected state.");
+ }
+
+ String dbBucketKeyForDir = getOzoneManager().getMetadataManager()
+ .getBucketKey(Long.toString(volumeId),
+ Long.toString(bucketInfo.getObjectID())) + OM_KEY_PREFIX;
+
+ try (ReferenceCounted<IOmMetadataReader, SnapshotCache>
+ rcCurrOmSnapshot = omSnapshotManager.checkForSnapshot(
+ currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName(),
+ getSnapshotPrefix(currSnapInfo.getName()),
+ true)) {
Review Comment:
I guess this should be false here. skipActiveCheck has to be false.
##########
hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/service/DirectoryDeletingService.java:
##########
@@ -223,12 +227,153 @@ public BackgroundTaskResult call() {
getOzoneManager().getMetadataManager().getTableLock(
OmMetadataManagerImpl.DELETED_DIR_TABLE).writeLock().unlock();
}
+
+ try {
+ if (remainNum > 0) {
+ expandSnapshotDirectories(remainNum);
+ }
+ } catch (Exception e) {
+ LOG.error("Error while running deep clean on snapshots. Will " +
+ "retry at next run.", e);
+ }
}
// place holder by returning empty results of this call back.
return BackgroundTaskResult.EmptyTaskResult.newResult();
}
+ private void expandSnapshotDirectories(long remainNum) throws IOException {
+ OmSnapshotManager omSnapshotManager =
+ getOzoneManager().getOmSnapshotManager();
+ Table<String, SnapshotInfo> snapshotInfoTable =
+ getOzoneManager().getMetadataManager().getSnapshotInfoTable();
+
+ long dirNum = 0L;
+ long subDirNum = 0L;
+ long subFileNum = 0L;
+ int consumedSize = 0;
+ List<PurgePathRequest> purgePathRequestList = new ArrayList<>();
+ try (TableIterator<String, ? extends Table.KeyValue
+ <String, SnapshotInfo>> iterator = snapshotInfoTable.iterator()) {
+
+ while (remainNum > 0 && iterator.hasNext()) {
+ SnapshotInfo currSnapInfo = iterator.next().getValue();
+
+ // Expand deleted dirs only on active snapshot. Deleted Snapshots
+ // will be cleaned up by SnapshotDeletingService.
+ if (currSnapInfo.getSnapshotStatus() != SNAPSHOT_ACTIVE ||
+ currSnapInfo.getExpandedDeletedDir()) {
+ continue;
+ }
+
+ long volumeId = getOzoneManager().getMetadataManager()
+ .getVolumeId(currSnapInfo.getVolumeName());
+ // Get bucketInfo for the snapshot bucket to get bucket layout.
+ String dbBucketKey = getOzoneManager().getMetadataManager()
+ .getBucketKey(currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName());
+ OmBucketInfo bucketInfo = getOzoneManager().getMetadataManager()
+ .getBucketTable().get(dbBucketKey);
+
+ if (bucketInfo == null) {
+ throw new IllegalStateException("Bucket " + "/" +
+ currSnapInfo.getVolumeName() + "/" + currSnapInfo
+ .getBucketName() + " is not found. BucketInfo should not be " +
+ "null for snapshotted bucket. The OM is in unexpected state.");
+ }
+
+ String dbBucketKeyForDir = getOzoneManager().getMetadataManager()
+ .getBucketKey(Long.toString(volumeId),
+ Long.toString(bucketInfo.getObjectID())) + OM_KEY_PREFIX;
+
+ try (ReferenceCounted<IOmMetadataReader, SnapshotCache>
+ rcCurrOmSnapshot = omSnapshotManager.checkForSnapshot(
+ currSnapInfo.getVolumeName(),
+ currSnapInfo.getBucketName(),
+ getSnapshotPrefix(currSnapInfo.getName()),
+ true)) {
+
+ OmSnapshot currOmSnapshot = (OmSnapshot) rcCurrOmSnapshot.get();
+ Table<String, OmKeyInfo> snapDeletedDirTable =
+ currOmSnapshot.getMetadataManager().getDeletedDirTable();
+
+ if (snapDeletedDirTable.isEmpty()) {
+ // TODO: [SNAPSHOT] Update Snapshot state using
+ // SetSnapshotProperty from HDDS-7743 (YET TO BE MERGED)
+ continue;
+ }
+
+ List<Pair<String, OmKeyInfo>> allSubDirList
+ = new ArrayList<>((int) remainNum);
+
+ try (TableIterator<String, ? extends Table.KeyValue<String,
+ OmKeyInfo>> deletedIterator = snapDeletedDirTable.iterator()) {
+
+ long startTime = Time.monotonicNow();
+ deletedIterator.seek(dbBucketKeyForDir);
+
+ while (remainNum > 0 && deletedIterator.hasNext()) {
+ Table.KeyValue<String, OmKeyInfo> deletedDirInfo =
+ deletedIterator.next();
+ String deletedDirKey = deletedDirInfo.getKey();
+
+ // Exit if it is out of the bucket scope.
+ if (!deletedDirKey.startsWith(dbBucketKeyForDir)) {
+ break;
+ }
+
+ PurgePathRequest request = prepareDeleteDirRequest(
+ remainNum, deletedDirInfo.getValue(),
+ deletedDirInfo.getKey(), allSubDirList,
+ currOmSnapshot.getKeyManager());
+ if (isBufferLimitCrossed(ratisByteLimit, consumedSize,
+ request.getSerializedSize())) {
+ if (purgePathRequestList.size() != 0) {
+ // if message buffer reaches max limit,
+ // avoid sending further
+ remainNum = 0;
+ break;
+ }
+ // if directory itself is having a lot of keys / files,
+ // reduce capacity to minimum level
+ remainNum = MIN_ERR_LIMIT_PER_TASK;
+ request = prepareDeleteDirRequest(
+ remainNum, deletedDirInfo.getValue(),
+ deletedDirInfo.getKey(), allSubDirList,
+ currOmSnapshot.getKeyManager());
+ }
+
+ consumedSize += request.getSerializedSize();
+ purgePathRequestList.add(request);
+ remainNum = remainNum - request.getDeletedSubFilesCount();
+ remainNum = remainNum - request.getMarkDeletedSubDirsCount();
+ // Count up the purgeDeletedDir, subDirs and subFiles
+ if (request.hasDeletedDir() &&
+ !request.getDeletedDir().isEmpty()) {
+ dirNum++;
+ }
+ subDirNum += request.getMarkDeletedSubDirsCount();
+ subFileNum += request.getDeletedSubFilesCount();
+ }
+
+ optimizeDirDeletesAndSubmitRequest(
Review Comment:
Full diff and Dag based diff will start showing different outputs.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
