[STORE] Allow to get metadata from arbitrary commit points

Today we always use the latest commit point to return the metadata from
the store. This might cause problems for snapshot and restore since in
contrast to recovery it won't prevent concurrent flushes (lucene commits).
This can lead to all kinds of interesting effects if we are snapshotting
while flushing. This change uses the IndexCommit to open the metadata snapshot
from the store which is consistent with what we snapshot.

Closes #7376

src/main/java/org/elasticsearch/common/lucene/Lucene.java

@@ -99,6 +99,15 @@ public static SegmentInfos readSegmentInfos(Directory directory) throws IOExcept
         return sis;
     }
 
+    /**
+     * Reads the segments infos from the given commit, failing if it fails to load
+     */
+    public static SegmentInfos readSegmentInfos(IndexCommit commit, Directory directory) throws IOException {
+        final SegmentInfos sis = new SegmentInfos();
+        sis.read(directory, commit.getSegmentsFileName());
+        return sis;
+    }
+
     public static void checkSegmentInfoIntegrity(final Directory directory) throws IOException {
         new SegmentInfos.FindSegmentsFile(directory) {
 

src/main/java/org/elasticsearch/index/snapshots/blobstore/BlobStoreIndexShardRepository.java

@@ -427,7 +427,7 @@ public void snapshot(SnapshotIndexCommit snapshotIndexCommit) {
                 ArrayList<FileInfo> filesToSnapshot = newArrayList();
                 final Store.MetadataSnapshot metadata;
                 try {
-                    metadata = store.getMetadata();
+                    metadata = store.getMetadata(snapshotIndexCommit);
                 } catch (IOException e) {
                     throw new IndexShardSnapshotFailedException(shardId, "Failed to get store file metadata", e);
                 }

src/main/java/org/elasticsearch/index/store/Store.java

@@ -112,16 +112,16 @@ public Directory directory() {
      * @throws IOException if the index is corrupted or the segments file is not present
      */
     public SegmentInfos readLastCommittedSegmentsInfo() throws IOException {
-        return readLastCommittedSegmentsInfo(directory());
+        return readSegmentsInfo(null, directory());
     }
 
     /**
-     * Returns the last committed segments info for the given directory
+     * Returns the segments info for the given commit or for the latest commit if the given commit is <code>null</code>
      * @throws IOException if the index is corrupted or the segments file is not present
      */
-    private static SegmentInfos readLastCommittedSegmentsInfo(Directory directory) throws IOException {
+    private static SegmentInfos readSegmentsInfo(IndexCommit commit, Directory directory) throws IOException {
         try {
-            return Lucene.readSegmentInfos(directory);
+            return commit == null ? Lucene.readSegmentInfos(directory) : Lucene.readSegmentInfos(commit, directory);
         } catch (EOFException eof) {
             // TODO this should be caught by lucene - EOF is almost certainly an index corruption
             throw new CorruptIndexException("Read past EOF while reading segment infos", eof);
@@ -135,13 +135,21 @@ private final void ensureOpen() {
     }
 
     /**
-     * Returns a new MetadataSnapshot.
+     * Returns a new MetadataSnapshot for the latest commit in this store.
      */
     public MetadataSnapshot getMetadata() throws IOException {
+        return getMetadata(null);
+    }
+
+    /**
+     * Returns a new MetadataSnapshot for the given commit. If the given commit is <code>null</code>
+     * the latest commit point is used.
+     */
+    public MetadataSnapshot getMetadata(IndexCommit commit) throws IOException {
         ensureOpen();
         failIfCorrupted();
         try {
-            return new MetadataSnapshot(distributorDirectory, logger);
+            return new MetadataSnapshot(commit, distributorDirectory, logger);
         } catch (CorruptIndexException ex) {
             markStoreCorrupted(ex);
             throw ex;
@@ -268,7 +276,7 @@ public static MetadataSnapshot readMetadataSnapshot(File[] indexLocations, ESLog
             }
             DistributorDirectory dir = new DistributorDirectory(dirs);
             failIfCorrupted(dir, new ShardId("", 1));
-            return new MetadataSnapshot(dir, logger);
+            return new MetadataSnapshot(null, dir, logger);
         } finally {
             IOUtils.close(dirs);
         }
@@ -429,17 +437,17 @@ public String toString() {
     public final static class MetadataSnapshot implements Iterable<StoreFileMetaData> {
         private final ImmutableMap<String, StoreFileMetaData> metadata;
 
-        MetadataSnapshot(Directory directory, ESLogger logger) throws IOException {
-            metadata = buildMetadata(directory, logger);
+        MetadataSnapshot(IndexCommit commit, Directory directory, ESLogger logger) throws IOException {
+            metadata = buildMetadata(commit, directory, logger);
         }
 
-        ImmutableMap<String, StoreFileMetaData> buildMetadata(Directory directory, ESLogger logger) throws IOException {
+        ImmutableMap<String, StoreFileMetaData> buildMetadata(IndexCommit commit, Directory directory, ESLogger logger) throws IOException {
             ImmutableMap.Builder<String, StoreFileMetaData> builder = ImmutableMap.builder();
             Map<String, String> checksumMap = readLegacyChecksums(directory);
             try {
                 final SegmentInfos segmentCommitInfos;
                 try {
-                    segmentCommitInfos = Store.readLastCommittedSegmentsInfo(directory);
+                    segmentCommitInfos = Store.readSegmentsInfo(commit, directory);
                 } catch (FileNotFoundException | NoSuchFileException ex) {
                     // no segments file -- can't read metadata
                     logger.trace("Can't read segment infos", ex);

src/main/java/org/elasticsearch/indices/recovery/RecoverySource.java

@@ -142,7 +142,7 @@ public void phase1(final SnapshotIndexCommit snapshot) throws ElasticsearchExcep
                 try {
                     StopWatch stopWatch = new StopWatch().start();
                     final Store.MetadataSnapshot metadata;
-                    metadata = store.getMetadata();
+                    metadata = store.getMetadata(snapshot);
                     for (String name : snapshot.getFiles()) {
                         final StoreFileMetaData md = metadata.get(name);
                         if (md == null) {