Fix RepoCleanup not Removed on Master-Failover

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java

@@ -93,7 +93,7 @@ public TransportCleanupRepositoryAction(TransportService transportService, Clust
         clusterService.addStateApplier(event -> {
             if (event.localNodeMaster() && event.previousState().nodes().isLocalNodeElectedMaster() == false) {
                 final RepositoryCleanupInProgress repositoryCleanupInProgress = event.state().custom(RepositoryCleanupInProgress.TYPE);
-                if (repositoryCleanupInProgress == null || repositoryCleanupInProgress.cleanupInProgress() == false) {
+                if (repositoryCleanupInProgress == null || repositoryCleanupInProgress.hasCleanupInProgress() == false) {
                     return;
                 }
                 clusterService.submitStateUpdateTask("clean up repository cleanup task after master failover",
@@ -122,7 +122,7 @@ private static ClusterState removeInProgressCleanup(final ClusterState currentSt
         RepositoryCleanupInProgress cleanupInProgress = currentState.custom(RepositoryCleanupInProgress.TYPE);
         if (cleanupInProgress != null) {
             boolean changed = false;
-            if (cleanupInProgress.cleanupInProgress() == false) {
+            if (cleanupInProgress.hasCleanupInProgress()) {
                 cleanupInProgress = new RepositoryCleanupInProgress();
                 changed = true;
             }
@@ -172,10 +172,13 @@ private void cleanupRepo(String repositoryName, ActionListener<RepositoryCleanup
         logger.info("Running cleanup operations on repository [{}][{}]", repositoryName, repositoryStateId);
         clusterService.submitStateUpdateTask("cleanup repository [" + repositoryName + "][" + repositoryStateId + ']',
             new ClusterStateUpdateTask() {
+
+                private boolean startedCleanup = false;
+
                 @Override
                 public ClusterState execute(ClusterState currentState) {
                     final RepositoryCleanupInProgress repositoryCleanupInProgress = currentState.custom(RepositoryCleanupInProgress.TYPE);
-                    if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.cleanupInProgress() == false) {
+                    if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.hasCleanupInProgress()) {
                         throw new IllegalStateException(
                             "Cannot cleanup [" + repositoryName + "] - a repository cleanup is already in-progress");
                     }
@@ -199,6 +202,7 @@ public void onFailure(String source, Exception e) {
 
                 @Override
                 public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
+                    startedCleanup = true;
                     logger.debug("Initialized repository cleanup in cluster state for [{}][{}]", repositoryName, repositoryStateId);
                     threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(ActionRunnable.wrap(listener,
                         l -> blobStoreRepository.cleanup(
@@ -215,6 +219,11 @@ private void after(@Nullable Exception failure, @Nullable RepositoryCleanupResul
                             "Failed to finish repository cleanup operations on [{}][{}]", repositoryName, repositoryStateId), failure);
                     }
                     assert failure != null || result != null;
+                    if (startedCleanup == false) {
+                        logger.debug("No cleanup task to remove from cluster state because we failed to start one", failure);
+                        listener.onFailure(failure);
+                        return;
+                    }
                     clusterService.submitStateUpdateTask(
                         "remove repository cleanup task [" + repositoryName + "][" + repositoryStateId + ']',
                         new ClusterStateUpdateTask() {

server/src/main/java/org/elasticsearch/cluster/RepositoryCleanupInProgress.java

@@ -51,9 +51,13 @@ public static Entry startedEntry(String repository, long repositoryStateId) {
         return new Entry(repository, repositoryStateId);
     }
 
-    public boolean cleanupInProgress() {
+    public boolean hasCleanupInProgress() {
         // TODO: Should we allow parallelism across repositories here maybe?
-        return entries.isEmpty();
+        return entries.isEmpty() == false;
+    }
+
+    public List<Entry> entries() {
+        return List.copyOf(entries);
     }
 
     @Override
@@ -106,6 +110,10 @@ public Entry(String repository, long repositoryStateId) {
             this.repositoryStateId = repositoryStateId;
         }
 
+        public String repository() {
+            return repository;
+        }
+
         @Override
         public void writeTo(StreamOutput out) throws IOException {
             out.writeString(repository);

server/src/main/java/org/elasticsearch/repositories/RepositoriesService.java

@@ -435,8 +435,7 @@ private static void validate(final String repositoryName) {
         }
     }
 
-
-    private void ensureRepositoryNotInUse(ClusterState clusterState, String repository) {
+    private static void ensureRepositoryNotInUse(ClusterState clusterState, String repository) {
         if (SnapshotsService.isRepositoryInUse(clusterState, repository) || RestoreService.isRepositoryInUse(clusterState, repository)) {
             throw new IllegalStateException("trying to modify or unregister repository that is currently used ");
         }

server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java

@@ -275,7 +275,7 @@ public ClusterState execute(ClusterState currentState) {
                         "cannot snapshot while a snapshot deletion is in-progress");
                 }
                 final RepositoryCleanupInProgress repositoryCleanupInProgress = currentState.custom(RepositoryCleanupInProgress.TYPE);
-                if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.cleanupInProgress() == false) {
+                if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.hasCleanupInProgress()) {
                     throw new ConcurrentSnapshotExecutionException(repositoryName, snapshotName,
                         "cannot snapshot while a repository cleanup is in-progress");
                 }
@@ -1185,7 +1185,7 @@ public ClusterState execute(ClusterState currentState) {
                         "cannot delete - another snapshot is currently being deleted");
                 }
                 final RepositoryCleanupInProgress repositoryCleanupInProgress = currentState.custom(RepositoryCleanupInProgress.TYPE);
-                if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.cleanupInProgress() == false) {
+                if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.hasCleanupInProgress()) {
                     throw new ConcurrentSnapshotExecutionException(snapshot.getRepository(), snapshot.getSnapshotId().getName(),
                         "cannot delete snapshot while a repository cleanup is in-progress");
                 }
@@ -1343,6 +1343,14 @@ public static boolean isRepositoryInUse(ClusterState clusterState, String reposi
                 }
             }
         }
+        final RepositoryCleanupInProgress repositoryCleanupInProgress = clusterState.custom(RepositoryCleanupInProgress.TYPE);
+        if (repositoryCleanupInProgress != null) {
+            for (RepositoryCleanupInProgress.Entry entry : repositoryCleanupInProgress.entries()) {
+                if (entry.repository().equals(repository)) {
+                    return true;
+                }
+            }
+        }
         return false;
     }
 

server/src/test/java/org/elasticsearch/repositories/blobstore/BlobStoreRepositoryCleanupIT.java

@@ -0,0 +1,110 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch 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.elasticsearch.repositories.blobstore;
+
+import org.elasticsearch.action.ActionRunnable;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.cluster.RepositoryCleanupInProgress;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeUnit;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.repositories.RepositoriesService;
+import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
+import org.elasticsearch.test.ESIntegTestCase;
+
+import java.io.ByteArrayInputStream;
+import java.util.concurrent.TimeUnit;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertAcked;
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertThrows;
+
+@ESIntegTestCase.ClusterScope(scope = ESIntegTestCase.Scope.TEST, numDataNodes = 0)
+public class BlobStoreRepositoryCleanupIT extends AbstractSnapshotIntegTestCase {
+
+    public void testMasterFailoverDuringCleanup() throws Exception {
+        startBlockedCleanup("test-repo");
+
+        logger.info("-->  stopping master node");
+        internalCluster().stopCurrentMasterNode();
+
+        logger.info("-->  wait for cleanup to finish and disappear from cluster state");
+        assertBusy(() -> {
+            RepositoryCleanupInProgress cleanupInProgress =
+                client().admin().cluster().prepareState().get().getState().custom(RepositoryCleanupInProgress.TYPE);
+            assertFalse(cleanupInProgress.hasCleanupInProgress());
+        }, 30, TimeUnit.SECONDS);
+    }
+
+    public void testRepeatCleanupsDontRemove() throws Exception {
+        final String masterNode = startBlockedCleanup("test-repo");
+
+        logger.info("-->  sending another cleanup");
+        assertThrows(client().admin().cluster().prepareCleanupRepository("test-repo").execute(), IllegalStateException.class);
+
+        logger.info("-->  ensure cleanup is still in progress");
+        final RepositoryCleanupInProgress cleanup =
+            client().admin().cluster().prepareState().get().getState().custom(RepositoryCleanupInProgress.TYPE);
+        assertTrue(cleanup.hasCleanupInProgress());
+
+        logger.info("-->  unblocking master node");
+        unblockNode("test-repo", masterNode);
+
+        logger.info("-->  wait for cleanup to finish and disappear from cluster state");
+        assertBusy(() -> {
+            RepositoryCleanupInProgress cleanupInProgress =
+                client().admin().cluster().prepareState().get().getState().custom(RepositoryCleanupInProgress.TYPE);
+            assertFalse(cleanupInProgress.hasCleanupInProgress());
+        }, 30, TimeUnit.SECONDS);
+    }
+
+    private String startBlockedCleanup(String repoName) throws Exception {
+        logger.info("-->  starting two master nodes and one data node");
+        internalCluster().startMasterOnlyNodes(2);
+        internalCluster().startDataOnlyNodes(1);
+
+        logger.info("-->  creating repository");
+        assertAcked(client().admin().cluster().preparePutRepository(repoName)
+            .setType("mock").setSettings(Settings.builder()
+                .put("location", randomRepoPath())
+                .put("compress", randomBoolean())
+                .put("chunk_size", randomIntBetween(100, 1000), ByteSizeUnit.BYTES)));
+
+        logger.info("-->  snapshot");
+        client().admin().cluster().prepareCreateSnapshot(repoName, "test-snap")
+            .setWaitForCompletion(true).get();
+
+        final RepositoriesService service = internalCluster().getInstance(RepositoriesService.class, internalCluster().getMasterName());
+        final BlobStoreRepository repository = (BlobStoreRepository) service.repository(repoName);
+
+        logger.info("--> creating a garbage data blob");
+        final PlainActionFuture<Void> garbageFuture = PlainActionFuture.newFuture();
+        repository.threadPool().generic().execute(ActionRunnable.run(garbageFuture, () -> repository.blobStore()
+            .blobContainer(repository.basePath()).writeBlob("snap-foo.dat", new ByteArrayInputStream(new byte[1]), 1, true)));
+        garbageFuture.get();
+
+        final String masterNode = blockMasterFromFinalizingSnapshotOnIndexFile(repoName);
+
+        logger.info("--> starting repository cleanup");
+        client().admin().cluster().prepareCleanupRepository(repoName).execute();
+
+        logger.info("--> waiting for block to kick in on " + masterNode);
+        waitForBlock(masterNode, repoName, TimeValue.timeValueSeconds(60));
+        return masterNode;
+    }
+}

x-pack/plugin/ilm/src/main/java/org/elasticsearch/xpack/slm/SnapshotRetentionTask.java

@@ -440,7 +440,7 @@ public static boolean okayToDeleteSnapshots(ClusterState state) {
 
         // Cannot delete while a repository is being cleaned
         final RepositoryCleanupInProgress repositoryCleanupInProgress = state.custom(RepositoryCleanupInProgress.TYPE);
-        if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.cleanupInProgress() == false) {
+        if (repositoryCleanupInProgress != null && repositoryCleanupInProgress.hasCleanupInProgress()) {
             return false;
         }