diff --git a/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java b/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java index 87874bd45000c..f97f34670f19a 100644 --- a/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java +++ b/server/src/main/java/org/elasticsearch/env/NodeEnvironment.java @@ -605,14 +605,6 @@ protected void closeInternal() { }; } - /** - * A functional interface that people can use to reference {@link #shardLock(ShardId, long)} - */ - @FunctionalInterface - public interface ShardLocker { - ShardLock lock(ShardId shardId, long lockTimeoutMS) throws ShardLockObtainFailedException; - } - /** * Returns all currently lock shards. * diff --git a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java index e854584b150d8..695d2d1d4adac 100644 --- a/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java +++ b/server/src/main/java/org/elasticsearch/gateway/TransportNodesListGatewayStartedShards.java @@ -41,7 +41,10 @@ import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.env.NodeEnvironment; +import org.elasticsearch.env.ShardLock; +import org.elasticsearch.env.ShardLockObtainFailedException; import org.elasticsearch.index.IndexSettings; +import org.elasticsearch.index.shard.IndexShard; import org.elasticsearch.index.shard.ShardId; import org.elasticsearch.index.shard.ShardPath; import org.elasticsearch.index.shard.ShardStateMetaData; @@ -52,6 +55,7 @@ import java.io.IOException; import java.util.List; +import java.util.concurrent.TimeUnit; /** * This transport action is used to fetch the shard version from each node during primary allocation in {@link GatewayAllocator}. @@ -116,9 +120,25 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) { try { final ShardId shardId = request.getShardId(); logger.trace("{} loading local shard state info", shardId); - ShardStateMetaData shardStateMetaData = ShardStateMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY, - nodeEnv.availableShardPaths(request.shardId)); - if (shardStateMetaData != null) { + + final IndexShard indexShard = indicesService.getShardOrNull(shardId); + if (indexShard != null) { + final ShardStateMetaData shardStateMetaData = indexShard.getShardStateMetaData(); + final String allocationId = shardStateMetaData.allocationId != null ? + shardStateMetaData.allocationId.getId() : null; + logger.trace("{} shard state info found: [{}]", shardId, shardStateMetaData); + return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary); + } + + try (ShardLock shardLock = nodeEnv.shardLock(shardId, TimeUnit.SECONDS.toMillis(5))) { + final ShardStateMetaData shardStateMetaData + = ShardStateMetaData.FORMAT.loadLatestState(logger, NamedXContentRegistry.EMPTY, nodeEnv.availableShardPaths(shardId)); + + if (shardStateMetaData == null) { + logger.trace("{} no local shard info found", shardId); + return new NodeGatewayStartedShards(clusterService.localNode(), null, false); + } + IndexMetaData metaData = clusterService.state().metaData().index(shardId.getIndex()); if (metaData == null) { // we may send this requests while processing the cluster state that recovered the index @@ -133,38 +153,38 @@ protected NodeGatewayStartedShards nodeOperation(NodeRequest request) { throw e; } - if (indicesService.getShardOrNull(shardId) == null) { - // we don't have an open shard on the store, validate the files on disk are openable - ShardPath shardPath = null; - try { - IndexSettings indexSettings = new IndexSettings(metaData, settings); - shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings); - if (shardPath == null) { - throw new IllegalStateException(shardId + " no shard path found"); - } - Store.tryOpenIndex(shardPath.resolveIndex(), shardId, nodeEnv::shardLock, logger); - } catch (Exception exception) { - final ShardPath finalShardPath = shardPath; - logger.trace(() -> new ParameterizedMessage( - "{} can't open index for shard [{}] in path [{}]", - shardId, - shardStateMetaData, - (finalShardPath != null) ? finalShardPath.resolveIndex() : ""), - exception); - String allocationId = shardStateMetaData.allocationId != null ? - shardStateMetaData.allocationId.getId() : null; - return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary, - exception); + // we don't have an open shard on the store, validate the files on disk are openable + ShardPath shardPath = null; + try { + IndexSettings indexSettings = new IndexSettings(metaData, settings); + shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings); + if (shardPath == null) { + throw new IllegalStateException(shardId + " no shard path found"); } + Store.tryOpenIndex(shardPath.resolveIndex(), shardId, shardLock, logger); + } catch (Exception exception) { + final ShardPath finalShardPath = shardPath; + logger.trace(() -> new ParameterizedMessage( + "{} can't open index for shard [{}] in path [{}]", + shardId, + shardStateMetaData, + (finalShardPath != null) ? finalShardPath.resolveIndex() : ""), + exception); + String allocationId = shardStateMetaData.allocationId != null ? + shardStateMetaData.allocationId.getId() : null; + return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary, + exception); } logger.debug("{} shard state info found: [{}]", shardId, shardStateMetaData); String allocationId = shardStateMetaData.allocationId != null ? shardStateMetaData.allocationId.getId() : null; return new NodeGatewayStartedShards(clusterService.localNode(), allocationId, shardStateMetaData.primary); + + } catch (ShardLockObtainFailedException e) { + return new NodeGatewayStartedShards(clusterService.localNode(), null, false, e); } - logger.trace("{} no local shard info found", shardId); - return new NodeGatewayStartedShards(clusterService.localNode(), null, false); + } catch (Exception e) { throw new ElasticsearchException("failed to load started shards", e); } diff --git a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java index 60392ab7990df..5d346d544c768 100644 --- a/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java +++ b/server/src/main/java/org/elasticsearch/index/shard/IndexShard.java @@ -2065,6 +2065,11 @@ public void startRecovery(RecoveryState recoveryState, PeerRecoveryTargetService } } + public ShardStateMetaData getShardStateMetaData() { + final ShardRouting shardRouting = this.shardRouting; + return new ShardStateMetaData(shardRouting.primary(), indexSettings.getUUID(), shardRouting.allocationId()); + } + /** * Returns whether the shard is in primary mode, i.e., in charge of replicating changes (see {@link ReplicationTracker}). */ diff --git a/server/src/main/java/org/elasticsearch/index/store/Store.java b/server/src/main/java/org/elasticsearch/index/store/Store.java index ccaae9d5f79df..5a46e1b9cbdb9 100644 --- a/server/src/main/java/org/elasticsearch/index/store/Store.java +++ b/server/src/main/java/org/elasticsearch/index/store/Store.java @@ -72,9 +72,7 @@ import org.elasticsearch.common.util.concurrent.RefCounted; import org.elasticsearch.common.util.iterable.Iterables; import org.elasticsearch.core.internal.io.IOUtils; -import org.elasticsearch.env.NodeEnvironment; import org.elasticsearch.env.ShardLock; -import org.elasticsearch.env.ShardLockObtainFailedException; import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.engine.CombinedDeletionPolicy; import org.elasticsearch.index.engine.Engine; @@ -241,7 +239,6 @@ final void ensureOpen() { * Note that this method requires the caller verify it has the right to access the store and * no concurrent file changes are happening. If in doubt, you probably want to use one of the following: * - * {@link #readMetadataSnapshot(Path, ShardId, NodeEnvironment.ShardLocker, Logger)} to read a meta data while locking * {@link IndexShard#snapshotStoreMetadata()} to safely read from an existing shard * {@link IndexShard#acquireLastIndexCommit(boolean)} to get an {@link IndexCommit} which is safe to use but has to be freed * @param commit the index commit to read the snapshot from or null if the latest snapshot should be read from the @@ -265,7 +262,6 @@ public MetadataSnapshot getMetadata(IndexCommit commit) throws IOException { * Note that this method requires the caller verify it has the right to access the store and * no concurrent file changes are happening. If in doubt, you probably want to use one of the following: * - * {@link #readMetadataSnapshot(Path, ShardId, NodeEnvironment.ShardLocker, Logger)} to read a meta data while locking * {@link IndexShard#snapshotStoreMetadata()} to safely read from an existing shard * {@link IndexShard#acquireLastIndexCommit(boolean)} to get an {@link IndexCommit} which is safe to use but has to be freed * @@ -456,18 +452,16 @@ private void closeInternal() { * * @throws IOException if the index we try to read is corrupted */ - public static MetadataSnapshot readMetadataSnapshot(Path indexLocation, ShardId shardId, NodeEnvironment.ShardLocker shardLocker, - Logger logger) throws IOException { - try (ShardLock lock = shardLocker.lock(shardId, TimeUnit.SECONDS.toMillis(5)); - Directory dir = new SimpleFSDirectory(indexLocation)) { + public static MetadataSnapshot readMetadataSnapshot(Path indexLocation, ShardId shardId, Logger logger, ShardLock shardLock) + throws IOException { + assert shardLock.isOpen(); + try (Directory dir = new SimpleFSDirectory(indexLocation)) { failIfCorrupted(dir, shardId); return new MetadataSnapshot(null, dir, logger); } catch (IndexNotFoundException ex) { // that's fine - happens all the time no need to log } catch (FileNotFoundException | NoSuchFileException ex) { logger.info("Failed to open / find files while reading metadata snapshot"); - } catch (ShardLockObtainFailedException ex) { - logger.info(() -> new ParameterizedMessage("{}: failed to obtain shard lock", shardId), ex); } return MetadataSnapshot.EMPTY; } @@ -477,9 +471,9 @@ public static MetadataSnapshot readMetadataSnapshot(Path indexLocation, ShardId * can be successfully opened. This includes reading the segment infos and possible * corruption markers. */ - public static boolean canOpenIndex(Logger logger, Path indexLocation, ShardId shardId, NodeEnvironment.ShardLocker shardLocker) throws IOException { + public static boolean canOpenIndex(Logger logger, Path indexLocation, ShardId shardId, ShardLock shardLock) { try { - tryOpenIndex(indexLocation, shardId, shardLocker, logger); + tryOpenIndex(indexLocation, shardId, shardLock, logger); } catch (Exception ex) { logger.trace(() -> new ParameterizedMessage("Can't open index for path [{}]", indexLocation), ex); return false; @@ -492,9 +486,9 @@ public static boolean canOpenIndex(Logger logger, Path indexLocation, ShardId sh * segment infos and possible corruption markers. If the index can not * be opened, an exception is thrown */ - public static void tryOpenIndex(Path indexLocation, ShardId shardId, NodeEnvironment.ShardLocker shardLocker, Logger logger) throws IOException, ShardLockObtainFailedException { - try (ShardLock lock = shardLocker.lock(shardId, TimeUnit.SECONDS.toMillis(5)); - Directory dir = new SimpleFSDirectory(indexLocation)) { + public static void tryOpenIndex(Path indexLocation, ShardId shardId, ShardLock shardLock, Logger logger) throws IOException { + assert shardLock.isOpen(); + try (Directory dir = new SimpleFSDirectory(indexLocation)) { failIfCorrupted(dir, shardId); SegmentInfos segInfo = Lucene.readSegmentInfos(dir); logger.trace("{} loaded segment info [{}]", shardId, segInfo); diff --git a/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java b/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java index 404a19b0ab354..fa5ce6e6533da 100644 --- a/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java +++ b/server/src/main/java/org/elasticsearch/indices/store/TransportNodesListShardStoreMetaData.java @@ -19,6 +19,7 @@ package org.elasticsearch.indices.store; +import org.apache.logging.log4j.message.ParameterizedMessage; import org.elasticsearch.ElasticsearchException; import org.elasticsearch.action.ActionListener; import org.elasticsearch.action.FailedNodeException; @@ -41,6 +42,8 @@ import org.elasticsearch.common.unit.TimeValue; import org.elasticsearch.common.xcontent.NamedXContentRegistry; import org.elasticsearch.env.NodeEnvironment; +import org.elasticsearch.env.ShardLock; +import org.elasticsearch.env.ShardLockObtainFailedException; import org.elasticsearch.gateway.AsyncShardFetch; import org.elasticsearch.index.IndexService; import org.elasticsearch.index.IndexSettings; @@ -138,17 +141,23 @@ private StoreFilesMetaData listStoreMetaData(ShardId shardId) throws IOException logger.trace("{} node doesn't have meta data for the requests index, responding with empty", shardId); return new StoreFilesMetaData(shardId, Store.MetadataSnapshot.EMPTY); } - final IndexSettings indexSettings = indexService != null ? indexService.getIndexSettings() : new IndexSettings(metaData, settings); - final ShardPath shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings); - if (shardPath == null) { - return new StoreFilesMetaData(shardId, Store.MetadataSnapshot.EMPTY); - } + final IndexSettings indexSettings + = indexService != null ? indexService.getIndexSettings() : new IndexSettings(metaData, settings); + // note that this may fail if it can't get access to the shard lock. Since we check above there is an active shard, this means: - // 1) a shard is being constructed, which means the master will not use a copy of this replica - // 2) A shard is shutting down and has not cleared it's content within lock timeout. In this case the master may not + // 1) a shard is being constructed, which means the master will not use a copy of this replica. + // 2) a shard is shutting down and has not cleared its content within the lock timeout. In this case the master may not // reuse local resources. - return new StoreFilesMetaData(shardId, Store.readMetadataSnapshot(shardPath.resolveIndex(), shardId, - nodeEnv::shardLock, logger)); + try (ShardLock shardLock = nodeEnv.shardLock(shardId, TimeUnit.SECONDS.toMillis(5))) { + final ShardPath shardPath = ShardPath.loadShardPath(logger, nodeEnv, shardId, indexSettings); + if (shardPath != null) { + return new StoreFilesMetaData(shardId, + Store.readMetadataSnapshot(shardPath.resolveIndex(), shardId, logger, shardLock)); + } + } catch (ShardLockObtainFailedException ex) { + logger.info(() -> new ParameterizedMessage("{}: failed to obtain shard lock", shardId), ex); + } + return new StoreFilesMetaData(shardId, Store.MetadataSnapshot.EMPTY); } finally { TimeValue took = new TimeValue(System.nanoTime() - startTimeNS, TimeUnit.NANOSECONDS); if (exists) { diff --git a/server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java b/server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java index 154d702e7fb77..845a9e7dda144 100644 --- a/server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java +++ b/server/src/test/java/org/elasticsearch/gateway/RecoveryFromGatewayIT.java @@ -34,6 +34,7 @@ import org.elasticsearch.env.Environment; import org.elasticsearch.env.NodeEnvironment; import org.elasticsearch.index.Index; +import org.elasticsearch.index.IndexNotFoundException; import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.engine.Engine; import org.elasticsearch.index.query.QueryBuilders; @@ -46,8 +47,10 @@ import org.elasticsearch.test.ESIntegTestCase.Scope; import org.elasticsearch.test.InternalTestCluster; import org.elasticsearch.test.InternalTestCluster.RestartCallback; +import org.elasticsearch.test.junit.annotations.TestLogging; import org.elasticsearch.test.store.MockFSIndexStore; +import java.io.File; import java.nio.file.DirectoryStream; import java.nio.file.Files; import java.nio.file.Path; @@ -58,6 +61,8 @@ import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.stream.IntStream; import static org.elasticsearch.cluster.metadata.IndexMetaData.SETTING_NUMBER_OF_REPLICAS; @@ -567,4 +572,70 @@ public Settings onNodeStopped(String nodeName) throws Exception { // start another node so cluster consistency checks won't time out due to the lack of state internalCluster().startNode(); } + + public void testLoadLatestStateWhileClosingShardDoesNotResurrectMetadataDirectory() throws Exception { + + // This test pertains to a race condition in which deleting a shard concurrently with a TransportNodesListGatewayStartedShards + // request could resurrect the shard's metadata folder after it was deleted. Here we try and recreate the race, but it is quite + // delicate so this test does not always fail. Experimentation showed that setting the thread counts as below would yield a failure + // after a reasonable number of iterations: running repeatedly with -Dtests.iters=1000 saw 6 failures out of 10 runs. + // + // NB this experiment was run with + // @TestLogging("org.elasticsearch.env.NodeEnvironment:TRACE,org.elasticsearch.gateway.MetaDataStateFormat:TRACE," + + // "org.elasticsearch.gateway.TransportNodesListGatewayStartedShards:TRACE,org.elasticsearch.index.shard.IndexShard:TRACE") + // but with less verbose logging the failures seem rarer. + + final String nodeName = internalCluster().startNode(); + DiscoveryNode node = internalCluster().getInstance(ClusterService.class, nodeName).localNode(); + + assertAcked(prepareCreate("test").setSettings(Settings.builder() + .put(SETTING_NUMBER_OF_SHARDS, 1).put(SETTING_NUMBER_OF_REPLICAS, 0))); + final ShardId shardId = new ShardId(resolveIndex("test"), 0); + + final int listingThreadCount = 2; + final int deletingThreadCount = 2; + + final CountDownLatch countDownLatch = new CountDownLatch(listingThreadCount + deletingThreadCount); + + Thread threads[] = new Thread[listingThreadCount + deletingThreadCount]; + for (int threadIndex = 0; threadIndex < listingThreadCount + deletingThreadCount; threadIndex++) { + final boolean isListingThread = threadIndex < listingThreadCount; + threads[threadIndex] = new Thread(() -> { + try { + countDownLatch.countDown(); + countDownLatch.await(); + + if (isListingThread) { + internalCluster().getInstance(TransportNodesListGatewayStartedShards.class) + .execute(new TransportNodesListGatewayStartedShards.Request(shardId, new DiscoveryNode[]{node})) + .get(); + } else { + assertAcked(client().admin().indices().prepareDelete("test")); + } + } catch (InterruptedException | ExecutionException | IndexNotFoundException ignored) { + // don't care if this fails + } + }, (isListingThread ? "Listing" : "Deleting") + "[" + threadIndex + "]"); + } + + NodeEnvironment nodeEnvironment = internalCluster().getInstance(NodeEnvironment.class, nodeName); + + boolean directoryExists = false; + for (Path path : nodeEnvironment.availableShardPaths(shardId)) { + directoryExists = directoryExists || Files.exists(path); + } + assertTrue(directoryExists); + + for (final Thread thread : threads) { + thread.start(); + } + + for (final Thread thread : threads) { + thread.join(); + } + + for (Path path : nodeEnvironment.availableShardPaths(shardId)) { + assertFalse(path + " should not exist", Files.exists(path)); + } + } } diff --git a/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java b/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java index 027b595ee761f..b1b804853d0f0 100644 --- a/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java +++ b/server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java @@ -234,8 +234,7 @@ public void testFailShard() throws Exception { assertEquals(shardStateMetaData, getShardStateMetadata(shard)); // but index can't be opened for a failed shard assertThat("store index should be corrupted", Store.canOpenIndex(logger, shardPath.resolveIndex(), shard.shardId(), - (shardId, lockTimeoutMS) -> new DummyShardLock(shardId)), - equalTo(false)); + new DummyShardLock(shard.shardId())), equalTo(false)); } ShardStateMetaData getShardStateMetadata(IndexShard shard) { diff --git a/server/src/test/java/org/elasticsearch/index/store/StoreTests.java b/server/src/test/java/org/elasticsearch/index/store/StoreTests.java index 9352d978e6e46..63eaa88af9705 100644 --- a/server/src/test/java/org/elasticsearch/index/store/StoreTests.java +++ b/server/src/test/java/org/elasticsearch/index/store/StoreTests.java @@ -948,14 +948,14 @@ public void testCanOpenIndex() throws IOException { IndexWriterConfig iwc = newIndexWriterConfig(); Path tempDir = createTempDir(); final BaseDirectoryWrapper dir = newFSDirectory(tempDir); - assertFalse(Store.canOpenIndex(logger, tempDir, shardId, (id, l) -> new DummyShardLock(id))); + assertFalse(Store.canOpenIndex(logger, tempDir, shardId, new DummyShardLock(shardId))); IndexWriter writer = new IndexWriter(dir, iwc); Document doc = new Document(); doc.add(new StringField("id", "1", random().nextBoolean() ? Field.Store.YES : Field.Store.NO)); writer.addDocument(doc); writer.commit(); writer.close(); - assertTrue(Store.canOpenIndex(logger, tempDir, shardId, (id, l) -> new DummyShardLock(id))); + assertTrue(Store.canOpenIndex(logger, tempDir, shardId, new DummyShardLock(shardId))); DirectoryService directoryService = new DirectoryService(shardId, INDEX_SETTINGS) { @@ -966,7 +966,7 @@ public Directory newDirectory() throws IOException { }; Store store = new Store(shardId, INDEX_SETTINGS, directoryService, new DummyShardLock(shardId)); store.markStoreCorrupted(new CorruptIndexException("foo", "bar")); - assertFalse(Store.canOpenIndex(logger, tempDir, shardId, (id, l) -> new DummyShardLock(id))); + assertFalse(Store.canOpenIndex(logger, tempDir, shardId, new DummyShardLock(shardId))); store.close(); }