[CORE] Delete shard content under lock

Once we delete the the index on a node we are closing all resources
and subsequently need to delete all shards contents from disk. Yet
this happens today under a lock (the shard lock) that needs to be
acquried in order to execute any operation on the shards data
path. We try to delete all the index meta-data once we acquired
all the shard lock but this operation can run into a timeout which causes
the index to remain on disk. Further, all shard data will be left on
disk if the timeout is reached.

This commit removes all the shards data just before the shard lock
is release as the last operation on a shard that belongs to a deleted
index.

src/main/java/org/elasticsearch/action/admin/cluster/stats/TransportClusterStatsAction.java

@@ -115,7 +115,7 @@ protected ClusterStatsNodeResponse nodeOperation(ClusterStatsNodeRequest nodeReq
         NodeInfo nodeInfo = nodeService.info(false, true, false, true, false, false, true, false, true);
         NodeStats nodeStats = nodeService.stats(CommonStatsFlags.NONE, false, true, true, false, false, true, false, false, false);
         List<ShardStats> shardsStats = new ArrayList<>();
-        for (IndexService indexService : indicesService.indices().values()) {
+        for (IndexService indexService : indicesService) {
             for (IndexShard indexShard : indexService) {
                 if (indexShard.routingEntry() != null && indexShard.routingEntry().active()) {
                     // only report on fully started shards

src/main/java/org/elasticsearch/action/admin/indices/recovery/TransportRecoveryAction.java

@@ -153,7 +153,7 @@ protected ShardRecoveryResponse shardOperation(ShardRecoveryRequest request) thr
 
         if (state == null) {
             IndexShardGatewayService gatewayService =
-                    indexService.shardInjector(request.shardId().id()).getInstance(IndexShardGatewayService.class);
+                    indexService.shardInjectorSafe(request.shardId().id()).getInstance(IndexShardGatewayService.class);
             state = gatewayService.recoveryState();
         }
 

src/main/java/org/elasticsearch/common/component/LifecycleComponent.java

@@ -20,11 +20,14 @@
 package org.elasticsearch.common.component;
 
 import org.elasticsearch.ElasticsearchException;
+import org.elasticsearch.common.lease.Releasable;
+
+import java.io.Closeable;
 
 /**
  *
  */
-public interface LifecycleComponent<T> extends CloseableComponent {
+public interface LifecycleComponent<T> extends Releasable {
 
     Lifecycle.State lifecycleState();
 

src/main/java/org/elasticsearch/common/inject/Injectors.java

@@ -230,11 +230,6 @@ public static <T> Class<?> getKeyType(Key<?> key) {
         return keyType;
     }
 
-
-    public static void close(Injector injector) {
-
-    }
-
     public static void cleanCaches(Injector injector) {
         ((InjectorImpl) injector).clearCache();
         if (injector.getParent() != null) {

src/main/java/org/elasticsearch/env/NodeEnvironment.java

@@ -182,15 +182,38 @@ public void deleteShardDirectorySafe(ShardId shardId, @IndexSettings Settings in
         assert indexSettings != ImmutableSettings.EMPTY;
         final Path[] paths = shardPaths(shardId);
         logger.trace("deleting shard {} directory, paths: [{}]", shardId, paths);
-        try (Closeable lock = shardLock(shardId)) {
-            IOUtils.rm(paths);
-            if (hasCustomDataPath(indexSettings)) {
-                Path customLocation = resolveCustomLocation(indexSettings, shardId);
-                logger.trace("deleting custom shard {} directory [{}]", shardId, customLocation);
-                IOUtils.rm(customLocation);
-            }
-            logger.trace("deleted shard {} directory, paths: [{}]", shardId, paths);
-            assert FileSystemUtils.exists(paths) == false;
+        try (ShardLock lock = shardLock(shardId)) {
+            deleteShardDirectoryUnderLock(lock, indexSettings);
+        }
+    }
+
+    /**
+     * Deletes a shard data directory. Note: this method assumes that the shard lock is acquired
+     *
+     * @param lock the shards lock
+     * @throws IOException if an IOException occurs
+     */
+    public void deleteShardDirectoryUnderLock(ShardLock lock, @IndexSettings Settings indexSettings) throws IOException {
+        assert indexSettings != ImmutableSettings.EMPTY;
+        final ShardId shardId = lock.getShardId();
+        assert isShardLocked(shardId) : "shard " + shardId + " is not locked";
+        final Path[] paths = shardPaths(shardId);
+        IOUtils.rm(paths);
+        if (hasCustomDataPath(indexSettings)) {
+            Path customLocation = resolveCustomLocation(indexSettings, shardId);
+            logger.trace("deleting custom shard {} directory [{}]", shardId, customLocation);
+            IOUtils.rm(customLocation);
+        }
+        logger.trace("deleted shard {} directory, paths: [{}]", shardId, paths);
+        assert FileSystemUtils.exists(paths) == false;
+    }
+
+    private boolean isShardLocked(ShardId id) {
+        try {
+            shardLock(id, 0).close();
+            return false;
+        } catch (IOException ex) {
+            return true;
         }
     }
 

src/main/java/org/elasticsearch/gateway/GatewayMetaState.java

@@ -192,10 +192,6 @@ public MetaData loadMetaState() throws Exception {
         return loadState();
     }
 
-    public boolean isDangling(String index) {
-        return danglingIndices.containsKey(index);
-    }
-
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
         if (event.state().blocks().disableStatePersistence()) {