Internal: dangling indices import ignores aliases

Dangling indices are indexes found on disk which are not part of the cluster state. By default, we don't delete them but rather import them into the cluster state in order to not accidentally delete data and also allow for the ease of copying index data folders from one cluster to another. Currently, the import logic doesn't check for existing aliases of the same name as the imported dangling index, causing both an index and an alias with the same name.

 This commit add a protection against this. Note that the index is still kept as dangling and is only deleted from disk after `gateway.local.dangling_timeout` has passed (2 hours).

 We also change the log message indicating deletion of dangling indices to a `WARN` level.

 Closes #8059

src/main/java/org/elasticsearch/gateway/local/state/meta/LocalAllocateDangledIndices.java

@@ -128,6 +128,11 @@ public class LocalAllocateDangledIndices extends AbstractComponent {
                         if (currentState.metaData().hasIndex(indexMetaData.index())) {
                             continue;
                         }
+                        if (currentState.metaData().aliases().containsKey(indexMetaData.index())) {
+                            logger.warn("ignoring dangled index [{}] on node [{}] due to an existing alias with the same name",
+                                    indexMetaData.index(), request.fromNode);
+                            continue;
+                        }
                         importNeeded = true;
                         metaData.put(indexMetaData, false);
                         blocks.addBlocks(indexMetaData);

src/main/java/org/elasticsearch/gateway/local/state/meta/LocalGatewayMetaState.java

@@ -478,7 +478,7 @@ public class LocalGatewayMetaState extends AbstractComponent implements ClusterS
             return;
         }
 
-        logger.info("found old metadata state, loading metadata from [{}] and converting to new metadata location and strucutre...", metaDataFile.getAbsolutePath());
+        logger.info("found old metadata state, loading metadata from [{}] and converting to new metadata location and structure...", metaDataFile.getAbsolutePath());
 
         writeGlobalState("upgrade", MetaData.builder(metaData).version(version).build());
         for (IndexMetaData indexMetaData : metaData) {
@@ -532,7 +532,7 @@ public class LocalGatewayMetaState extends AbstractComponent implements ClusterS
                 if (remove == null) {
                     return;
                 }
-                logger.info("[{}] deleting dangling index", index);
+                logger.warn("[{}] deleting dangling index", index);
                 FileSystemUtils.deleteRecursively(nodeEnv.indexLocations(new Index(index)));
             }
         }

src/test/java/org/elasticsearch/gateway/local/LocalGatewayIndexStateTests.java

@@ -273,12 +273,73 @@ public class LocalGatewayIndexStateTests extends ElasticsearchIntegrationTest {
         }
     }
 
+    @Test
+    public void testDanglingIndicesConflictWithAlias() throws Exception {
+        Settings settings = settingsBuilder().put("gateway.type", "local").build();
+        logger.info("--> starting two nodes");
+        internalCluster().startNodesAsync(2, settings).get();
+
+        logger.info("--> indexing a simple document");
+        client().prepareIndex("test", "type1", "1").setSource("field1", "value1").setRefresh(true).execute().actionGet();
+
+        logger.info("--> waiting for green status");
+        ensureGreen();
+
+        logger.info("--> verify 1 doc in the index");
+        for (int i = 0; i < 10; i++) {
+            assertHitCount(client().prepareSearch().setQuery(matchAllQuery()).get(), 1l);
+        }
+        assertThat(client().prepareGet("test", "type1", "1").execute().actionGet().isExists(), equalTo(true));
+
+        internalCluster().stopRandomNonMasterNode();
+
+
+        logger.info("--> deleting index");
+        assertAcked(client().admin().indices().prepareDelete("test"));
+
+        index("test2", "type1", "2", "{}");
+
+        logger.info("--> creating index with an alias");
+        assertAcked(client().admin().indices().prepareAliases().addAlias("test2", "test"));
+
+        logger.info("--> starting node back up");
+        internalCluster().startNode(settings);
+
+        ensureGreen();
+
+        // make sure that any other events were processed
+        assertFalse(client().admin().cluster().prepareHealth().setWaitForRelocatingShards(0).setWaitForEvents(Priority.LANGUID).get().isTimedOut());
+
+        logger.info("--> verify we read the right thing through alias");
+        assertThat(client().prepareGet("test", "type1", "2").execute().actionGet().isExists(), equalTo(true));
+
+        logger.info("--> deleting alias");
+        assertAcked(client().admin().indices().prepareAliases().removeAlias("test2", "test"));
+
+        logger.info("--> waiting for dangling index to be imported");
+
+        assertBusy(new Runnable() {
+            @Override
+            public void run() {
+                assertTrue(client().admin().indices().prepareExists("test").execute().actionGet().isExists());
+            }
+        });
+
+        ensureGreen();
+
+        logger.info("--> verifying dangling index contains doc");
+
+        assertThat(client().prepareGet("test", "type1", "1").execute().actionGet().isExists(), equalTo(true));
+
+    }
+
     @Test
     public void testDanglingIndicesAutoImportYes() throws Exception {
         Settings settings = settingsBuilder()
                 .put("gateway.type", "local").put("gateway.local.auto_import_dangled", "yes")
                 .build();
         logger.info("--> starting two nodes");
+
         final String node_1 = internalCluster().startNode(settings);
         internalCluster().startNode(settings);