Add changes to make sure no two remoteDirectory cleanup for same shar…

…d is ongoing at same time. Signed-off-by: Harish Bhakuni <[email protected]>
opensearch-project · Mar 9, 2024 · b3c2ea3 · b3c2ea3
1 parent fcd4257
commit b3c2ea3
Show file tree

Hide file tree

Showing 2 changed files with 36 additions and 22 deletions.
diff --git a/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java b/server/src/main/java/org/opensearch/index/store/RemoteSegmentStoreDirectory.java
@@ -48,6 +48,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.ConcurrentModificationException;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -898,30 +899,38 @@ public static void cleanupAsync(
         ThreadPool threadpool,
         String remoteStoreRepoForIndex,
         String indexUUID,
-        ShardId shardId
+        ShardId shardId,
+        String indexShardIdentifier,
+        Set<String> ongoingDirectoryCleanups
     ) {
-        threadpool.executor(ThreadPool.Names.REMOTE_PURGE).execute(() -> {
+
+        if (ongoingDirectoryCleanups.add(indexShardIdentifier)) {
             try {
-                try (
-                    RemoteSegmentStoreDirectory remoteDirectory = (RemoteSegmentStoreDirectory) remoteDirectoryFactory.newDirectory(
-                        remoteStoreRepoForIndex,
-                        indexUUID,
-                        shardId
-                    )
-                ) {
-                    remoteDirectory.deleteStaleSegmentsAsync(
-                        0,
-                        ActionListener.wrap(
-                            r -> remoteDirectory.deleteIfEmpty(),
-                            e -> staticLogger.error("Failed to cleanup remote directory")
-                        ),
-                        ThreadPool.Names.SAME
-                    );
-                }
-            } catch (IOException e) {
-                staticLogger.error("Exception occurred while deleting directory", e);
+                threadpool.executor(ThreadPool.Names.REMOTE_PURGE).execute(() -> {
+                    try {
+                        try (
+                            RemoteSegmentStoreDirectory remoteDirectory = (RemoteSegmentStoreDirectory) remoteDirectoryFactory.newDirectory(
+                                remoteStoreRepoForIndex,
+                                indexUUID,
+                                shardId
+                            )
+                        ) {
+                            remoteDirectory.deleteStaleSegments(0); // cleanup stale segments in sync.
+                            remoteDirectory.deleteIfEmpty();
+                        }
+                    } catch (IOException e) {
+                        staticLogger.error("Exception occurred while deleting directory", e);
+                    } finally {
+                        ongoingDirectoryCleanups.remove(indexShardIdentifier);
+                    }
+                });
+            } catch (Exception e) {
+                ongoingDirectoryCleanups.remove(indexShardIdentifier);
+                throw e;
             }
-        });
+        } else {
+            throw new ConcurrentModificationException("cannot be executed since another cleanup ongoing");
+        }
     }
 
     /*

diff --git a/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java b/server/src/main/java/org/opensearch/repositories/blobstore/BlobStoreRepository.java
@@ -173,6 +173,7 @@
 import java.util.stream.LongStream;
 import java.util.stream.Stream;
 
+import static org.opensearch.common.util.concurrent.ConcurrentCollections.newConcurrentSet;
 import static org.opensearch.index.snapshots.blobstore.BlobStoreIndexShardSnapshot.FileInfo.canonicalName;
 import static org.opensearch.repositories.blobstore.ChecksumBlobStoreFormat.SNAPSHOT_ONLY_FORMAT_PARAMS;
 
@@ -315,6 +316,8 @@ public abstract class BlobStoreRepository extends AbstractLifecycleComponent imp
 
     private volatile RateLimiter remoteDownloadRateLimiter;
 
+    private final Set<String> ongoingRemoteDirectoryCleanups = newConcurrentSet();
+
     private final CounterMetric snapshotRateLimitingTimeInNanos = new CounterMetric();
 
     private final CounterMetric restoreRateLimitingTimeInNanos = new CounterMetric();
@@ -1142,7 +1145,9 @@ protected void releaseRemoteStoreLockAndCleanup(
                 threadPool,
                 remoteStoreRepoForIndex,
                 indexUUID,
-                new ShardId(Index.UNKNOWN_INDEX_NAME, indexUUID, Integer.parseInt(shardId))
+                new ShardId(Index.UNKNOWN_INDEX_NAME, indexUUID, Integer.parseInt(shardId)),
+                indexUUID + shardId,
+                ongoingRemoteDirectoryCleanups
             );
         }
     }